feat(simulation): establish polygon S0 qualification

2026-07-24 15:29:08 +03:00
parent 7fba39a629
commit 81761067e9
11 changed files with 2318 additions and 0 deletions
@@ -0,0 +1,227 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+import yaml
+from typer.testing import CliRunner
+
+from k1link.simulation import (
+    CheckStatus,
+    DoctorVerdict,
+    S0ProfileError,
+    load_s0_profile,
+    run_s0_doctor,
+)
+from k1link.simulation.cli import app
+
+PROFILE = Path(__file__).resolve().parents[1] / "simulation" / "s0" / "qualification-profile.yaml"
+runner = CliRunner()
+
+
+def _profile_document() -> dict[str, object]:
+    loaded = yaml.safe_load(PROFILE.read_text(encoding="utf-8"))
+    assert isinstance(loaded, dict)
+    return loaded
+
+
+def _write_profile(tmp_path: Path, document: dict[str, object]) -> Path:
+    path = tmp_path / "profile.yaml"
+    path.write_text(yaml.safe_dump(document, sort_keys=False), encoding="utf-8")
+    return path
+
+
+def test_canonical_s0_profile_is_strict_and_safe() -> None:
+    profile = load_s0_profile(PROFILE)
+
+    assert profile.profile_id == "ai-worker-px4-gazebo-v1"
+    assert profile.storage.windows_root.drive.upper() == "D:"
+    assert profile.storage.wsl_root.as_posix() == "/mnt/d/NDC_MISSIONCORE"
+    assert profile.clock.authority == "gazebo:/clock"
+    assert profile.clock.ros_use_sim_time is True
+    assert profile.clock.px4_uxrce_dds_sync_enabled is False
+    assert profile.authority.simulation_or_shadow_only is True
+    assert profile.authority.actuator_authority is False
+    assert profile.authority.navigation_or_safety_accepted is False
+    assert profile.authority.direct_actuator_setpoints_allowed is False
+    assert profile.ros.domain_id == 42
+    assert {item.identifier for item in profile.components} >= {
+        "px4-autopilot",
+        "px4-msgs",
+        "gazebo",
+        "nav2",
+    }
+    assert {item.identifier for item in profile.processes} == {
+        "simulation-orchestrator",
+        "gazebo-server",
+        "px4-sitl",
+        "micro-xrce-dds-agent",
+        "ros2-bridge",
+        "nav2",
+    }
+
+
+def test_profile_rejects_real_actuator_authority(tmp_path: Path) -> None:
+    document = _profile_document()
+    authority = document["authority"]
+    assert isinstance(authority, dict)
+    authority["actuator_authority"] = True
+
+    with pytest.raises(S0ProfileError, match="cannot grant real"):
+        load_s0_profile(_write_profile(tmp_path, document))
+
+
+def test_profile_rejects_mutable_c_drive_path(tmp_path: Path) -> None:
+    document = _profile_document()
+    storage = document["storage"]
+    assert isinstance(storage, dict)
+    mutable = storage["mutable_paths"]
+    assert isinstance(mutable, list)
+    first = mutable[0]
+    assert isinstance(first, dict)
+    first["windows_path"] = r"C:\NDC_MISSIONCORE\wsl"
+
+    with pytest.raises(S0ProfileError, match="escapes D-only root"):
+        load_s0_profile(_write_profile(tmp_path, document))
+
+
+def test_profile_rejects_duplicate_scoped_port(tmp_path: Path) -> None:
+    document = _profile_document()
+    ports = document["ports"]
+    assert isinstance(ports, list)
+    duplicate = dict(ports[0])
+    duplicate["id"] = "duplicate-port"
+    duplicate["owner"] = "simulation-orchestrator"
+    ports.append(duplicate)
+
+    with pytest.raises(S0ProfileError, match="duplicates a scoped binding"):
+        load_s0_profile(_write_profile(tmp_path, document))
+
+
+def test_doctor_never_claims_go_without_target_evidence(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "k1link.simulation.s0._target_host_match",
+        lambda target: (False, "test host is not target"),
+    )
+
+    report = run_s0_doctor(PROFILE)
+
+    assert report.verdict is DoctorVerdict.INCOMPLETE
+    assert report.target_host_match is False
+    assert any(
+        check.identifier == "factual-evidence" and check.status is CheckStatus.UNKNOWN
+        for check in report.checks
+    )
+    payload = report.to_dict()
+    assert payload["authority"] == {
+        "simulation_or_shadow_only": True,
+        "actuator_authority": False,
+        "navigation_or_safety_accepted": False,
+        "direct_actuator_setpoints_allowed": False,
+    }
+
+
+def test_doctor_validates_digest_bound_evidence(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setattr(
+        "k1link.simulation.s0._target_host_match",
+        lambda target: (False, "test host is not target"),
+    )
+    profile = load_s0_profile(PROFILE)
+    checks: list[dict[str, object]] = []
+    for identifier in profile.required_evidence:
+        artifact = tmp_path / f"{identifier}.txt"
+        artifact.write_text(f"evidence for {identifier}\n", encoding="utf-8")
+        checks.append(
+            {
+                "id": identifier,
+                "status": "pass",
+                "artifact": artifact.name,
+                "sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
+                "note": "synthetic unit-test evidence",
+            }
+        )
+    evidence = tmp_path / "evidence.yaml"
+    evidence.write_text(
+        yaml.safe_dump(
+            {
+                "schema_version": "missioncore.simulation-s0-evidence/v1",
+                "profile_sha256": hashlib.sha256(PROFILE.read_bytes()).hexdigest(),
+                "checks": checks,
+            },
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+
+    report = run_s0_doctor(PROFILE, evidence_path=evidence)
+
+    assert report.verdict is DoctorVerdict.INCOMPLETE
+    assert any(
+        check.identifier == "factual-evidence" and check.status is CheckStatus.PASS
+        for check in report.checks
+    )
+
+
+def test_doctor_rejects_changed_evidence_artifact(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setattr(
+        "k1link.simulation.s0._target_host_match",
+        lambda target: (False, "test host is not target"),
+    )
+    profile = load_s0_profile(PROFILE)
+    checks: list[dict[str, object]] = []
+    artifacts: list[Path] = []
+    for identifier in profile.required_evidence:
+        artifact = tmp_path / f"{identifier}.txt"
+        artifact.write_text("original\n", encoding="utf-8")
+        artifacts.append(artifact)
+        checks.append(
+            {
+                "id": identifier,
+                "status": "pass",
+                "artifact": artifact.name,
+                "sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
+                "note": "synthetic unit-test evidence",
+            }
+        )
+    evidence = tmp_path / "evidence.yaml"
+    evidence.write_text(
+        yaml.safe_dump(
+            {
+                "schema_version": "missioncore.simulation-s0-evidence/v1",
+                "profile_sha256": hashlib.sha256(PROFILE.read_bytes()).hexdigest(),
+                "checks": checks,
+            },
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+    artifacts[0].write_text("changed\n", encoding="utf-8")
+
+    report = run_s0_doctor(PROFILE, evidence_path=evidence)
+
+    assert report.verdict is DoctorVerdict.BLOCKED
+    assert any(
+        check.identifier == "factual-evidence"
+        and check.status is CheckStatus.FAIL
+        and "digest changed" in check.detail
+        for check in report.checks
+    )
+
+
+def test_simulation_doctor_cli_emits_json_without_side_effects() -> None:
+    result = runner.invoke(app, ["s0", "doctor", "--profile", str(PROFILE), "--json"])
+
+    assert result.exit_code == 0
+    payload = json.loads(result.stdout)
+    assert payload["schema_version"] == "missioncore.simulation-s0-doctor/v1"
+    assert payload["profile_id"] == "ai-worker-px4-gazebo-v1"
+    assert payload["verdict"] in {"incomplete", "blocked"}
+    assert payload["authority"]["actuator_authority"] is False