commit 9b08be5aa42c77599b40e534bafd6e71e9b91aeb
parent f7517ee7413e989db0dcfc31063ae6c505e1ea61
Author: triesap <tyson@radroots.org>
Date: Mon, 21 Sep 2026 03:41:28 +0000
spec: implement fixture structural validation
Diffstat:
2 files changed, 272 insertions(+), 0 deletions(-)
diff --git a/tests/fixture_validator.mojo b/tests/fixture_validator.mojo
@@ -0,0 +1,200 @@
+from std.collections import List
+
+from std.os.path import exists
+from std.pathlib import Path
+from json import Value, loads
+
+
+@fieldwise_init
+struct FixtureValidationIssue(Copyable, Movable):
+ var case_id: String
+ var rule: String
+ var detail: String
+
+
+def registered_fixture_operators() -> List[String]:
+ var operators = List[String]()
+ operators.append("equals")
+ operators.append("absent")
+ operators.append("present")
+ operators.append("contains")
+ operators.append("not_equals")
+ return operators^
+
+
+def _has_key(value: Value, key: String) -> Bool:
+ for candidate in value.object_keys():
+ if candidate == key:
+ return True
+ return False
+
+
+def _is_registered_operator(
+ operator: String, registered: List[String]
+) -> Bool:
+ for candidate in registered:
+ if candidate == operator:
+ return True
+ return False
+
+
+def validate_fixture_corpus(
+ directory: String,
+) raises -> List[FixtureValidationIssue]:
+ var issues = List[FixtureValidationIssue]()
+ var base = Path(directory)
+ var manifest = loads((base / "manifest.json").read_text())
+ if not _has_key(manifest, "cases"):
+ issues.append(
+ FixtureValidationIssue(
+ case_id="", rule="missing_cases", detail="manifest has no cases"
+ )
+ )
+ return issues^
+
+ var cases = manifest["cases"].array_items()
+ if len(cases) == 0:
+ issues.append(
+ FixtureValidationIssue(
+ case_id="",
+ rule="empty_corpus",
+ detail="manifest declares no cases",
+ )
+ )
+
+ var seen = List[String]()
+ var registered = registered_fixture_operators()
+ for entry in cases:
+ var case_id = entry["case_id"].string_value()
+
+ var duplicate = False
+ for prior in seen:
+ if prior == case_id:
+ duplicate = True
+ if duplicate:
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="duplicate_case_id",
+ detail="case_id appears more than once",
+ )
+ )
+ seen.append(String(case_id))
+
+ var relative_path = entry["path"].string_value()
+ var case_path = base / relative_path
+ if not exists(case_path):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="dangling_path",
+ detail=relative_path,
+ )
+ )
+ continue
+
+ var doc = loads(case_path.read_text())
+ if (
+ not _has_key(doc, "case_id")
+ or doc["case_id"].string_value() != case_id
+ ):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="case_id_mismatch",
+ detail="case_id does not match manifest",
+ )
+ )
+ if (
+ not _has_key(doc, "requirements")
+ or len(doc["requirements"].array_items()) == 0
+ ):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="empty_requirements",
+ detail="no requirement references",
+ )
+ )
+ if (
+ not _has_key(doc, "required_from_step")
+ or doc["required_from_step"].string_value() == ""
+ ):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="missing_activation_step",
+ detail="required_from_step is empty",
+ )
+ )
+ elif _has_key(entry, "required_from_step") and (
+ doc["required_from_step"].string_value()
+ != entry["required_from_step"].string_value()
+ ):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="activation_step_mismatch",
+ detail="required_from_step does not match manifest",
+ )
+ )
+ if not _has_key(doc, "then") or len(doc["then"].array_items()) == 0:
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="empty_expectations",
+ detail="no expected assertions",
+ )
+ )
+ else:
+ for assertion in doc["then"].array_items():
+ if not _has_key(assertion, "operator"):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="missing_operator",
+ detail="assertion has no operator",
+ )
+ )
+ continue
+ var operator = assertion["operator"].string_value()
+ if not _is_registered_operator(operator, registered):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="unknown_operator",
+ detail=operator,
+ )
+ )
+ if not _has_key(doc, "provenance"):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="missing_provenance",
+ detail="provenance is required",
+ )
+ )
+ if (
+ not _has_key(doc, "implementation_status")
+ or doc["implementation_status"].string_value() != "planned"
+ ):
+ issues.append(
+ FixtureValidationIssue(
+ case_id=case_id,
+ rule="unexpected_status",
+ detail="new corpus cases must be planned until activated",
+ )
+ )
+
+ if _has_key(manifest, "raw_files"):
+ for raw in manifest["raw_files"].array_items():
+ if not exists(base / raw.string_value()):
+ issues.append(
+ FixtureValidationIssue(
+ case_id="",
+ rule="dangling_raw",
+ detail=raw.string_value(),
+ )
+ )
+
+ return issues^
diff --git a/tests/test_hyf.mojo b/tests/test_hyf.mojo
@@ -26,6 +26,7 @@ from fixture_loader import (
load_fixture_scenario_request,
load_fixture_top_level_field_from_path,
)
+from fixture_validator import validate_fixture_corpus
from hyf_core.backends.selector import (
execute_capability as execute_core_capability,
resolve_backend,
@@ -1198,5 +1199,76 @@ def test_semantic_fixture_corpus_is_installed_and_planned() raises:
assert_true(exists(fixture_dir / raw.string_value()))
+def _write_min_fixture_corpus(base: Path, mutation: String) raises:
+ var domain = base / "domain"
+ std.os.makedirs(domain.__fspath__(), exist_ok=True)
+ var case_text = (
+ '{"fixture_format_version":1,"case_id":"T001",'
+ '"requirements":["HYF-TEST-001"],"implementation_status":"planned",'
+ '"required_from_step":"S009","mandatory":true,"given":{},'
+ '"provider_script":[],'
+ '"then":[{"operator":"equals","path":"/x","value":1}],'
+ '"provenance":{"kind":"synthetic"}}'
+ )
+ if mutation == "unknown_operator":
+ case_text = case_text.replace('"equals"', '"not_registered"')
+ elif mutation == "empty_then":
+ case_text = case_text.replace(
+ '[{"operator":"equals","path":"/x","value":1}]', "[]"
+ )
+ elif mutation == "missing_provenance":
+ case_text = case_text.replace(',"provenance":{"kind":"synthetic"}', "")
+ elif mutation == "empty_requirements":
+ case_text = case_text.replace('["HYF-TEST-001"]', "[]")
+ (domain / "T001.json").write_text(case_text)
+
+ var case_entry = (
+ '{"case_id":"T001","path":"domain/T001.json",'
+ '"required_from_step":"S009","mandatory":true}'
+ )
+ if mutation == "duplicate_case_id":
+ case_entry = case_entry + "," + case_entry
+ elif mutation == "dangling_path":
+ case_entry = (
+ '{"case_id":"T001","path":"domain/missing.json",'
+ '"required_from_step":"S009","mandatory":true}'
+ )
+ elif mutation == "activation_step_mismatch":
+ case_entry = (
+ '{"case_id":"T001","path":"domain/T001.json",'
+ '"required_from_step":"S999","mandatory":true}'
+ )
+ var manifest = (
+ '{"schema_version":1,"spec_id":"hyf_v1_jev",'
+ '"installation_status":"installed","cases":['
+ + case_entry
+ + '],"raw_files":[]}'
+ )
+ (base / "manifest.json").write_text(manifest)
+
+
+def test_fixture_validator_accepts_corpus_and_rejects_corruptions() raises:
+ var corpus_dir = _dir_of_current_file() / "fixtures" / "hyf_v1_jev"
+ assert_equal(
+ len(validate_fixture_corpus(corpus_dir.__fspath__())), 0
+ )
+ var mutations = List[String]()
+ mutations.append("duplicate_case_id")
+ mutations.append("dangling_path")
+ mutations.append("unknown_operator")
+ mutations.append("empty_then")
+ mutations.append("missing_provenance")
+ mutations.append("empty_requirements")
+ mutations.append("activation_step_mismatch")
+ for mutation in mutations:
+ with TemporaryDirectory() as temp_dir:
+ var base = Path(temp_dir)
+ _write_min_fixture_corpus(base, mutation)
+ assert_true(
+ len(validate_fixture_corpus(base.__fspath__())) > 0,
+ "validator accepted corruption: " + mutation,
+ )
+
+
def main() raises:
TestSuite.discover_tests[__functions_in_module()]().run()
