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 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,
    api_panorama,
    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"]