mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
test: require exhaustive contract ownership
This commit is contained in:
Joseph Magly
+198
-14
@@ -5,11 +5,33 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
RISK_CLASSES = {"cpu-contract", "mixed-runtime", "conditional-runtime"}
|
||||
CONTRACT_TYPES = {
|
||||
"ablation-strategy",
|
||||
"architecture-selection",
|
||||
"configuration",
|
||||
"device-boundary",
|
||||
"external-service",
|
||||
"model-mutation",
|
||||
"model-runtime",
|
||||
"numerical-invariant",
|
||||
"operator-ui",
|
||||
"orchestration",
|
||||
"package-entrypoint",
|
||||
"persistence",
|
||||
"public-interface",
|
||||
"remote-execution",
|
||||
"reproducibility",
|
||||
"research-input",
|
||||
"research-metric",
|
||||
"research-output",
|
||||
}
|
||||
SURFACE_ID = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
||||
|
||||
|
||||
def _load_object(path: Path, label: str, errors: list[str]) -> dict[str, Any]:
|
||||
@@ -24,6 +46,152 @@ def _load_object(path: Path, label: str, errors: list[str]) -> dict[str, Any]:
|
||||
return value
|
||||
|
||||
|
||||
def _project_root(risk_path: Path, quality_path: Path) -> Path:
|
||||
"""Resolve the checkout root even when a test supplies temporary policy files."""
|
||||
|
||||
candidates = [
|
||||
risk_path.resolve().parent.parent,
|
||||
quality_path.resolve().parent.parent,
|
||||
Path.cwd().resolve(),
|
||||
]
|
||||
for candidate in candidates:
|
||||
if (candidate / "obliteratus").is_dir() and (candidate / "tests").is_dir():
|
||||
return candidate
|
||||
return Path.cwd().resolve()
|
||||
|
||||
|
||||
def _validate_test_paths(
|
||||
tests: object,
|
||||
*,
|
||||
label: str,
|
||||
project_root: Path,
|
||||
errors: list[str],
|
||||
) -> None:
|
||||
if not isinstance(tests, list) or not tests:
|
||||
errors.append(f"{label} requires at least one test")
|
||||
return
|
||||
if len(tests) != len(set(map(str, tests))):
|
||||
errors.append(f"{label} has duplicate test paths")
|
||||
for test_path in tests:
|
||||
if not isinstance(test_path, str) or not test_path.startswith("tests/"):
|
||||
errors.append(f"{label} has invalid test path: {test_path!r}")
|
||||
elif not (project_root / test_path).is_file():
|
||||
errors.append(f"{label} maps missing test path: {test_path}")
|
||||
|
||||
|
||||
def _inventory_sources(
|
||||
inventory: object,
|
||||
*,
|
||||
project_root: Path,
|
||||
errors: list[str],
|
||||
) -> set[str]:
|
||||
if not isinstance(inventory, dict):
|
||||
errors.append("test risk map source_inventory must be an object")
|
||||
return set()
|
||||
roots = inventory.get("roots")
|
||||
if not isinstance(roots, list) or not roots:
|
||||
errors.append("test risk map source_inventory requires non-empty roots")
|
||||
return set()
|
||||
if len(roots) != len(set(map(str, roots))):
|
||||
errors.append("test risk map source_inventory has duplicate roots")
|
||||
|
||||
sources: set[str] = set()
|
||||
for value in roots:
|
||||
if not isinstance(value, str) or not value or Path(value).is_absolute():
|
||||
errors.append(f"source_inventory has invalid root: {value!r}")
|
||||
continue
|
||||
root = project_root / value
|
||||
if root.is_file():
|
||||
if root.suffix != ".py":
|
||||
errors.append(f"source_inventory root is not Python source: {value}")
|
||||
else:
|
||||
sources.add(root.relative_to(project_root).as_posix())
|
||||
elif root.is_dir():
|
||||
sources.update(
|
||||
path.relative_to(project_root).as_posix()
|
||||
for path in root.rglob("*.py")
|
||||
if path.is_file()
|
||||
)
|
||||
else:
|
||||
errors.append(f"source_inventory root does not exist: {value}")
|
||||
return sources
|
||||
|
||||
|
||||
def _validate_contract_surfaces(
|
||||
surfaces: object,
|
||||
*,
|
||||
inventory_sources: set[str],
|
||||
project_root: Path,
|
||||
errors: list[str],
|
||||
) -> dict[str, str]:
|
||||
if not isinstance(surfaces, list) or not surfaces:
|
||||
errors.append("test risk map contract_surfaces must be a non-empty list")
|
||||
return {}
|
||||
|
||||
surface_ids: set[str] = set()
|
||||
source_owner: dict[str, str] = {}
|
||||
for index, surface in enumerate(surfaces):
|
||||
label = f"contract surface {index}"
|
||||
if not isinstance(surface, dict):
|
||||
errors.append(f"{label} must be an object")
|
||||
continue
|
||||
surface_id = surface.get("id")
|
||||
if not isinstance(surface_id, str) or not SURFACE_ID.fullmatch(surface_id):
|
||||
errors.append(f"{label} has invalid id: {surface_id!r}")
|
||||
surface_id = f"index-{index}"
|
||||
elif surface_id in surface_ids:
|
||||
errors.append(f"duplicate contract surface id: {surface_id}")
|
||||
surface_ids.add(surface_id)
|
||||
label = f"contract surface {surface_id}"
|
||||
|
||||
if not isinstance(surface.get("owner"), str) or not surface["owner"].strip():
|
||||
errors.append(f"{label} requires a non-empty owner")
|
||||
if not isinstance(surface.get("description"), str) or not surface["description"].strip():
|
||||
errors.append(f"{label} requires a non-empty description")
|
||||
|
||||
contract_types = surface.get("contract_types")
|
||||
if not isinstance(contract_types, list) or not contract_types:
|
||||
errors.append(f"{label} requires at least one contract type")
|
||||
else:
|
||||
if len(contract_types) != len(set(map(str, contract_types))):
|
||||
errors.append(f"{label} has duplicate contract types")
|
||||
for contract_type in contract_types:
|
||||
if not isinstance(contract_type, str) or contract_type not in CONTRACT_TYPES:
|
||||
errors.append(f"{label} has invalid contract type: {contract_type!r}")
|
||||
|
||||
_validate_test_paths(
|
||||
surface.get("required_tests"),
|
||||
label=label,
|
||||
project_root=project_root,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
paths = surface.get("paths")
|
||||
if not isinstance(paths, list) or not paths:
|
||||
errors.append(f"{label} requires at least one source path")
|
||||
continue
|
||||
if len(paths) != len(set(map(str, paths))):
|
||||
errors.append(f"{label} has duplicate source paths")
|
||||
for source_path in paths:
|
||||
if not isinstance(source_path, str) or not source_path:
|
||||
errors.append(f"{label} has invalid source path: {source_path!r}")
|
||||
continue
|
||||
if source_path not in inventory_sources:
|
||||
errors.append(f"{label} maps source outside inventory: {source_path}")
|
||||
previous = source_owner.get(source_path)
|
||||
if previous is not None:
|
||||
errors.append(
|
||||
f"production source path has multiple contract owners: "
|
||||
f"{source_path} ({previous}, {surface_id})"
|
||||
)
|
||||
else:
|
||||
source_owner[source_path] = surface_id
|
||||
|
||||
for source_path in sorted(inventory_sources - source_owner.keys()):
|
||||
errors.append(f"unmapped production source path: {source_path}")
|
||||
return source_owner
|
||||
|
||||
|
||||
def validate(risk_path: Path, quality_path: Path, conditional_path: Path) -> list[str]:
|
||||
"""Return structural and cross-policy failures for the risk map."""
|
||||
|
||||
@@ -34,11 +202,24 @@ def validate(risk_path: Path, quality_path: Path, conditional_path: Path) -> lis
|
||||
if errors:
|
||||
return errors
|
||||
|
||||
if risk.get("schema_version") != 1:
|
||||
errors.append("test risk map schema_version must be 1")
|
||||
project_root = _project_root(risk_path, quality_path)
|
||||
if risk.get("schema_version") != 2:
|
||||
errors.append("test risk map schema_version must be 2")
|
||||
if not isinstance(risk.get("owner"), str) or not risk["owner"].strip():
|
||||
errors.append("test risk map requires a non-empty owner")
|
||||
|
||||
inventory_sources = _inventory_sources(
|
||||
risk.get("source_inventory"),
|
||||
project_root=project_root,
|
||||
errors=errors,
|
||||
)
|
||||
source_contracts = _validate_contract_surfaces(
|
||||
risk.get("contract_surfaces"),
|
||||
inventory_sources=inventory_sources,
|
||||
project_root=project_root,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
gates = conditional.get("gates")
|
||||
gate_by_id = {
|
||||
gate.get("id"): gate
|
||||
@@ -62,30 +243,33 @@ def validate(risk_path: Path, quality_path: Path, conditional_path: Path) -> lis
|
||||
if path in module_by_path:
|
||||
errors.append(f"duplicate risk module path: {path}")
|
||||
module_by_path[path] = module
|
||||
if not Path(path).is_file():
|
||||
if not (project_root / path).is_file():
|
||||
errors.append(f"risk module maps missing source path: {path}")
|
||||
if module.get("risk_class") not in RISK_CLASSES:
|
||||
if path not in source_contracts:
|
||||
errors.append(f"risk module is missing a contract owner: {path}")
|
||||
risk_class = module.get("risk_class")
|
||||
if not isinstance(risk_class, str) or risk_class not in RISK_CLASSES:
|
||||
errors.append(f"risk module {path} has invalid risk_class")
|
||||
if not isinstance(module.get("risk"), str) or not module["risk"].strip():
|
||||
errors.append(f"risk module {path} requires a non-empty risk")
|
||||
|
||||
tests = module.get("required_tests")
|
||||
if not isinstance(tests, list) or not tests:
|
||||
errors.append(f"risk module {path} requires at least one test")
|
||||
else:
|
||||
for test_path in tests:
|
||||
if not isinstance(test_path, str) or not test_path.startswith("tests/"):
|
||||
errors.append(f"risk module {path} has invalid test path: {test_path!r}")
|
||||
elif not Path(test_path).is_file():
|
||||
errors.append(f"risk module {path} maps missing test path: {test_path}")
|
||||
_validate_test_paths(
|
||||
module.get("required_tests"),
|
||||
label=f"risk module {path}",
|
||||
project_root=project_root,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
module_gates = module.get("conditional_gates")
|
||||
if not isinstance(module_gates, list):
|
||||
errors.append(f"risk module {path} conditional_gates must be a list")
|
||||
continue
|
||||
if len(module_gates) != len(set(module_gates)):
|
||||
if len(module_gates) != len(set(map(str, module_gates))):
|
||||
errors.append(f"risk module {path} has duplicate conditional gates")
|
||||
for gate_id in module_gates:
|
||||
if not isinstance(gate_id, str):
|
||||
errors.append(f"risk module {path} has invalid conditional gate: {gate_id!r}")
|
||||
continue
|
||||
gate = gate_by_id.get(gate_id)
|
||||
if gate is None:
|
||||
errors.append(f"risk module {path} references unknown gate {gate_id}")
|
||||
|
||||
Reference in New Issue
Block a user