SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

refactor(lab): canonize selected evidence reports

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-05 11:49:44 +03:00
parent 4c763bd8aa
commit de12e96297
30 changed files with 2876 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_advanced_laboratory_api.py
+24
View File
@@ -74,6 +74,30 @@ def test_advanced_index_is_empty_when_not_configured() -> None:
}
def test_advanced_index_includes_valid_l31_identity(
tmp_path: Path,
monkeypatch: MonkeyPatch,
) -> None:
identity = {
"result_id": f"l31-pointpillars-ravnoves-{'a' * 64}",
"created_at_utc": "2026-07-31T10:56:45.861Z",
}
monkeypatch.setattr(advanced_api, "latest_l31_identity", lambda _provider: identity)
router = build_advanced_laboratory_router(
l31_ravnoves_root_provider=lambda: tmp_path,
)
index = _endpoint(router, "/api/v1/laboratory/advanced-index")()
assert index["items"] == [ # type: ignore[index]
{
"work_id": "l31-pointpillars-ravnoves",
**identity,
"access": "read-only",
}
]
def test_registry_index_does_not_follow_a_runtime_symlink(tmp_path: Path) -> None:
actual = tmp_path / "actual"
actual.mkdir()
tests/test_laboratory_evidence_report.py
+159
View File
@@ -0,0 +1,159 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import pytest
from k1link.laboratory import (
LaboratoryEvidenceRegistry,
LaboratoryEvidenceReportError,
LaboratoryEvidenceReportNotFound,
LaboratoryEvidenceReportService,
)
def _canonical_sha256(value: object) -> str:
return hashlib.sha256(
json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
).hexdigest()
def _write_json(path: Path, value: object) -> bytes:
encoded = json.dumps(value, ensure_ascii=False, indent=2).encode("utf-8")
path.write_bytes(encoded)
return encoded
def _service(tmp_path: Path) -> tuple[LaboratoryEvidenceReportService, str, Path]:
definitions = tmp_path / "definitions"
definitions.mkdir()
_write_json(
definitions / "e99-proof.json",
{
"schema_version": "missioncore.laboratory-evidence-definition/v1",
"work_id": "e99-proof",
"evidence": {
"runtime_relative_root": "e99/results",
"result_id_prefix": "e99-proof",
"document_name": "manifest.json",
"schema_version": "missioncore.e99-result/v1",
},
},
)
runtime = tmp_path / "runtime"
results = runtime / "e99" / "results"
results.mkdir(parents=True)
source = {"session_id": "immutable-source", "stream_sha256": "a" * 64}
model = {"name": "detector", "sha256": "b" * 64}
identity = {
"schema_version": "missioncore.e99-result/v1",
"profile": {
"source": source,
"model": model,
},
"worker": {"node": "worker-006", "container": "ndc-worker@sha256:proof"},
"authority": {"commands_enabled": False, "navigation_or_safety_accepted": False},
}
identity_sha256 = _canonical_sha256(identity)
result_id = f"e99-proof-{identity_sha256}"
result_root = results / result_id
result_root.mkdir()
report = {
"schema_version": "missioncore.e99-report/v1",
"source": source,
"method": {"components": [model]},
"metrics": {
"frames": 100,
"resources": {"gpu_utilization_p95_percent": 48.0, "rss_p95_mib": 91.0},
},
"acceptance": {"passed": True, "checks": {"frame_coverage": True}},
"decision": {"promoted": False, "next_action": "temporal gate"},
"limitations": ["No independent ground truth."],
"authority": identity["authority"],
}
report_bytes = _write_json(result_root / "report.json", report)
visual_bytes = b"visual-proof"
(result_root / "visual.png").write_bytes(visual_bytes)
manifest = {
"schema_version": "missioncore.e99-result/v1",
"result_id": result_id,
"identity_sha256": identity_sha256,
"identity": identity,
"created_at_utc": "2026-08-05T09:00:00Z",
"artifacts": [
{
"role": "engineering-report",
"path": "report.json",
"byte_length": len(report_bytes),
"sha256": hashlib.sha256(report_bytes).hexdigest(),
},
{
"role": "visual-contact-sheet",
"path": "visual.png",
"byte_length": len(visual_bytes),
"sha256": hashlib.sha256(visual_bytes).hexdigest(),
},
],
}
_write_json(result_root / "manifest.json", manifest)
registry = LaboratoryEvidenceRegistry.from_directory(definitions)
return LaboratoryEvidenceReportService(registry, lambda: runtime), result_id, result_root
def test_evidence_report_projects_verified_canonical_evidence(tmp_path: Path) -> None:
service, result_id, _ = _service(tmp_path)
report = service.read("e99-proof", result_id)
assert report["schema_version"] == "missioncore.laboratory-evidence-report/v1"
assert report["result_id"] == result_id
assert report["source"]["session_id"] == "immutable-source" # type: ignore[index]
assert report["configuration"]["profile"]["model"]["name"] == "detector" # type: ignore[index]
assert report["resources"]["gpu_utilization_p95_percent"] == 48.0 # type: ignore[index]
assert report["proof"]["verified_artifact_count"] == 2 # type: ignore[index]
assert report["completeness"]["visual_evidence"] == "recorded" # type: ignore[index]
assert report["completeness"]["execution"] == "recorded" # type: ignore[index]
def test_evidence_report_rejects_tampered_artifact(tmp_path: Path) -> None:
service, result_id, result_root = _service(tmp_path)
(result_root / "visual.png").write_bytes(b"tampered")
with pytest.raises(LaboratoryEvidenceReportError, match="does not match"):
service.read("e99-proof", result_id)
def test_evidence_report_rejects_unknown_identity(tmp_path: Path) -> None:
service, _, _ = _service(tmp_path)
with pytest.raises(LaboratoryEvidenceReportNotFound, match="unavailable"):
service.read("e99-proof", f"e99-proof-{'f' * 64}")
def test_evidence_report_marks_missing_fields_without_inventing_them(tmp_path: Path) -> None:
service, result_id, result_root = _service(tmp_path)
report_path = result_root / "report.json"
report = json.loads(report_path.read_text(encoding="utf-8"))
report.pop("metrics")
report.pop("limitations")
report_bytes = _write_json(report_path, report)
manifest_path = result_root / "manifest.json"
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
manifest["artifacts"][0]["byte_length"] = len(report_bytes)
manifest["artifacts"][0]["sha256"] = hashlib.sha256(report_bytes).hexdigest()
_write_json(manifest_path, manifest)
evidence = service.read("e99-proof", result_id)
assert evidence["metrics"] is None
assert evidence["limitations"] is None
assert evidence["completeness"]["metrics"] == "not-recorded" # type: ignore[index]
assert evidence["completeness"]["limitations"] == "not-recorded" # type: ignore[index]
tests/test_laboratory_value_review_registry.py
+90
View File
@@ -0,0 +1,90 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from fastapi.routing import APIRoute
from k1link.laboratory import (
LaboratoryValueReviewRegistry,
LaboratoryValueReviewRegistryError,
)
from k1link.web.laboratory_report_api import build_laboratory_report_router
def _document() -> dict[str, object]:
return {
"schema_version": "missioncore.laboratory-value-review-registry/v1",
"reviewed_at_utc": "2026-08-05T08:30:00Z",
"entries": [
{
"catalog_id": "e46j-raw-fisheye-realtime",
"evidence_id": f"e46j-raw-fisheye-realtime-{'a' * 64}",
"signal": "progress",
"lifecycle": "current",
"visual_evidence": "available",
}
],
}
def _write(path: Path, document: dict[str, object]) -> None:
path.write_text(json.dumps(document), encoding="utf-8")
def test_value_review_registry_is_strict_and_projects_read_only_index(tmp_path: Path) -> None:
path = tmp_path / "laboratory-value-review.json"
_write(path, _document())
registry = LaboratoryValueReviewRegistry.from_file(path)
router = build_laboratory_report_router(registry)
route = next(
route
for route in router.routes
if isinstance(route, APIRoute)
and route.path == "/api/v1/laboratory/value-review-index"
)
payload = route.endpoint()
assert payload["schema_version"] == "missioncore.laboratory-value-review-index/v1"
assert payload["items"][0]["signal"] == "progress"
assert payload["items"][0]["access"] == "read-only"
def test_value_review_registry_rejects_unknown_keys(tmp_path: Path) -> None:
path = tmp_path / "laboratory-value-review.json"
document = _document()
document["renderer"] = "local"
_write(path, document)
with pytest.raises(LaboratoryValueReviewRegistryError, match="unexpected"):
LaboratoryValueReviewRegistry.from_file(path)
def test_value_review_registry_rejects_duplicate_catalog_ids(tmp_path: Path) -> None:
path = tmp_path / "laboratory-value-review.json"
document = _document()
entries = document["entries"]
assert isinstance(entries, list)
entries.append({**entries[0], "evidence_id": f"duplicate-{'b' * 64}"})
_write(path, document)
with pytest.raises(LaboratoryValueReviewRegistryError, match="duplicate"):
LaboratoryValueReviewRegistry.from_file(path)
def test_product_value_review_registry_covers_reviewed_laboratory_families() -> None:
root = Path(__file__).resolve().parents[1]
registry = LaboratoryValueReviewRegistry.from_file(
root / "config" / "laboratory-value-review.json"
)
assert len(registry.entries) == 34
assert {entry.catalog_id for entry in registry.entries} >= {
"e28-local-surface",
"e46d-temporal-failure-audit",
"e46j-raw-fisheye-realtime",
"l31-pointpillars-ravnoves",
"l34f-adjudicated-reference",
}