feat(perception): audit E36 second-source eligibility
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,231 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from k1link.compute.e36_source_catalog_audit import (
|
||||
E36SourceCatalogAuditError,
|
||||
build_e36_source_catalog_audit,
|
||||
read_e36_source_catalog_audit,
|
||||
)
|
||||
|
||||
SHA = "a" * 64
|
||||
BASELINE = "20260720T065719Z_viewer_live"
|
||||
CANDIDATE = "20260718T201659Z_viewer_live"
|
||||
|
||||
|
||||
def _profile() -> dict[str, Any]:
|
||||
return {
|
||||
"schema_version": "missioncore.e36-source-eligibility-profile/v1",
|
||||
"profile_id": "test/v1",
|
||||
"baseline": {
|
||||
"source_session_id": BASELINE,
|
||||
"frozen_upstream": {
|
||||
"e31": "e31-" + SHA,
|
||||
"e32": "e32-" + SHA,
|
||||
"e33": "e33-" + SHA,
|
||||
"e34": "e34-" + SHA,
|
||||
"e35": "e35-" + SHA,
|
||||
},
|
||||
},
|
||||
"eligibility": {
|
||||
"required_status": "ready",
|
||||
"required_origin": "xgrids-k1.viewer-live.evidence",
|
||||
"required_modalities": ["point-cloud", "trajectory", "video"],
|
||||
"minimum_contiguous_camera_seconds": 60.0,
|
||||
"require_host_time_identity": True,
|
||||
"require_device_identity": True,
|
||||
"require_session_bound_calibration": True,
|
||||
"require_session_bound_mount": True,
|
||||
"retuning_allowed": False,
|
||||
},
|
||||
"authority": {
|
||||
"commands_enabled": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
"transfer_replay_authorized": False,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _source(session_id: str, display_name: str) -> dict[str, Any]:
|
||||
return {
|
||||
"session_id": session_id,
|
||||
"display_name": display_name,
|
||||
"status": "ready",
|
||||
"origin": "xgrids-k1.viewer-live.evidence",
|
||||
"duration_seconds": 200.0,
|
||||
"modalities": ["point-cloud", "trajectory", "video"],
|
||||
"raw_transport_sha256": SHA,
|
||||
"host_time_identity_sha256": SHA,
|
||||
"device_identity": {
|
||||
"device_serial_sha256": SHA,
|
||||
"vendor_device_id_sha256": SHA,
|
||||
},
|
||||
"camera": {
|
||||
"source_id": "sensor.camera.left",
|
||||
"frame_count": 1000,
|
||||
"longest_contiguous_seconds": 100.0,
|
||||
"status": "complete",
|
||||
},
|
||||
"calibration_binding": {
|
||||
"source_session_id": session_id,
|
||||
"calibration_identity_sha256": SHA,
|
||||
},
|
||||
"mount_binding": {
|
||||
"source_session_id": session_id,
|
||||
"mount_identity_sha256": SHA,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _write(path: Path, value: object) -> None:
|
||||
path.write_text(json.dumps(value), encoding="utf-8")
|
||||
|
||||
|
||||
def _build(
|
||||
tmp_path: Path,
|
||||
*,
|
||||
candidate: dict[str, Any] | None = None,
|
||||
):
|
||||
profile_path = tmp_path / "profile.json"
|
||||
inventory_path = tmp_path / "inventory.json"
|
||||
_write(profile_path, _profile())
|
||||
_write(
|
||||
inventory_path,
|
||||
{
|
||||
"schema_version": "missioncore.e36-source-catalog-inventory/v1",
|
||||
"catalog_audited_at_utc": "2026-07-27T13:00:00Z",
|
||||
"sources": [
|
||||
_source(BASELINE, "RAVNOVES00"),
|
||||
candidate or _source(CANDIDATE, "TEST007"),
|
||||
],
|
||||
},
|
||||
)
|
||||
return build_e36_source_catalog_audit(
|
||||
profile_path=profile_path,
|
||||
inventory_path=inventory_path,
|
||||
output_root=tmp_path / "results",
|
||||
)
|
||||
|
||||
|
||||
def test_eligible_second_source_authorizes_only_the_next_replay_gate(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
result = _build(tmp_path)
|
||||
|
||||
assert result.report["status"] == "eligible-second-mounted-source-found"
|
||||
assert result.eligible_source_ids == (CANDIDATE,)
|
||||
assert result.report["decision"] == {
|
||||
"e36_replay_authorized": True,
|
||||
"eligible_source_count": 1,
|
||||
"retuning_allowed": False,
|
||||
"empty_lab_publication_allowed": False,
|
||||
"new_capture_authorized": False,
|
||||
"public_dataset_substitution_allowed": False,
|
||||
"nearest_candidate": {
|
||||
"session_id": CANDIDATE,
|
||||
"display_name": "TEST007",
|
||||
"remaining_blocker_codes": [],
|
||||
},
|
||||
}
|
||||
assert result.report["authority"]["commands_enabled"] is False
|
||||
|
||||
|
||||
def test_missing_session_bound_provenance_blocks_without_empty_lab(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
candidate = _source(CANDIDATE, "TEST007")
|
||||
candidate["calibration_binding"] = None
|
||||
candidate["mount_binding"] = None
|
||||
|
||||
result = _build(tmp_path, candidate=candidate)
|
||||
|
||||
assert result.eligible_source_ids == ()
|
||||
assert result.report["status"] == "blocked-no-eligible-second-mounted-source"
|
||||
assert result.report["next_gate"] == (
|
||||
"bind-candidate-calibration-and-mount-provenance"
|
||||
)
|
||||
nearest = result.report["decision"]["nearest_candidate"]
|
||||
assert nearest["remaining_blocker_codes"] == [
|
||||
"calibration-bound-to-source-session",
|
||||
"mount-bound-to-source-session",
|
||||
]
|
||||
assert result.report["decision"]["empty_lab_publication_allowed"] is False
|
||||
|
||||
|
||||
def test_calibration_from_another_session_does_not_transfer(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
candidate = _source(CANDIDATE, "TEST007")
|
||||
candidate["calibration_binding"]["source_session_id"] = BASELINE
|
||||
|
||||
result = _build(tmp_path, candidate=candidate)
|
||||
|
||||
nearest = result.report["decision"]["nearest_candidate"]
|
||||
assert nearest["remaining_blocker_codes"] == [
|
||||
"calibration-bound-to-source-session"
|
||||
]
|
||||
|
||||
|
||||
def test_result_is_content_addressed_and_detects_report_tampering(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
first = _build(tmp_path)
|
||||
second = _build(tmp_path)
|
||||
assert first.result_id == second.result_id
|
||||
|
||||
report_path = first.result_root / "audit-report.json"
|
||||
report = json.loads(report_path.read_text(encoding="utf-8"))
|
||||
report["status"] = "changed"
|
||||
_write(report_path, report)
|
||||
with pytest.raises(E36SourceCatalogAuditError):
|
||||
read_e36_source_catalog_audit(first.result_root)
|
||||
|
||||
|
||||
def test_profile_cannot_relax_the_mount_gate(tmp_path: Path) -> None:
|
||||
profile = _profile()
|
||||
relaxed = copy.deepcopy(profile)
|
||||
relaxed["eligibility"]["require_session_bound_mount"] = False
|
||||
profile_path = tmp_path / "profile.json"
|
||||
inventory_path = tmp_path / "inventory.json"
|
||||
_write(profile_path, relaxed)
|
||||
_write(
|
||||
inventory_path,
|
||||
{
|
||||
"schema_version": "missioncore.e36-source-catalog-inventory/v1",
|
||||
"catalog_audited_at_utc": "2026-07-27T13:00:00Z",
|
||||
"sources": [_source(BASELINE, "RAVNOVES00")],
|
||||
},
|
||||
)
|
||||
with pytest.raises(E36SourceCatalogAuditError):
|
||||
build_e36_source_catalog_audit(
|
||||
profile_path=profile_path,
|
||||
inventory_path=inventory_path,
|
||||
output_root=tmp_path / "results",
|
||||
)
|
||||
|
||||
|
||||
def test_profile_rejects_non_finite_camera_gate(tmp_path: Path) -> None:
|
||||
profile = _profile()
|
||||
profile["eligibility"]["minimum_contiguous_camera_seconds"] = float("inf")
|
||||
profile_path = tmp_path / "profile.json"
|
||||
inventory_path = tmp_path / "inventory.json"
|
||||
_write(profile_path, profile)
|
||||
_write(
|
||||
inventory_path,
|
||||
{
|
||||
"schema_version": "missioncore.e36-source-catalog-inventory/v1",
|
||||
"catalog_audited_at_utc": "2026-07-27T13:00:00Z",
|
||||
"sources": [_source(BASELINE, "RAVNOVES00")],
|
||||
},
|
||||
)
|
||||
with pytest.raises(E36SourceCatalogAuditError):
|
||||
build_e36_source_catalog_audit(
|
||||
profile_path=profile_path,
|
||||
inventory_path=inventory_path,
|
||||
output_root=tmp_path / "results",
|
||||
)
|
||||
Reference in New Issue
Block a user