vent_ai_platform/tests/test_video_platform_api.py

from __future__ import annotations

from pathlib import Path

from fastapi.testclient import TestClient

from app.core.config import settings
from app.db.init_db import init_db
from app.db.seed import seed_demo_data
from app.db.session import SessionLocal
from app.main import app


def _prepare() -> None:
    init_db()
    with SessionLocal() as db:
        seed_demo_data(db)


def _mine_id(client: TestClient) -> str:
    response = client.get("/api/v1/mines")
    assert response.status_code == 200
    return response.json()[0]["id"]


def test_video_platform_overview_and_modules() -> None:
    _prepare()
    with TestClient(app) as client:
        mine_id = _mine_id(client)

        overview = client.get(
            "/api/v1/video-platform/overview", params={"mine_id": mine_id}
        )
        assert overview.status_code == 200
        payload = overview.json()
        assert payload["channel_total"] == 18
        assert payload["channel_online"] == 15
        assert payload["channel_offline"] == 1
        assert payload["channel_fault"] == 2
        assert payload["snapshot_count"] > 0
        assert len(payload["summary"]) == 4

        modules = client.get(
            "/api/v1/video-platform/modules", params={"mine_id": mine_id}
        )
        assert modules.status_code == 200
        configs = {item["key"]: item for item in modules.json()}
        assert set(configs) == {"door", "local_fan", "main_fan"}
        assert len(configs["door"]["record_tabs"]) == 3
        assert configs["door"]["record_tabs"][0]["record_type"] == "vehicle_passage"
        assert "AI防夹人预警" in configs["door"]["type_filter_options"]
        assert "全部设备" in configs["local_fan"]["scope_filter_options"]
        assert len(configs["local_fan"]["scope_filter_options"]) == 6


def test_video_platform_channels_keep_disk_paths() -> None:
    _prepare()
    with TestClient(app) as client:
        mine_id = _mine_id(client)

        response = client.get(
            "/api/v1/video-platform/channels",
            params={"mine_id": mine_id, "module": "door"},
        )
        assert response.status_code == 200
        channels = response.json()
        assert len(channels) == 8
        doo_channel = channels[0]
        assert Path(doo_channel["file_path"]).is_absolute()
        assert doo_channel["file_path"].startswith(str(settings.video_media_path))
        assert doo_channel["file_name"].endswith(".mp4")
        assert doo_channel["playback_url"].endswith("/stream")
        assert doo_channel["snapshot_url"].endswith("/snapshot")
        assert doo_channel["has_video"] is True

        search = client.get(
            "/api/v1/video-platform/channels",
            params={"mine_id": mine_id, "keyword": "变电所"},
        )
        assert search.status_code == 200
        assert [item["code"] for item in search.json()] == ["VC-DOOR-04"]


def test_video_platform_status_summary_and_records() -> None:
    _prepare()
    with TestClient(app) as client:
        mine_id = _mine_id(client)

        summary = client.get(
            "/api/v1/video-platform/modules/door/summary",
            params={"mine_id": mine_id},
        )
        assert summary.status_code == 200
        metrics = {item["key"]: item["value"] for item in summary.json()["metrics"]}
        assert metrics["normal_door"] == 6
        assert metrics["car_today"] == 12
        assert metrics["person_today"] == 11
        assert metrics["open_warning"] > 0

        fan_summary = client.get(
            "/api/v1/video-platform/modules/local_fan/summary",
            params={"mine_id": mine_id},
        ).json()
        fan_metrics = {item["key"]: item["value"] for item in fan_summary["metrics"]}
        assert fan_metrics["normal_fan"] == 4
        assert fan_metrics["fault_fan"] == 1
        assert fan_metrics["onsite_person"] == 2

        main_summary = client.get(
            "/api/v1/video-platform/modules/main_fan/summary",
            params={"mine_id": mine_id},
        ).json()
        main_metrics = {item["key"]: item["value"] for item in main_summary["metrics"]}
        assert main_metrics["running_fan"] == 1
        assert main_metrics["stopped_fan"] == 1
        assert main_metrics["person_detect"] == 6

        records = client.get(
            "/api/v1/video-platform/records",
            params={
                "mine_id": mine_id,
                "module": "door",
                "record_type": "vehicle_passage",
                "page_size": 5,
            },
        )
        assert records.status_code == 200
        record_page = records.json()
        assert record_page["total"] == 12
        assert len(record_page["items"]) == 5
        assert record_page["items"][0]["plate_no"].startswith("蒙K")
        assert record_page["items"][0]["snapshot_image_url"].endswith("/image")

        tracks = client.get(
            "/api/v1/video-platform/records",
            params={
                "mine_id": mine_id,
                "module": "local_fan",
                "record_type": "person_track",
            },
        ).json()
        assert tracks["total"] == 8
        area_rows = client.get(
            "/api/v1/video-platform/records",
            params={
                "mine_id": mine_id,
                "module": "local_fan",
                "record_type": "area_stay",
            },
        ).json()
        assert area_rows["total"] == 5
        assert "avg_stay" in area_rows["items"][0]["properties"]

        warn_rows = client.get(
            "/api/v1/video-platform/records",
            params={
                "mine_id": mine_id,
                "module": "main_fan",
                "record_type": "device_warn",
            },
        ).json()
        assert warn_rows["total"] == 4
        assert warn_rows["items"][0]["warn_type"]


