feat(perception): establish object centric contracts
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
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) == 6
|
||||
|
||||
|
||||
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 == {}
|
||||
Reference in New Issue
Block a user