101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from k1link.compute.e44_data_amplification_audit import (
|
|
build_e44_data_amplification_audit,
|
|
)
|
|
from k1link.compute.e50_content_reference_index import (
|
|
build_e50_content_reference_index,
|
|
)
|
|
from k1link.web.artifact_health_api import ArtifactHealthService
|
|
|
|
|
|
def _write(path: Path, payload: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(payload)
|
|
|
|
|
|
def test_artifact_health_reports_live_inventory_and_exact_audit(tmp_path: Path) -> None:
|
|
runtime = tmp_path / ".runtime"
|
|
first = runtime / "compute-experiments" / "e30" / "result-a"
|
|
second = runtime / "compute-experiments" / "e40" / "result-b"
|
|
_write(first / "shared.bin", b"shared")
|
|
_write(first / "only-a.bin", b"a")
|
|
_write(second / "shared.bin", b"shared")
|
|
_write(second / "only-b.bin", b"bb")
|
|
e44_results = runtime / "compute-experiments" / "e44" / "results"
|
|
e44 = build_e44_data_amplification_audit(
|
|
artifact_roots={"e30": first, "e40": second},
|
|
output_root=e44_results,
|
|
)
|
|
e50_results = runtime / "compute-experiments" / "e50" / "results"
|
|
e50 = build_e50_content_reference_index(
|
|
e44_result_root=e44.result_root,
|
|
artifact_roots={"e30": first, "e40": second},
|
|
output_root=e50_results,
|
|
)
|
|
|
|
service = ArtifactHealthService(
|
|
lambda: runtime,
|
|
lambda: e44_results,
|
|
e50_results_root_provider=lambda: e50_results,
|
|
cache_seconds=0,
|
|
)
|
|
document = service.snapshot()
|
|
|
|
assert document.inventory.state == "live"
|
|
assert document.inventory.file_count >= 6
|
|
assert document.inventory.logical_bytes > 0
|
|
assert document.exact_audit.state == "exact-current"
|
|
assert document.exact_audit.duplicate_bytes == len(b"shared")
|
|
assert document.content_references.state == "indexed"
|
|
assert document.content_references.result_id == e50.result_id
|
|
assert document.content_references.logical_reference_count == 4
|
|
assert document.content_references.canonical_content_count == 3
|
|
assert document.content_references.duplicate_reference_count == 1
|
|
assert document.content_references.exact_duplicate_bytes == len(b"shared")
|
|
assert document.content_references.physical_reclamation_applied is False
|
|
assert document.content_references.storage_migration_authorized is False
|
|
|
|
|
|
def test_artifact_health_marks_exact_audit_stale_after_source_change(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
runtime = tmp_path / ".runtime"
|
|
first = runtime / "e30" / "result-a"
|
|
second = runtime / "e40" / "result-b"
|
|
_write(first / "a.bin", b"a")
|
|
_write(second / "b.bin", b"b")
|
|
e44_results = runtime / "e44" / "results"
|
|
audit = build_e44_data_amplification_audit(
|
|
artifact_roots={"e30": first, "e40": second},
|
|
output_root=e44_results,
|
|
)
|
|
manifest = json.loads((audit.result_root / "manifest.json").read_text())
|
|
first_name = manifest["identity"]["artifact_roots"][0]["root_name"]
|
|
_write(first / "new.bin", b"new")
|
|
|
|
document = ArtifactHealthService(
|
|
lambda: runtime,
|
|
lambda: e44_results,
|
|
cache_seconds=0,
|
|
).snapshot()
|
|
|
|
assert document.exact_audit.state == "exact-stale"
|
|
assert first_name in document.exact_audit.changed_roots
|
|
|
|
|
|
def test_artifact_health_reports_unavailable_runtime(tmp_path: Path) -> None:
|
|
document = ArtifactHealthService(
|
|
lambda: tmp_path / "missing",
|
|
lambda: tmp_path / "missing-e44",
|
|
cache_seconds=0,
|
|
).snapshot()
|
|
|
|
assert document.overall_state == "unavailable"
|
|
assert document.inventory.state == "unavailable"
|
|
assert document.exact_audit.state == "unavailable"
|
|
assert document.content_references.state == "not-indexed"
|