def test_video_platform_snapshots_and_events() -> None:
    _prepare()
    with TestClient(app) as client:
        mine_id = _mine_id(client)

        snapshots = client.get(
            "/api/v1/video-platform/snapshots",
            params={"mine_id": mine_id, "module": "door", "page_size": 100},
        )
        assert snapshots.status_code == 200
        page = snapshots.json()
        assert page["total"] == 40
        assert page["items"][0]["image_url"].endswith("/image")

        filtered = client.get(
            "/api/v1/video-platform/snapshots",
            params={
                "mine_id": mine_id,
                "module": "door",
                "event_type": "风门未关严",
                "page_size": 100,
            },
        ).json()
        assert filtered["total"] == 9
        assert all(item["gap_mm"] for item in filtered["items"])

        keyword = client.get(
            "/api/v1/video-platform/snapshots",
            params={"mine_id": mine_id, "keyword": "蒙K3571"},
        ).json()
        assert keyword["total"] == 1

        events = client.get(
            "/api/v1/video-platform/events",
            params={"mine_id": mine_id, "module": "door"},
        )
        assert events.status_code == 200
        event_items = events.json()
        assert len(event_items) == 10
        assert {item["level"] for item in event_items} >= {"danger", "warning", "info"}
        linked = [item for item in event_items if item["snapshot_image_url"]]
        assert linked

        open_events = client.get(
            "/api/v1/video-platform/events",
            params={"mine_id": mine_id, "module": "main_fan", "status": "pending"},
        ).json()
        assert open_events
        event_id = open_events[0]["id"]
        updated = client.patch(
            f"/api/v1/video-platform/events/{event_id}",
            json={"status": "handled", "handled_by": "通防科值班员", "handle_remark": "现场已复核"},
        )
        assert updated.status_code == 200
        updated_payload = updated.json()
        assert updated_payload["status"] == "handled"
        assert updated_payload["handled_at"] is not None
        assert updated_payload["handle_remark"] == "现场已复核"


def test_video_platform_channel_media_stream_and_crud(tmp_path: Path) -> None:
    _prepare()
    video_file = tmp_path / "demo.mp4"
    video_file.write_bytes(b"0123456789")
    image_file = tmp_path / "demo.jpg"
    image_file.write_bytes(b"fake-image-bytes")

    with TestClient(app) as client:
        mine_id = _mine_id(client)

        created = client.post(
            "/api/v1/video-platform/channels",
            json={
                "mine_id": mine_id,
                "code": "VC-TEST-01",
                "name": "测试通道-煤仓口",
                "module": "door",
                "area_group": "风门区域",
                "location": "测试地点",
                "device_type": "door",
                "status": "online",
                "file_path": str(video_file),
                "properties": {"door_side": "前门"},
            },
        )
        assert created.status_code == 201
        channel = created.json()
        channel_id = channel["id"]
        assert channel["file_path"] == str(video_file)
        assert channel["file_name"] == "demo.mp4"
        assert channel["file_size"] == 10
        assert channel["media_available"] is True

        duplicated = client.post(
            "/api/v1/video-platform/channels",
            json={
                "mine_id": mine_id,
                "code": "VC-TEST-01",
                "name": "重复编码",
                "module": "door",
            },
        )
        assert duplicated.status_code == 409

        full = client.get(f"/api/v1/video-platform/channels/{channel_id}/stream")
        assert full.status_code == 200
        assert full.headers["accept-ranges"] == "bytes"
        assert full.content == b"0123456789"

        partial = client.get(
            f"/api/v1/video-platform/channels/{channel_id}/stream",
            headers={"Range": "bytes=2-5"},
        )
        assert partial.status_code == 206
        assert partial.headers["content-range"] == "bytes 2-5/10"
        assert partial.content == b"2345"

        patched = client.patch(
            f"/api/v1/video-platform/channels/{channel_id}",
            json={"name": "测试通道-煤仓口（改）", "status": "offline"},
        )
        assert patched.status_code == 200
        assert patched.json()["name"] == "测试通道-煤仓口（改）"
        assert patched.json()["status"] == "offline"

        snapshot = client.get(
            "/api/v1/video-platform/snapshots",
            params={"mine_id": mine_id, "page_size": 200},
        ).json()["items"][0]
        image = client.get(f"/api/v1/video-platform/snapshots/{snapshot['id']}/image")
        assert image.status_code in (200, 404)

        missing = client.post(
            "/api/v1/video-platform/channels",
            json={
                "mine_id": mine_id,
                "code": "VC-TEST-02",
                "name": "测试通道-缺失文件",
                "module": "door",
                "file_path": str(tmp_path / "not-found.mp4"),
            },
        ).json()
        missing_stream = client.get(
            f"/api/v1/video-platform/channels/{missing['id']}/stream"
        )
        assert missing_stream.status_code == 404
        assert missing_stream.json()["code"] == "video_media_missing"

        deleted = client.delete(f"/api/v1/video-platform/channels/{channel_id}")
        assert deleted.status_code == 204
        assert (
            client.get(f"/api/v1/video-platform/channels/{channel_id}").status_code
            == 404
        )
        client.delete(f"/api/v1/video-platform/channels/{missing['id']}")

        assert image_file.exists()  # 图像文件保持磁盘原样，仅记录路径