103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from k1link.perception.baseline import (
|
|
BaselineContractError,
|
|
load_m4_baseline,
|
|
validate_reuse_inventory,
|
|
verify_m4_baseline,
|
|
)
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
PERCEPTION_ROOT = REPOSITORY_ROOT / "src" / "k1link" / "perception"
|
|
BASELINE_PATH = REPOSITORY_ROOT / "config" / "perception" / "m4-recorded-realtime-baseline-v1.json"
|
|
REUSE_PATH = REPOSITORY_ROOT / "config" / "perception" / "m4-reuse-inventory-v1.json"
|
|
|
|
|
|
def _imports(path: Path) -> set[str]:
|
|
tree = ast.parse(path.read_text("utf-8"), filename=str(path))
|
|
modules: set[str] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
modules.update(alias.name for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom) and node.module:
|
|
modules.add(node.module)
|
|
return modules
|
|
|
|
|
|
def test_m4_baseline_is_exact_and_every_local_evidence_digest_resolves() -> None:
|
|
profile = load_m4_baseline(BASELINE_PATH)
|
|
verification = verify_m4_baseline(REPOSITORY_ROOT, profile)
|
|
assert verification.source_id == "RAVNOVES00"
|
|
assert verification.session_id == "20260720T065719Z_viewer_live"
|
|
assert len(verification.verified_paths) == 10
|
|
|
|
|
|
def test_m4_baseline_cannot_silently_select_another_source(tmp_path: Path) -> None:
|
|
document = json.loads(BASELINE_PATH.read_text("utf-8"))
|
|
incompatible = copy.deepcopy(document)
|
|
incompatible["source"]["source_id"] = "RAVNOVES01"
|
|
path = tmp_path / "baseline.json"
|
|
path.write_text(json.dumps(incompatible), "utf-8")
|
|
with pytest.raises(BaselineContractError, match="RAVNOVES00"):
|
|
load_m4_baseline(path)
|
|
|
|
|
|
def test_reuse_inventory_separates_primitives_from_historical_wrappers() -> None:
|
|
document = validate_reuse_inventory(REUSE_PATH)
|
|
assert document["rules"]["bulk_legacy_migration_required"] is False
|
|
|
|
|
|
def test_perception_contracts_import_no_compute_lab_graph_or_web_module() -> None:
|
|
imports = _imports(PERCEPTION_ROOT / "contracts.py")
|
|
forbidden = {
|
|
module
|
|
for module in imports
|
|
if module.startswith(
|
|
(
|
|
"k1link.compute",
|
|
"k1link.laboratory",
|
|
"k1link.web",
|
|
"k1link.perception.providers",
|
|
"k1link.perception.graph",
|
|
)
|
|
)
|
|
}
|
|
assert forbidden == set()
|
|
|
|
|
|
def test_new_perception_boundary_has_no_experiment_specific_imports() -> None:
|
|
violations: dict[str, str] = {}
|
|
for path in PERCEPTION_ROOT.glob("*.py"):
|
|
for module in _imports(path):
|
|
leaf = module.rsplit(".", 1)[-1]
|
|
if module.startswith("k1link.compute") and (
|
|
leaf.startswith("e") or leaf.startswith("l")
|
|
):
|
|
violations[path.name] = module
|
|
assert violations == {}
|
|
|
|
|
|
def test_product_perception_imports_only_admitted_compute_primitives() -> None:
|
|
inventory = validate_reuse_inventory(REUSE_PATH)
|
|
admitted = {
|
|
item["module"]
|
|
for item in inventory["reusable_primitives"]
|
|
if isinstance(item, dict) and isinstance(item.get("module"), str)
|
|
}
|
|
violations: dict[str, set[str]] = {}
|
|
for path in PERCEPTION_ROOT.glob("*.py"):
|
|
compute_imports = {
|
|
module for module in _imports(path) if module.startswith("k1link.compute")
|
|
}
|
|
unadmitted = compute_imports - admitted
|
|
if unadmitted:
|
|
violations[path.name] = unadmitted
|
|
assert violations == {}
|