dsh/dual_camera_system/tests/test_web_routes.py

"""Tests for web.routes helpers and endpoint guards."""
import sys
import os
import threading
import time

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

from types import SimpleNamespace
from unittest.mock import MagicMock

import numpy as np
import pytest
from fastapi import HTTPException
from fastapi.responses import JSONResponse
from pydantic import ValidationError

import web.routes as routes
import web.state as _web_state_module
from web.routes import (
    _get_state,
    _resolve_panorama_path,
    AddPointPayload,
    PreviewPayload,
    api_panorama,
    api_preview,
    api_start_scan,
    api_live,
)


@pytest.fixture(autouse=True)
def reset_web_state(monkeypatch):
    """Isolate the module-level web_state between tests."""
    monkeypatch.setattr(_web_state_module, "web_state", None)


def test_get_state_raises_503_when_not_initialized():
    _web_state_module.web_state = None
    with pytest.raises(HTTPException) as exc_info:
        _get_state()
    assert exc_info.value.status_code == 503
    assert "not initialized" in exc_info.value.detail.lower()


def test_resolve_panorama_path_returns_valid_path(tmp_path, monkeypatch):
    base = tmp_path / "panorama_data"
    (base / "panorama").mkdir(parents=True)
    img = base / "panorama" / "scan.jpg"
    img.write_bytes(b"")
    monkeypatch.setattr(routes, "PANORAMA_BASE", base.resolve())

    resolved = _resolve_panorama_path("panorama/scan.jpg")
    assert resolved == img.resolve()


def test_resolve_panorama_path_rejects_escape(tmp_path, monkeypatch):
    base = tmp_path / "panorama_data"
    base.mkdir()
    monkeypatch.setattr(routes, "PANORAMA_BASE", base.resolve())

    with pytest.raises(HTTPException) as exc_info:
        _resolve_panorama_path("../evil.jpg")
    assert exc_info.value.status_code == 400
    assert "Invalid panorama path" in exc_info.value.detail


def test_api_panorama_rejects_traversal_from_store(tmp_path, monkeypatch):
    base = tmp_path / "data"
    base.mkdir()
    monkeypatch.setattr(routes, "PANORAMA_BASE", base.resolve())

    scan_store = SimpleNamespace(
        get_group=lambda _gid: {"panorama": {"equirectangular": "../secret.txt"}}
    )
    _web_state_module.web_state = SimpleNamespace(
        scan_store=scan_store,
        group_state=None,
        scanners={},
        schedulers={},
        stream_manager=None,
    )

    with pytest.raises(HTTPException) as exc_info:
        api_panorama("g1")
    assert exc_info.value.status_code == 400


def test_start_scan_returns_202_and_launches_thread():
    started = threading.Event()

    scanner = MagicMock()

    def run_with_event(*args, **kwargs):
        started.set()
        return {
            "samples": [],
            "panorama_path": None,
            "config": {},
        }

    scanner.run.side_effect = run_with_event

    group_data = {"polling_state": "idle"}

    class MockGroupState:
        def get(self, _gid):
            return dict(group_data)

        def update(self, _gid, key, value):
            group_data[key] = value

        def compare_and_update(self, _gid, key, expected, new_value):
            if group_data.get(key) == expected:
                group_data[key] = new_value
                return True
            return False

    state = SimpleNamespace(
        group_state=MockGroupState(),
        scan_store=SimpleNamespace(
            set_samples=lambda *_args, **_kwargs: None,
            set_panorama=lambda *_args, **_kwargs: None,
            set_scan_config=lambda *_args, **_kwargs: None,
        ),
        scanners={"g1": scanner},
        schedulers={},
        stream_manager=None,
    )
    _web_state_module.web_state = state

    resp = api_start_scan("g1")
    assert isinstance(resp, JSONResponse)
    assert resp.status_code == 202
    assert resp.body is not None
    assert b"Scan started" in resp.body

    deadline = time.time() + 1.0
    while time.time() < deadline and not started.is_set():
        time.sleep(0.01)
    started.wait(timeout=0)
    scanner.run.assert_called_once()


