254 lines
8.6 KiB
Python
254 lines
8.6 KiB
Python
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.target_host.wsl_distribution == "MissionCore-Sim"
|
|
assert profile.target_host.container_runtime == "none"
|
|
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 == 0
|
|
assert "docker" not in profile.required_tools
|
|
assert {item.identifier for item in profile.components} >= {
|
|
"px4-autopilot",
|
|
"px4-msgs",
|
|
"gazebo",
|
|
"nav2",
|
|
}
|
|
components = {item.identifier: item for item in profile.components}
|
|
assert components["px4-gazebo-models"].resolved_commit == (
|
|
"b6127f4ec20de867e215fb5f78ae88b80f371909"
|
|
)
|
|
assert components["ros2"].qualification_state == "candidate"
|
|
assert components["gazebo"].qualification_state == "candidate"
|
|
micro_xrce = next(item for item in profile.ports if item.identifier == "micro-xrce-dds")
|
|
assert micro_xrce.host_scope == "loopback-only-netns"
|
|
assert micro_xrce.bind == "0.0.0.0"
|
|
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_profile_rejects_wildcard_bind_outside_loopback_namespace(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
document = _profile_document()
|
|
ports = document["ports"]
|
|
assert isinstance(ports, list)
|
|
first = ports[0]
|
|
assert isinstance(first, dict)
|
|
first["host_scope"] = "worker-wsl"
|
|
|
|
with pytest.raises(S0ProfileError, match="wildcard bind requires loopback-only-netns"):
|
|
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
|