Cognitive_Agent/tests/e2e/test_self_calibration_api.py

from __future__ import annotations

import json
import tempfile
import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path

import pytest

from cognitive_agent.application.calibration.service import CalibrationService
from cognitive_agent.auth import AuthConfig, AuthRuntime
from cognitive_agent.server import REPOSITORY_ROOT, SeatDemoHandler


def _auth_disabled() -> AuthRuntime:
    return AuthRuntime(
        AuthConfig(
            enabled=False,
            username="",
            password_hash="",
            session_secret=b"",
            session_ttl_seconds=3600,
            login_attempts=5,
            login_window_seconds=300,
            requests_per_minute=120,
            trust_proxy_headers=False,
        )
    )


class TestSelfCalibrationApi:
    @classmethod
    def setup_class(cls) -> None:
        cls.temporary_directory = tempfile.TemporaryDirectory()
        cls.calibration = CalibrationService(
            Path(cls.temporary_directory.name),
            repository_root=REPOSITORY_ROOT,
        )
        cls.calibration.start()
        cls.original_calibration = SeatDemoHandler.calibration
        cls.original_auth = SeatDemoHandler.auth
        SeatDemoHandler.calibration = cls.calibration
        SeatDemoHandler.auth = _auth_disabled()
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), SeatDemoHandler)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()
        cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def teardown_class(cls) -> None:
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=2)
        cls.calibration.close()
        SeatDemoHandler.calibration = cls.original_calibration
        SeatDemoHandler.auth = cls.original_auth
        cls.temporary_directory.cleanup()

    def request(
        self,
        path: str,
        *,
        method: str = "GET",
        payload: dict[str, object] | None = None,
        role: str | None = "admin",
    ) -> urllib.request.Request:
        headers = {"Accept": "application/json"}
        data = None
        if role is not None:
            headers["X-Cognitive-Agent-Role"] = role
        if payload is not None:
            headers["Content-Type"] = "application/json"
            data = json.dumps(payload).encode("utf-8")
        return urllib.request.Request(
            f"{self.base_url}{path}",
            method=method,
            headers=headers,
            data=data,
        )

    def test_admin_gate_protects_calibration_api(self) -> None:
        with pytest.raises(urllib.error.HTTPError) as caught:
            urllib.request.urlopen(
                self.request("/api/calibration/state", role="user"),
                timeout=3,
            )
        assert caught.value.code == 403
        body = json.loads(caught.value.read().decode("utf-8"))
        assert body["error"] == "admin_required"

    def test_simulation_connect_move_and_stop(self) -> None:
        with urllib.request.urlopen(
            self.request(
                "/api/calibration/can/connect",
                method="POST",
                payload={"can": {"mode": "simulation"}},
            ),
            timeout=3,
        ) as response:
            connected = json.loads(response.read().decode("utf-8"))["state"]
        assert connected["transport"]["state"] == "online"
        assert [axis["range"] for axis in connected["axes"]] == [480, 647, 917, 373]

        with urllib.request.urlopen(
            self.request(
                "/api/calibration/move",
                method="POST",
                payload={"axis": 2, "permille": 650},
            ),
            timeout=3,
        ) as response:
            moving = json.loads(response.read().decode("utf-8"))["state"]
        assert moving["controller"]["armed"]

        with urllib.request.urlopen(
            self.request("/api/calibration/stop", method="POST", payload={}),
            timeout=3,
        ) as response:
            stopped = json.loads(response.read().decode("utf-8"))["state"]
        assert not stopped["controller"]["armed"]

        with urllib.request.urlopen(
            self.request("/api/calibration/session/reset", method="POST", payload={}),
            timeout=3,
        ) as response:
            reset = json.loads(response.read().decode("utf-8"))["state"]
        assert reset["transport"]["state"] == "disconnected"
        assert [axis["permille"] for axis in reset["axes"]] == [0, 0, 0, 0]

    def test_frontend_and_original_model_are_removed(self) -> None:
        with urllib.request.urlopen(f"{self.base_url}/", timeout=3) as response:
            html = response.read().decode("utf-8")
        assert "self-calibration" not in html
        assert "vendor/mediamtx" not in html

        for path in (
            "/self-calibration.js",
            "/self-calibration.css",
            "/vendor/mediamtx/reader.js",
            "/assets/models/n7_seat.glb",
        ):
            with pytest.raises(urllib.error.HTTPError) as missing_asset:
                urllib.request.urlopen(f"{self.base_url}{path}", timeout=3)
            assert missing_asset.value.code == 404