from __future__ import annotations import hashlib import json from collections.abc import Callable from pathlib import Path from typing import Any import pytest from fastapi import APIRouter, HTTPException from fastapi.routing import APIRoute from k1link.simulation import ( AuthorityProfile, ProviderPin, QualificationArtifact, QualificationRun, QualificationRunStore, ReproducibilityTier, RunKind, RunState, ) from k1link.simulation.stock_rover import STOCK_ROVER_PROVIDER_PROFILE from k1link.web.polygon_api import ( AckermannCommandRequest, StartStockRoverRequest, build_polygon_router, ) SHA_A = "a" * 64 SHA_B = "b" * 64 SHA_C = "c" * 64 def _run(run_id: str = "s1b-ui0-001") -> QualificationRun: return QualificationRun( run_id=run_id, episode_id=f"episode-{run_id}", kind=RunKind.SIMULATION_CLOSED_LOOP, state=RunState.ADMITTED, scenario_generation="px4-v1.17.0-stock-rover-ackermann", scenario_sha256=SHA_A, profile_generation="stock-rover-lifecycle-v1", profile_sha256=SHA_B, mission_core_commit="6cb1495a1234567890abcdef1234567890abcdef", providers=( ProviderPin( identifier="px4-autopilot", version="v1.17.0", revision="v1.17.0", ), ProviderPin( identifier="gazebo", version="harmonic", revision="8.9.0", ), ), host_profile_id="mission-gpu-s0", host_profile_sha256=SHA_C, seed=42, reproducibility_tier=ReproducibilityTier.R1, authority=AuthorityProfile( generation=1, command_ttl_max_ns=250_000_000, heartbeat_timeout_monotonic_ns=500_000_000, ), clock_domain="gazebo:/clock", created_at_utc="2026-07-24T15:35:00Z", ) def _accepted_run(root: Path) -> QualificationRun: store = QualificationRunStore(root) admitted = store.create(_run()) starting = store.transition( admitted.run_id, RunState.STARTING, expected_revision=0, observed_at_utc="2026-07-24T15:35:01Z", host_monotonic_ns=1, ) running = store.transition( admitted.run_id, RunState.RUNNING, expected_revision=starting.revision, observed_at_utc="2026-07-24T15:35:02Z", host_monotonic_ns=2, sim_time_ns=0, ) event = store.append_event( admitted.run_id, event_type="orchestrator.providers-started", observed_at_utc="2026-07-24T15:35:03Z", host_monotonic_ns=3, sim_time_ns=1_000_000, payload={"provider_ids": ["micro-xrce-dds-agent", "px4-gazebo-stock-rover"]}, expected_revision=running.revision, ) store.register_artifact( admitted.run_id, QualificationArtifact( artifact_id="provider-log-index", kind="log-index", relative_path="provider-runtime/processes.json", sha256=SHA_A, byte_length=512, source_of_record=True, ), ) stopping = store.transition( admitted.run_id, RunState.STOPPING, expected_revision=event.sequence, observed_at_utc="2026-07-24T15:35:04Z", host_monotonic_ns=4, sim_time_ns=2_000_000, ) return store.transition( admitted.run_id, RunState.COMPLETED, expected_revision=stopping.revision, observed_at_utc="2026-07-24T15:35:05Z", host_monotonic_ns=5, sim_time_ns=2_000_000, reason="operator-stop-clean", ) def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]: for route in router.routes: if isinstance(route, APIRoute) and route.path == path and method in route.methods: return route.endpoint raise AssertionError(f"{method} {path} route is missing") def test_polygon_api_is_read_only_and_fails_closed_when_unconfigured() -> None: router = build_polygon_router(root_provider=lambda: None) routes = [route for route in router.routes if isinstance(route, APIRoute)] archive_routes = [route for route in routes if "/worker" not in route.path] assert all(route.methods <= {"GET", "HEAD"} for route in archive_routes) with pytest.raises(HTTPException) as failure: _endpoint(router, "/api/v1/polygon/runs", "GET")(limit=20) assert failure.value.status_code == 503 assert "не настроен" in failure.value.detail def test_polygon_api_publishes_bounded_path_free_qualification_evidence( tmp_path: Path, ) -> None: root = tmp_path / "runs" accepted = _accepted_run(root) router = build_polygon_router(root_provider=lambda: root) catalog = _endpoint(router, "/api/v1/polygon/runs", "GET")(limit=20) detail = _endpoint(router, "/api/v1/polygon/runs/{run_id}", "GET")( run_id=accepted.run_id, event_limit=2, ) assert catalog["schema_version"] == "missioncore.polygon-run-catalog/v1" assert catalog["access"] == "read-only" assert catalog["total"] == 1 assert catalog["items"][0]["state"] == "completed" assert catalog["items"][0]["terminal_reason"] == "operator-stop-clean" assert detail["schema_version"] == "missioncore.polygon-run-detail/v1" assert detail["access"] == "read-only" assert detail["run"]["run_id"] == accepted.run_id assert detail["run"]["authority"]["actuator_authority"] is False assert detail["commands"] == {"count": 0, "content_exposed": False} assert detail["events_total"] == accepted.revision assert detail["events_truncated"] is True assert len(detail["events"]) == 2 assert detail["artifacts"][0]["relative_path"] == "provider-runtime/processes.json" serialized = repr({"catalog": catalog, "detail": detail}) assert str(tmp_path) not in serialized assert "command_ttl_max_ns" in serialized assert all("http" not in artifact for artifact in detail["artifacts"]) def test_polygon_api_rejects_corrupt_evidence_without_partial_response( tmp_path: Path, ) -> None: root = tmp_path / "runs" accepted = _accepted_run(root) event_path = root / accepted.run_id / "events" / "00000000000000000001.json" event_path.write_text("{broken", encoding="utf-8") router = build_polygon_router(root_provider=lambda: root) with pytest.raises(HTTPException) as failure: _endpoint(router, "/api/v1/polygon/runs", "GET")(limit=20) assert failure.value.status_code == 500 assert "целостности" in failure.value.detail def _qualification_run(root: Path) -> tuple[QualificationRun, str]: store = QualificationRunStore(root) admitted = store.create(_run("goose-ground-test")) starting = store.transition( admitted.run_id, RunState.STARTING, expected_revision=0, observed_at_utc="2026-07-24T15:35:01Z", host_monotonic_ns=1, ) store.transition( admitted.run_id, RunState.RUNNING, expected_revision=starting.revision, observed_at_utc="2026-07-24T15:35:02Z", host_monotonic_ns=2, ) evidence = root / admitted.run_id / "evidence" failures = evidence / "failures" failures.mkdir(parents=True) frame_id = "2022-07-22_flight__0071_0001" report = { "schema_version": "missioncore.goose-ground-qualification-report/v1", "identity_sha256": SHA_A, "source_id": "goose-3d/v2025-08-22", "split": "validation", "frame_count": 961, "aggregates": {"current": {}, "patchworkpp": {}}, "degradations": {}, "checks": [], "worst_frames": [{"frame_id": frame_id}], "decision": {"status": "shadow-candidate", "passed": True}, "safety": {"navigation_or_safety_accepted": False}, "frames": [{"private": "not-published"}], } preview = { "schema_version": "missioncore.goose-ground-qualification-failure-preview/v1", "source_id": "goose-3d/v2025-08-22", "frame_id": frame_id, "source_point_count": 2, "point_count": 2, "sampling": "deterministic-even-index", "points_xyz_m": [[0, 0, 0], [1, 1, 1]], "ground_truth_ground": [1, 0], "evaluated": [1, 1], "current_ground": [0, 0], "patchwork_ground": [1, 0], "current_disagreement": [1, 0], "patchwork_disagreement": [0, 0], "safety": {"navigation_or_safety_accepted": False}, } for artifact_id, kind, path, value in ( ( "qualification-report", "goose-ground-qualification-report", evidence / "qualification.json", report, ), ( "failure-preview-01", "goose-ground-qualification-failure-preview", failures / f"{frame_id}.json", preview, ), ): path.write_text(json.dumps(value), encoding="utf-8") store.register_artifact( admitted.run_id, QualificationArtifact( artifact_id=artifact_id, kind=kind, relative_path=path.relative_to(root / admitted.run_id).as_posix(), sha256=hashlib.sha256(path.read_bytes()).hexdigest(), byte_length=path.stat().st_size, source_of_record=True, ), ) current = store.load(admitted.run_id) stopping = store.transition( admitted.run_id, RunState.STOPPING, expected_revision=current.revision, observed_at_utc="2026-07-24T15:35:04Z", host_monotonic_ns=4, ) completed = store.transition( admitted.run_id, RunState.COMPLETED, expected_revision=stopping.revision, observed_at_utc="2026-07-24T15:35:05Z", host_monotonic_ns=5, reason="qualification-evidence-sealed", ) return completed, frame_id def test_polygon_api_publishes_verified_ground_qualification( tmp_path: Path, ) -> None: root = tmp_path / "runs" run, frame_id = _qualification_run(root) router = build_polygon_router(root_provider=lambda: root) qualification = _endpoint( router, "/api/v1/polygon/runs/{run_id}/qualification", "GET", )(run_id=run.run_id) preview = _endpoint( router, "/api/v1/polygon/runs/{run_id}/qualification/failures/{frame_id}", "GET", )(run_id=run.run_id, frame_id=frame_id) assert qualification["schema_version"] == "missioncore.polygon-ground-qualification/v1" assert qualification["frame_count"] == 961 assert "frames" not in qualification assert preview["schema_version"] == "missioncore.polygon-ground-failure-preview/v1" assert preview["frame_id"] == frame_id assert str(tmp_path) not in repr({"qualification": qualification, "preview": preview}) def test_polygon_api_rejects_tampered_ground_qualification( tmp_path: Path, ) -> None: root = tmp_path / "runs" run, _ = _qualification_run(root) (root / run.run_id / "evidence/qualification.json").write_text("{}", encoding="utf-8") router = build_polygon_router(root_provider=lambda: root) with pytest.raises(HTTPException) as failure: _endpoint( router, "/api/v1/polygon/runs/{run_id}/qualification", "GET", )(run_id=run.run_id) assert failure.value.status_code == 500 assert "SHA-256" in failure.value.detail or "файла" in failure.value.detail class _FakeWorkerGateway: def __init__(self) -> None: self.calls: list[tuple[str, str]] = [] self.active_run_id: str | None = None def status(self) -> dict[str, Any]: return self._status() def live(self) -> dict[str, Any]: if self.active_run_id is None: raise AssertionError("test worker has no active run") return { "schema_version": "missioncore.vehicle-state/v1", "run_id": self.active_run_id, "sequence": 1, "observed_at_utc": "2026-07-24T18:00:00Z", "host_monotonic_ns": 123, "sim_time_ns": 456, "frame_id": "map_enu", "child_frame_id": "base_link_flu", "pose": { "position_m": {"x": 1.0, "y": 2.0, "z": 0.1}, "orientation_xyzw": {"x": 0.0, "y": 0.0, "z": 0.0, "w": 1.0}, }, "source": { "provider": "gazebo", "topic": "/world/rover/dynamic_pose/info", "signal": "ground-truth", "quality": "diagnostic", }, "safety": { "scope": "virtual-only", "actuator_authority": False, "navigation_or_safety_accepted": False, }, } def start( self, *, run_id: str, mission_core_commit: str, idempotency_key: str, ) -> dict[str, Any]: self.calls.append(("start", idempotency_key)) assert mission_core_commit == "d" * 40 self.active_run_id = run_id return self._status() def stop(self, *, run_id: str, idempotency_key: str) -> dict[str, Any]: self.calls.append(("stop", idempotency_key)) assert run_id == self.active_run_id self.active_run_id = None return self._status() def command( self, *, run_id: str, speed_mps: float, steering_normalized: float, idempotency_key: str, ) -> dict[str, Any]: self.calls.append(("command", idempotency_key)) assert run_id == self.active_run_id return { "schema_version": "missioncore.command-acceptance/v2", "run_id": run_id, "command_id": "cmd-test", "sequence": 1, "issued_at_sim_ns": 1_000_000, "valid_until_sim_ns": 251_000_000, "speed_mps": speed_mps, "steering_normalized": steering_normalized, "authority_scope": "virtual-only", "delivery": { "provider_id": "px4-ros2-offboard", "control_profile": "rover-speed-steering/v1", "accepted": True, "controller_ready": True, "ttl_expired_count": 0, "diagnostics": {"armed": True, "offboard": True}, }, } def _status(self) -> dict[str, Any]: return { "schema_version": "missioncore.simulation-worker-status/v2", "worker_id": "mission-gpu-s1", "transport": "unix", "mode": "simulation", "available": True, "control_available": True, "active_run_id": self.active_run_id, "run_state": "running" if self.active_run_id else None, "active_provider_ids": (["px4-gazebo-stock-rover"] if self.active_run_id else []), "provider_profile": STOCK_ROVER_PROVIDER_PROFILE.to_dict(), "isolation": { "network": "loopback-only-netns", "process_identity": "missioncore", "artifact_policy": "d-only", }, "authority": { "scope": "virtual-only", "actuator_authority": False, "direct_actuator_setpoints_allowed": False, }, } def test_polygon_worker_api_fails_closed_then_proxies_virtual_only_lifecycle() -> None: worker = _FakeWorkerGateway() disabled = build_polygon_router( root_provider=lambda: None, worker_provider=lambda: worker, control_provider=lambda: False, commit_provider=lambda: "d" * 40, ) status = _endpoint(disabled, "/api/v1/polygon/worker", "GET")() assert status["available"] is True assert status["control_available"] is False with pytest.raises(HTTPException) as failure: _endpoint(disabled, "/api/v1/polygon/worker/runs", "POST")( request=StartStockRoverRequest(scenario_id="stock-rover-ackermann"), idempotency_key="start-disabled", ) assert failure.value.status_code == 403 enabled = build_polygon_router( root_provider=lambda: None, worker_provider=lambda: worker, control_provider=lambda: True, commit_provider=lambda: "d" * 40, ) running = _endpoint(enabled, "/api/v1/polygon/worker/runs", "POST")( request=StartStockRoverRequest(scenario_id="stock-rover-ackermann"), idempotency_key="start-001", ) assert running["active_run_id"].startswith("s1c-ddddddd-") live = _endpoint(enabled, "/api/v1/polygon/worker/live", "GET")() assert live["run_id"] == running["active_run_id"] assert live["source"]["quality"] == "diagnostic" command = _endpoint( enabled, "/api/v1/polygon/worker/runs/{run_id}/commands", "POST", )( run_id=running["active_run_id"], request=AckermannCommandRequest(speed_mps=1.0, steering_normalized=-0.25), idempotency_key="command-001", ) assert command["authority_scope"] == "virtual-only" assert command["delivery"]["controller_ready"] is True stopped = _endpoint(enabled, "/api/v1/polygon/worker/runs/{run_id}/stop", "POST")( run_id=running["active_run_id"], idempotency_key="stop-001", ) assert stopped["active_run_id"] is None assert worker.calls == [ ("start", "start-001"), ("command", "command-001"), ("stop", "stop-001"), ] def test_polygon_worker_command_rejects_values_outside_lab_envelope() -> None: with pytest.raises(ValueError): AckermannCommandRequest(speed_mps=1.51, steering_normalized=0) with pytest.raises(ValueError): AckermannCommandRequest(speed_mps=0, steering_normalized=-1.01) def test_polygon_worker_status_is_explicitly_unavailable_when_not_registered() -> None: router = build_polygon_router( root_provider=lambda: None, worker_provider=lambda: None, ) status = _endpoint(router, "/api/v1/polygon/worker", "GET")() assert status["schema_version"] == "missioncore.simulation-worker-status/v2" assert status["available"] is False assert status["control_available"] is False assert status["provider_profile"] is None assert status["authority"]["scope"] == "virtual-only"