Spaces:
Paused
Paused
| """API contract tests for the Brain-5D dashboard control endpoints. | |
| Tests every runtime command through the full HTTP dispatch path: | |
| - Canonical contract: POST /api/control { "command": "...", ... } | |
| - Legacy contract: POST /api/control { "action": "...", ... } | |
| - Validation: missing fields, invalid types, unsupported commands | |
| - Error responses: structured JSON errors with appropriate HTTP status codes | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from http.client import HTTPConnection | |
| from threading import Thread | |
| from typing import Any, cast | |
| import pytest | |
| from src.dashboard.models import JSONValue | |
| from src.dashboard.operator_bridge import OperatorBridge | |
| from src.dashboard.server import DashboardServer | |
| from src.dashboard.state import DashboardStateStore | |
| from src.dashboard.structural_api import JSONMapping, StructuralCommandResult | |
| # ============================================================================ | |
| # Stubs | |
| # ============================================================================ | |
| class _StubController: | |
| """Minimal controller stub for testing the HTTP dispatch layer.""" | |
| def __init__(self) -> None: | |
| self._tick = 0 | |
| def telemetry(self) -> object: | |
| return type( | |
| "TelemetryStub", | |
| (), | |
| { | |
| "controller_state": "idle", | |
| "tick": self._tick, | |
| "neurons": 100, | |
| "synapses": 500, | |
| "queue_depth": 0, | |
| "spikes_this_batch": 0, | |
| "ticks_per_second": 0.0, | |
| "batch_duration_ms": 0.0, | |
| "requested_ticks": 0, | |
| "completed_ticks": 0, | |
| "last_error": None, | |
| }, | |
| )() | |
| def snapshot(self) -> object: | |
| tick = self._tick | |
| return type( | |
| "SnapshotStub", | |
| (), | |
| { | |
| "to_json": lambda _self=None, tick=tick: { # type: ignore[assignment] | |
| "mode": "idle", | |
| "tick": tick, | |
| "queued_ticks": 0, | |
| "loop_size": 100, | |
| "delay_ms": 0.0, | |
| "last_batch_ticks": 0, | |
| "last_batch_ms": 0.0, | |
| "total_runtime_ms": 0.0, | |
| "fault": None, | |
| "can_snapshot": False, | |
| } | |
| }, | |
| )() | |
| def step(self, ticks: int = 1) -> object: | |
| self._tick += ticks | |
| return self.snapshot() | |
| def run_ticks(self, ticks: int) -> object: | |
| self._tick += ticks | |
| return self.snapshot() | |
| def start(self) -> object: | |
| return self.snapshot() | |
| def run(self, *, loop_size: int | None = None) -> object: | |
| return self.snapshot() | |
| def pause(self) -> object: | |
| return self.snapshot() | |
| def stop(self) -> object: | |
| return self.snapshot() | |
| def configure(self, **kwargs: Any) -> object: | |
| return self.snapshot() | |
| def request_snapshot(self) -> object: | |
| return self.snapshot() | |
| class _Bridge: | |
| """Minimal bridge stub for server startup.""" | |
| def __init__(self) -> None: | |
| self.controller = _StubController() | |
| def structural_status(self) -> JSONMapping: | |
| return {"configured": True} | |
| def structural_proposals(self) -> list[JSONMapping]: | |
| return [] | |
| def structural_history(self, _limit: int) -> list[JSONMapping]: | |
| return [] | |
| def structural_heatmap(self, kind: str) -> JSONMapping: | |
| return {"kind": kind, "values": []} | |
| def structural_config(self) -> JSONMapping: | |
| return {"configured": True} | |
| def approve_structural(self, _proposal_id: str) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, "approved") | |
| def reject_structural(self, _proposal_id: str) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, "rejected") | |
| def undo_structural(self) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, "undone") | |
| def set_auto_approval(self, enabled: bool) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, str(enabled)) | |
| def run_ticks(self, count: int) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, str(count)) | |
| def single_step(self) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, "one") | |
| def request_snapshot(self) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, "snapshot") | |
| def command(self, command: str, **_kwargs: Any) -> dict[str, JSONValue]: | |
| return {"ok": True, "status": command} | |
| def update_structural_config(self, **_body: Any) -> StructuralCommandResult: | |
| return StructuralCommandResult(True, "updated") | |
| # ============================================================================ | |
| # Test helpers | |
| # ============================================================================ | |
| def server_and_client() -> tuple[DashboardServer, Thread, str, int]: | |
| """Start a dashboard server with a stub bridge for testing.""" | |
| bridge = _Bridge() | |
| server = DashboardServer( | |
| ("127.0.0.1", 0), | |
| DashboardStateStore(), | |
| None, | |
| cast(OperatorBridge | None, bridge), | |
| ) | |
| thread = Thread(target=server.serve_forever, daemon=True) | |
| thread.start() | |
| host, port = server.server_address[:2] | |
| if not isinstance(host, str): | |
| raise AssertionError("server address is not a string") | |
| return server, thread, host, port | |
| def _post( | |
| host: str, port: int, path: str, body: object | |
| ) -> tuple[int, dict[str, JSONValue] | str]: | |
| """Send a POST request and return (status, parsed_body).""" | |
| conn = HTTPConnection(host, port) | |
| try: | |
| conn.request( | |
| "POST", | |
| path, | |
| json.dumps(body), | |
| {"Content-Type": "application/json"}, | |
| ) | |
| resp = conn.getresponse() | |
| data = resp.read() | |
| ct = resp.getheader("Content-Type", "") | |
| if "application/json" in ct: | |
| return resp.status, json.loads(data) | |
| return resp.status, data.decode("utf-8", errors="replace") | |
| finally: | |
| conn.close() | |
| def _get(host: str, port: int, path: str) -> tuple[int, dict[str, JSONValue] | str]: | |
| """Send a GET request and return (status, parsed_body).""" | |
| conn = HTTPConnection(host, port) | |
| try: | |
| conn.request("GET", path) | |
| resp = conn.getresponse() | |
| data = resp.read() | |
| ct = resp.getheader("Content-Type", "") | |
| if "application/json" in ct: | |
| return resp.status, json.loads(data) | |
| return resp.status, data.decode("utf-8", errors="replace") | |
| finally: | |
| conn.close() | |
| # ============================================================================ | |
| # Canonical Contract Tests | |
| # ============================================================================ | |
| class TestCanonicalContract: | |
| """Test the canonical { "command": "...", ... } contract.""" | |
| def test_run_ticks_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post( | |
| host, port, "/api/control", {"command": "run_ticks", "ticks": 10} | |
| ) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_step_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post( | |
| host, port, "/api/control", {"command": "step", "ticks": 1} | |
| ) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_pause_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "pause"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_stop_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "stop"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_snapshot_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "snapshot"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_start_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "start"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_resume_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "resume"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_configure_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post( | |
| host, | |
| port, | |
| "/api/control", | |
| {"command": "configure", "loop_size": 50, "delay_ms": 1.0}, | |
| ) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_single_step_command( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "single_step"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| # ============================================================================ | |
| # Legacy Contract Tests | |
| # ============================================================================ | |
| class TestLegacyContract: | |
| """Test the legacy { "action": "...", ... } contract still works.""" | |
| def test_legacy_step_action( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"action": "step", "ticks": 1}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_legacy_run_ticks_action( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post( | |
| host, port, "/api/control", {"action": "run_ticks", "ticks": 10} | |
| ) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_legacy_pause_action( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"action": "pause"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| def test_legacy_stop_action( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"action": "stop"}) | |
| assert status == 200 | |
| assert isinstance(body, dict) | |
| assert body.get("ok") is True | |
| # ============================================================================ | |
| # Validation Tests | |
| # ============================================================================ | |
| class TestValidation: | |
| """Test input validation and error responses.""" | |
| def test_missing_command_returns_400( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {}) | |
| assert status == 400 | |
| assert isinstance(body, dict) | |
| assert "error" in body | |
| def test_unknown_command_returns_400( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", {"command": "warp-core"}) | |
| assert status == 400 | |
| assert isinstance(body, dict) | |
| assert "error" in body | |
| def test_non_dict_body_returns_400( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/control", ["not", "a", "dict"]) | |
| assert status == 400 | |
| assert isinstance(body, dict) | |
| assert "error" in body | |
| def test_invalid_ticks_type_returns_400( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post( | |
| host, | |
| port, | |
| "/api/control", | |
| {"command": "run_ticks", "ticks": "not-a-number"}, | |
| ) | |
| assert status == 400 | |
| assert isinstance(body, dict) | |
| assert "error" in body | |
| def test_unsupported_content_type_returns_415( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| conn = HTTPConnection(host, port) | |
| try: | |
| conn.request( | |
| "POST", | |
| "/api/control", | |
| '{"command": "pause"}', | |
| {"Content-Type": "text/plain"}, | |
| ) | |
| resp = conn.getresponse() | |
| _data = resp.read() | |
| assert resp.status in (400, 415) | |
| finally: | |
| conn.close() | |
| # ============================================================================ | |
| # GET /api/control (status) Tests | |
| # ============================================================================ | |
| class TestControlStatus: | |
| """Test the GET /api/control status endpoint.""" | |
| def test_get_control_status_returns_200( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| conn = HTTPConnection(host, port) | |
| try: | |
| conn.request("GET", "/api/control") | |
| resp = conn.getresponse() | |
| data = resp.read() | |
| assert resp.status == 200 | |
| ct = resp.getheader("Content-Type", "") | |
| assert "application/json" in ct | |
| body = json.loads(data) | |
| assert isinstance(body, dict) | |
| assert "state" in body | |
| finally: | |
| conn.close() | |
| class TestUnknownApiRoutes: | |
| """Unknown API routes remain JSON errors instead of SPA fallbacks.""" | |
| def test_unknown_get_route_returns_json_404( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _get(host, port, "/api/does-not-exist") | |
| assert status == 404 | |
| assert isinstance(body, dict) | |
| assert body["error"] == "Unknown API endpoint: /api/does-not-exist" | |
| def test_unknown_post_route_returns_json_404( | |
| self, server_and_client: tuple[DashboardServer, Thread, str, int] | |
| ) -> None: | |
| _, _, host, port = server_and_client | |
| status, body = _post(host, port, "/api/does-not-exist", {}) | |
| assert status == 404 | |
| assert isinstance(body, dict) | |
| assert body["error"] == "Unknown API endpoint: /api/does-not-exist" | |