dsh/dual_camera_system/tests/test_routes.py

import sys
import os
import threading
import time

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from unittest.mock import MagicMock

import numpy as np
import pytest
from fastapi.testclient import TestClient
from app import create_app
import web.state as _web_state_module


@pytest.fixture
def client():
    app = create_app(test_mode=True)
    return TestClient(app)


@pytest.fixture
def client_with_mocks():
    app = create_app(test_mode=True)
    client = TestClient(app)

    scanner = MagicMock()
    scanner.run.return_value = {
        "samples": [],
        "panorama_path": None,
        "config": {},
    }
    scheduler = MagicMock()

    ptz = MagicMock()
    ptz_stream = MagicMock()
    ptz_stream.get_frame.return_value = np.zeros((100, 100, 3), dtype=np.uint8)
    stream_manager = MagicMock()
    stream_manager.get.return_value = ptz_stream

    _web_state_module.web_state.scanners["group_1"] = scanner
    _web_state_module.web_state.schedulers["group_1"] = scheduler
    _web_state_module.web_state.ptz_cameras["group_1"] = ptz
    _web_state_module.web_state.stream_manager = stream_manager
    return client, scanner, scheduler, ptz, ptz_stream


def test_status_endpoint(client):
    response = client.get("/api/status")
    assert response.status_code == 200
    assert "groups" in response.json()


def test_add_and_list_points(client):
    response = client.post("/api/points/group_1", json={
        "pan": 30.0,
        "tilt": 0.0,
        "zoom": 1,
        "dwell_time": 3.0,
    })
    assert response.status_code == 200
    data = response.json()
    assert data["pan"] == 30.0

    response = client.get("/api/points/group_1")
    assert response.status_code == 200
    points = response.json()["points"]
    assert len(points) == 1


def test_add_point_validation(client):
    response = client.post("/api/points/group_1", json={
        "pan": 400.0,
        "tilt": 0.0,
        "zoom": 1,
        "dwell_time": 3.0,
    })
    assert response.status_code == 422


def test_delete_point(client):
    response = client.post("/api/points/group_1", json={
        "pan": 60.0,
        "tilt": 10.0,
        "zoom": 1,
        "dwell_time": 2.0,
    })
    point_id = response.json()["id"]
    response = client.delete(f"/api/points/group_1/{point_id}")
    assert response.status_code == 200
    response = client.get("/api/points/group_1")
    assert len(response.json()["points"]) == 0


def test_static_index(client):
    response = client.get("/")
    assert response.status_code == 200
    assert "text/html" in response.headers["content-type"]


def test_panorama_not_found(client):
    response = client.get("/api/panorama/group_1")
    assert response.status_code == 404


def test_scan_start_returns_202(client_with_mocks):
    client, scanner, _, _, _ = client_with_mocks
    started = threading.Event()

    def run_with_event(*args, **kwargs):
        started.set()
        return scanner.run.return_value

    scanner.run.side_effect = run_with_event

    response = client.post("/api/scan/group_1")
    assert response.status_code == 202
    assert response.json()["group_id"] == "group_1"

    deadline = time.time() + 2.0
    while time.time() < deadline and not started.is_set():
        time.sleep(0.01)
    assert started.is_set(), "Scanner run() was not invoked"


def test_scan_uses_full_vertical_range(client_with_mocks):
    """360° 扫描应覆盖 -90° 到 +90° 的完整上下范围。"""
    client, scanner, _, _, _ = client_with_mocks
    started = threading.Event()

    def run_with_event(*args, **kwargs):
        started.set()
        return scanner.run.return_value

    scanner.run.side_effect = run_with_event

    response = client.post("/api/scan/group_1")
    assert response.status_code == 202

    deadline = time.time() + 2.0
    while time.time() < deadline and not started.is_set():
        time.sleep(0.01)
    assert started.is_set(), "Scanner run() was not invoked"

    _, kwargs = scanner.run.call_args
    tilt_layers = kwargs.get("tilt_layers", ())
    assert min(tilt_layers) <= -90
    assert max(tilt_layers) >= 90


def test_poll_start_and_stop(client_with_mocks):
    client, _, scheduler, _, _ = client_with_mocks
    response = client.post("/api/poll/group_1/start")
    assert response.status_code == 200
    scheduler.start.assert_called_once()

    response = client.post("/api/poll/group_1/stop")
    assert response.status_code == 200
    scheduler.stop.assert_called_once()


def test_live_stream_not_found(client):
    response = client.get("/api/live/panorama/group_1")
    assert response.status_code == 404


def test_static_css_served(client):
    response = client.get("/static/style.css")
    assert response.status_code == 200
    assert "text/css" in response.headers["content-type"]


def test_static_js_served(client):
    response = client.get("/static/app.js")
    assert response.status_code == 200
    assert "javascript" in response.headers["content-type"]


def test_mutating_endpoints_require_api_key(client_with_mocks, monkeypatch):
    client, scanner, scheduler, _, _ = client_with_mocks
    monkeypatch.setattr("web.auth.DEVICE_CONFIG", {"api_key": "secret123"})

    # Scan
    response = client.post("/api/scan/group_1")
    assert response.status_code == 401

    # Poll start/stop
    response = client.post("/api/poll/group_1/start")
    assert response.status_code == 401
    response = client.post("/api/poll/group_1/stop")
    assert response.status_code == 401

    # Points add/delete
    response = client.post("/api/points/group_1", json={"pan": 30, "tilt": 0})
    assert response.status_code == 401
    response = client.delete("/api/points/group_1/1")
    assert response.status_code == 401

    scanner.run.assert_not_called()
    scheduler.start.assert_not_called()


def test_mutating_endpoints_accept_api_key(client_with_mocks, monkeypatch):
    client, scanner, scheduler, _, _ = client_with_mocks
    monkeypatch.setattr("web.auth.DEVICE_CONFIG", {"api_key": "secret123"})

    response = client.post(
        "/api/poll/group_1/start",
        headers={"X-API-Key": "secret123"},
    )
    assert response.status_code == 200
    scheduler.start.assert_called_once()


def test_readonly_endpoints_remain_open_with_api_key(client, monkeypatch):
    monkeypatch.setattr("web.auth.DEVICE_CONFIG", {"api_key": "secret123"})

    response = client.get("/api/status")
    assert response.status_code == 200

    response = client.get("/api/points/group_1")
    assert response.status_code == 200

    response = client.get("/api/scan/group_1/progress")
    assert response.status_code == 200


def test_preview_does_not_update_saved_point_image(client_with_mocks):
    """点球机预览只临时抓拍并显示，不应更新保存点的 preview_image。"""
    client, _, _, _, _ = client_with_mocks

    # 创建一个带有固定 preview_image 的保存点
    response = client.post("/api/points/group_1", json={
        "pan": 30.0,
        "tilt": 0.0,
        "zoom": 1,
        "dwell_time": 3.0,
        "preview_image": "data/previews/group_1/existing.jpg",
    })
    assert response.status_code == 200
    point_id = response.json()["id"]

    response = client.post("/api/preview/group_1", json={
        "pan": 30.0,
        "tilt": 0.0,
        "zoom": 1,
        "point_id": point_id,
    })
    assert response.status_code == 200
    assert response.json()["snapshot_url"] is not None

    response = client.get(f"/api/points/group_1")
    assert response.status_code == 200
    point = next(p for p in response.json()["points"] if p["id"] == point_id)
    # 保存点的 preview_image 应保持原值，不被预览抓拍覆盖
    assert point["preview_image"] == "data/previews/group_1/existing.jpg"