def test_start_scan_rejects_concurrent_scan():
    group_data = {"polling_state": "scanning"}

    class MockGroupState:
        def get(self, _gid):
            return dict(group_data)

        def compare_and_update(self, _gid, key, expected, new_value):
            if group_data.get(key) == expected:
                group_data[key] = new_value
                return True
            return False

    state = SimpleNamespace(
        group_state=MockGroupState(),
        scanners={"g1": None},
        schedulers={},
        stream_manager=None,
        scan_store=None,
    )
    _web_state_module.web_state = state

    with pytest.raises(HTTPException) as exc_info:
        api_start_scan("g1")
    assert exc_info.value.status_code == 409
    assert "already in progress" in exc_info.value.detail.lower()


def test_start_scan_returns_404_when_group_not_found():
    class MockGroupState:
        def get(self, _gid):
            return {}

        def compare_and_update(self, _gid, _key, _expected, _new_value):
            return False

    state = SimpleNamespace(
        group_state=MockGroupState(),
        scanners={},
        schedulers={},
        stream_manager=None,
        scan_store=None,
    )
    _web_state_module.web_state = state

    with pytest.raises(HTTPException) as exc_info:
        api_start_scan("missing")
    assert exc_info.value.status_code == 404
    assert "Group not found" in exc_info.value.detail


def test_api_live_rejects_invalid_camera_type():
    _web_state_module.web_state = SimpleNamespace(stream_manager=MagicMock())

    with pytest.raises(HTTPException) as exc_info:
        api_live("front_door", "g1")
    assert exc_info.value.status_code == 400
    assert "Invalid camera type" in exc_info.value.detail


@pytest.mark.parametrize(
    "payload, should_raise",
    [
        ({"pan": 10, "tilt": 5}, False),
        ({"pan": 400, "tilt": 0}, True),
        ({"pan": 0, "tilt": 100}, True),
        ({"pan": 0, "tilt": 0, "zoom": 0}, True),
        ({"pan": 0, "tilt": 0, "dwell_time": -1}, True),
    ],
)
def test_add_point_payload_validation(payload, should_raise):
    if should_raise:
        with pytest.raises(ValidationError):
            AddPointPayload(**payload)
    else:
        model = AddPointPayload(**payload)
        assert model.pan == payload["pan"]
        assert model.tilt == payload["tilt"]


def test_api_preview_uses_fixed_settle_and_drain(monkeypatch):
    """preview 不依赖过期的 ptz_position 计算等待/排空，防止拍到旧帧。"""
    sleeps = []

    def fake_sleep(duration):
        sleeps.append(duration)

    monkeypatch.setattr(time, "sleep", fake_sleep)

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

    ptz = MagicMock()
    ptz.goto_exact_position.return_value = True

    # 故意让上一次目标与本次完全相同；旧实现会因此缩短等待时间
    group_data = {"ptz_position": {"pan": 30.0, "tilt": 0.0, "zoom": 1}}

    class MockGroupState:
        def get(self, _gid):
            return dict(group_data)

        def update(self, _gid, key, value):
            group_data[key] = value

    state = SimpleNamespace(
        group_state=MockGroupState(),
        ptz_cameras={"g1": ptz},
        stream_manager=stream_manager,
    )
    _web_state_module.web_state = state

    monkeypatch.setattr(routes.cv2, "imwrite", lambda _path, _img: True)

    payload = PreviewPayload(pan=30.0, tilt=0.0, zoom=1)
    resp = api_preview("g1", payload)

    assert resp["position"] == {"pan": 30.0, "tilt": 0.0, "zoom": 1}
    ptz.goto_exact_position.assert_called_once_with(30.0, 0.0, 1)

    # 应等待完整的稳定时间，并执行固定次数的排空
    assert sleeps[0] == pytest.approx(3.0, abs=0.01)
    assert sum(sleeps[1:]) == pytest.approx(20 * 0.15, abs=0.01)
    assert stream.get_frame.call_count >= 20