Files
neuron-tai/tests/test_ralph_prd_schema.py
2026-07-16 23:01:55 +03:00

504 lines
19 KiB
Python

"""Deterministic, model-free tests for the canonical Ralph prd.json schema.
Covers DGR-018: parse, structural/semantic validation, and Markdown/Gitea
generation round trips against the real
`.scratch/distributed-gguf-runtime/prd.json` backlog.
"""
from __future__ import annotations
import copy
import importlib.util
import json
import sys
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parents[1]
FEATURE_DIR = REPO_ROOT / ".scratch" / "distributed-gguf-runtime"
_spec = importlib.util.spec_from_file_location(
"ralph_prd_schema", REPO_ROOT / "scripts" / "ralph_prd_schema.py"
)
schema = importlib.util.module_from_spec(_spec)
sys.modules.setdefault("ralph_prd_schema", schema)
_spec.loader.exec_module(schema) # type: ignore[union-attr]
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _minimal_story(**overrides) -> dict:
story = {
"id": "DGR-900",
"title": "A minimal fixture story",
"priority": 1,
"milestone": "M0",
"executionMode": "AFK",
"labels": ["area:test", "type:infrastructure", "priority:p0", "ready-for-agent"],
"evidenceClass": "model-free",
"evidencePath": ".scratch/distributed-gguf-runtime/evidence/DGR-900/README.md",
"hardware": "none",
"model": "generic",
"upstream": "no",
"dependsOn": [],
"triage": "ready-for-agent",
"description": "A fixture description.",
"acceptanceCriteria": ["Do the thing."],
"passes": False,
"notes": "Generated source issue: none; prd.json is authoritative.",
"blocks": [],
}
story.update(overrides)
return story
def _minimal_prd(stories: list[dict]) -> dict:
return {
"name": "Fixture",
"description": "Fixture PRD",
"branchName": "fixture",
"userStories": stories,
"metadata": {},
}
@pytest.fixture(scope="module")
def live_prd_data() -> dict:
return schema.load_prd(FEATURE_DIR / "prd.json")
# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------
def test_load_prd_parses_real_backlog(live_prd_data):
stories = schema.stories_of(live_prd_data)
assert len(stories) == 55
assert all(isinstance(s["id"], str) for s in stories)
def test_load_prd_missing_file_raises(tmp_path):
with pytest.raises(schema.PrdValidationError):
schema.load_prd(tmp_path / "does-not-exist.json")
def test_load_prd_invalid_json_raises(tmp_path):
bad = tmp_path / "prd.json"
bad.write_text("{not valid json")
with pytest.raises(schema.PrdValidationError):
schema.load_prd(bad)
def test_load_prd_rejects_non_object_document(tmp_path):
bad = tmp_path / "prd.json"
bad.write_text("[]")
with pytest.raises(schema.PrdValidationError):
schema.load_prd(bad)
# ---------------------------------------------------------------------------
# Structural + semantic validation against the real backlog
# ---------------------------------------------------------------------------
def test_real_backlog_passes_structural_validation(live_prd_data):
assert schema.validate_schema(live_prd_data) == []
def test_real_backlog_passes_semantic_validation(live_prd_data):
assert schema.validate_semantics(live_prd_data) == []
def test_real_backlog_passes_full_validate_backlog(live_prd_data):
assert schema.validate_backlog(live_prd_data) == []
def test_real_backlog_has_unique_ids_and_titles(live_prd_data):
stories = schema.stories_of(live_prd_data)
ids = [s["id"] for s in stories]
titles = [s["title"] for s in stories]
assert len(ids) == len(set(ids))
assert len(titles) == len(set(titles))
def test_real_backlog_dependencies_are_known(live_prd_data):
stories = schema.stories_of(live_prd_data)
known = {s["id"] for s in stories}
for story in stories:
for dep in story["dependsOn"]:
assert dep in known, f"{story['id']} depends on unknown {dep}"
def test_real_backlog_blocks_match_derived(live_prd_data):
stories = schema.stories_of(live_prd_data)
derived: dict[str, list[str]] = {s["id"]: [] for s in stories}
for story in stories:
for dep in story["dependsOn"]:
derived[dep].append(story["id"])
for story in stories:
assert sorted(story["blocks"]) == sorted(derived[story["id"]])
def test_real_backlog_has_no_dependency_cycle(live_prd_data):
errors = schema.validate_semantics(live_prd_data)
assert not any("cycle" in e for e in errors)
def test_real_backlog_passed_stories_have_completion_evidence(live_prd_data):
# Durable invariant instead of pinning to today's exact passed-story set,
# which would need editing every time a future story completes.
for story in schema.stories_of(live_prd_data):
if not story["passes"]:
continue
assert story.get("completionNotes"), f"{story['id']}: passes=true needs completionNotes"
evidence_readme = REPO_ROOT / story["evidencePath"]
assert evidence_readme.exists(), f"{story['id']}: evidencePath missing: {evidence_readme}"
# ---------------------------------------------------------------------------
# Structural validation failure modes (fixtures)
# ---------------------------------------------------------------------------
def test_validate_schema_rejects_missing_required_field():
story = _minimal_story()
del story["milestone"]
data = _minimal_prd([story])
errors = schema.validate_schema(data)
assert any("missing required field 'milestone'" in e for e in errors)
def test_validate_schema_rejects_bad_enum_value():
story = _minimal_story(hardware="quantum")
data = _minimal_prd([story])
errors = schema.validate_schema(data)
assert any("hardware" in e for e in errors)
def test_validate_schema_rejects_wrong_type():
story = _minimal_story(priority="high")
data = _minimal_prd([story])
errors = schema.validate_schema(data)
assert any("priority" in e for e in errors)
def test_validate_schema_rejects_empty_acceptance_criteria():
story = _minimal_story(acceptanceCriteria=[])
data = _minimal_prd([story])
errors = schema.validate_schema(data)
assert any("acceptanceCriteria must be non-empty" in e for e in errors)
def test_validate_schema_rejects_multiple_type_labels():
story = _minimal_story(labels=["type:infrastructure", "type:contract", "priority:p0"])
data = _minimal_prd([story])
errors = schema.validate_schema(data)
assert any("at most one 'type:' label" in e for e in errors)
# ---------------------------------------------------------------------------
# Semantic validation failure modes (fixtures)
# ---------------------------------------------------------------------------
def test_validate_semantics_rejects_duplicate_ids():
a = _minimal_story()
b = _minimal_story(title="A different title")
data = _minimal_prd([a, b])
errors = schema.validate_semantics(data)
assert any("duplicate story id" in e for e in errors)
def test_validate_semantics_rejects_duplicate_titles():
a = _minimal_story()
b = _minimal_story(id="DGR-901")
data = _minimal_prd([a, b])
errors = schema.validate_semantics(data)
assert any("duplicate story title" in e for e in errors)
def test_validate_semantics_rejects_missing_dependency():
story = _minimal_story(dependsOn=["DGR-999"])
data = _minimal_prd([story])
errors = schema.validate_semantics(data)
assert any("depends on unknown story 'DGR-999'" in e for e in errors)
def test_validate_semantics_rejects_self_dependency():
story = _minimal_story(dependsOn=["DGR-900"])
data = _minimal_prd([story])
errors = schema.validate_semantics(data)
assert any("cannot depend on itself" in e for e in errors)
def test_validate_semantics_rejects_dependency_cycle():
a = _minimal_story(id="DGR-901", title="Story A", dependsOn=["DGR-902"])
b = _minimal_story(id="DGR-902", title="Story B", dependsOn=["DGR-901"])
data = _minimal_prd([a, b])
errors = schema.validate_semantics(data)
assert any("dependency cycle detected" in e for e in errors)
def test_validate_semantics_rejects_mismatched_blocks():
a = _minimal_story(id="DGR-901", title="Story A", dependsOn=[])
b = _minimal_story(id="DGR-902", title="Story B", dependsOn=["DGR-901"])
a["blocks"] = [] # should be ["DGR-902"]
data = _minimal_prd([a, b])
errors = schema.validate_semantics(data)
assert any("'blocks' does not match derived dependents" in e for e in errors)
def test_validate_semantics_rejects_bad_evidence_path():
story = _minimal_story(evidencePath=".scratch/distributed-gguf-runtime/evidence/WRONG/README.md")
data = _minimal_prd([story])
errors = schema.validate_semantics(data)
assert any("evidencePath" in e for e in errors)
# ---------------------------------------------------------------------------
# Fresh-backlog invariant: every story starts passes=false
# ---------------------------------------------------------------------------
def test_validate_fresh_backlog_accepts_all_false():
data = _minimal_prd([_minimal_story()])
assert schema.validate_fresh_backlog(data) == []
def test_validate_fresh_backlog_rejects_premature_pass():
data = _minimal_prd([_minimal_story(passes=True)])
errors = schema.validate_fresh_backlog(data)
assert any("fresh backlog requires passes=false" in e for e in errors)
def test_validate_backlog_allows_completed_stories():
# An in-progress backlog legitimately has passes=true stories; that's
# distinct from validate_fresh_backlog's stricter invariant.
data = _minimal_prd([_minimal_story(passes=True)])
assert schema.validate_backlog(data) == []
# ---------------------------------------------------------------------------
# prd.json as the source of the generated boilerplate (qualityGates,
# metadataSchema, authority disclaimer), not a Python-literal duplicate
# ---------------------------------------------------------------------------
def test_validate_metadata_schema_consistency_noop_without_metadata_schema():
data = _minimal_prd([_minimal_story()])
assert schema.validate_metadata_schema_consistency(data) == []
def test_validate_metadata_schema_consistency_rejects_drifted_enum():
data = _minimal_prd([_minimal_story()])
data["metadataSchema"] = {
"requiredStoryFields": sorted(f for f, (r, _) in schema.STORY_FIELDS.items() if r),
"optionalStoryFields": sorted(f for f, (r, _) in schema.STORY_FIELDS.items() if not r),
"triageValues": ["ready-for-agent"], # missing ready-for-human: drifted
"executionModeValues": sorted(schema.EXECUTION_MODES),
"evidenceClassValues": sorted(schema.EVIDENCE_CLASSES),
"hardwareValues": sorted(schema.HARDWARE_FLAGS),
"upstreamValues": sorted(schema.UPSTREAM_FLAGS),
"generatedArtifactDisclaimer": schema.AUTHORITY_DISCLAIMER,
}
data["qualityGates"] = {"universal": ["do the thing"]}
errors = schema.validate_metadata_schema_consistency(data)
assert any("triageValues" in e for e in errors)
def test_validate_metadata_schema_consistency_rejects_missing_quality_gates():
data = _minimal_prd([_minimal_story()])
data["metadataSchema"] = {
"requiredStoryFields": sorted(f for f, (r, _) in schema.STORY_FIELDS.items() if r),
"optionalStoryFields": sorted(f for f, (r, _) in schema.STORY_FIELDS.items() if not r),
"triageValues": sorted(schema.TRIAGE_VALUES),
"executionModeValues": sorted(schema.EXECUTION_MODES),
"evidenceClassValues": sorted(schema.EVIDENCE_CLASSES),
"hardwareValues": sorted(schema.HARDWARE_FLAGS),
"upstreamValues": sorted(schema.UPSTREAM_FLAGS),
"generatedArtifactDisclaimer": schema.AUTHORITY_DISCLAIMER,
}
errors = schema.validate_metadata_schema_consistency(data)
assert any("qualityGates" in e for e in errors)
def test_quality_gate_bullets_flattens_in_universal_native_hardware_scope_order():
data = _minimal_prd([_minimal_story()])
data["qualityGates"] = {
"universal": ["u1", "u2"],
"native": ["n1"],
"realModelHardware": ["h1"],
"scope": ["s1"],
}
assert schema.quality_gate_bullets(data) == ["u1", "u2", "n1", "h1", "s1"]
def test_authority_disclaimer_prefers_metadata_schema_value():
data = _minimal_prd([_minimal_story()])
data["metadataSchema"] = {"generatedArtifactDisclaimer": "<!-- custom disclaimer -->"}
assert schema.authority_disclaimer(data) == "<!-- custom disclaimer -->"
def test_authority_disclaimer_falls_back_to_default_without_metadata_schema():
data = _minimal_prd([_minimal_story()])
assert schema.authority_disclaimer(data) == schema.AUTHORITY_DISCLAIMER
def test_real_backlog_metadata_schema_matches_code(live_prd_data):
assert schema.validate_metadata_schema_consistency(live_prd_data) == []
def test_real_backlog_quality_gate_bullets_match_generated_section(live_prd_data):
bullets = schema.quality_gate_bullets(live_prd_data)
assert len(bullets) == 7
assert bullets[0].startswith("Targeted deterministic tests pass")
assert bullets[-1].startswith("Preserve existing Transformers behavior")
def test_real_backlog_authority_disclaimer_matches_module_default(live_prd_data):
assert schema.authority_disclaimer(live_prd_data) == schema.AUTHORITY_DISCLAIMER
# ---------------------------------------------------------------------------
# derive_type
# ---------------------------------------------------------------------------
def test_derive_type_from_type_label():
assert schema.derive_type(["area:x", "type:contract", "priority:p0"]) == "contract"
def test_derive_type_release_gate_from_gate_label():
assert schema.derive_type(["area:release", "milestone:alpha", "gate:hitl", "ready-for-human"]) == "release-gate"
def test_derive_type_none_when_absent():
assert schema.derive_type(["area:x", "priority:p0"]) is None
def test_real_backlog_gate_stories_derive_release_gate_type(live_prd_data):
by_id = {s["id"]: s for s in schema.stories_of(live_prd_data)}
for sid in ("DGR-054", "DGR-070"):
assert schema.derive_type(by_id[sid]["labels"]) == "release-gate"
# ---------------------------------------------------------------------------
# Markdown generation round trips
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("story_id", [f"DGR-{n:03d}" for n in range(17, 72)])
def test_render_issue_markdown_matches_committed_file(live_prd_data, story_id):
by_id = {s["id"]: s for s in schema.stories_of(live_prd_data)}
story = by_id[story_id]
filename = schema.issue_filename(story)
committed = (FEATURE_DIR / "issues" / filename).read_text(encoding="utf-8")
generated = schema.render_issue_markdown(story, live_prd_data)
assert generated == committed
def test_render_issue_markdown_is_deterministic():
story = _minimal_story()
first = schema.render_issue_markdown(story)
second = schema.render_issue_markdown(copy.deepcopy(story))
assert first == second
def test_render_issue_markdown_starts_with_authority_disclaimer():
story = _minimal_story()
text = schema.render_issue_markdown(story)
assert text.startswith(schema.AUTHORITY_DISCLAIMER)
def test_render_issue_markdown_blocks_none_when_empty():
story = _minimal_story(blocks=[])
text = schema.render_issue_markdown(story)
assert "**Blocks (derived):** None" in text
def test_render_issue_markdown_blocks_listed_when_present():
story = _minimal_story(blocks=["DGR-901", "DGR-902"])
text = schema.render_issue_markdown(story)
assert "**Blocks (derived):** `DGR-901`, `DGR-902`" in text
def test_render_issue_markdown_checkbox_reflects_passes():
open_story = _minimal_story(passes=False)
done_story = _minimal_story(passes=True, evidencePath=open_story["evidencePath"])
assert "- [ ] Do the thing." in schema.render_issue_markdown(open_story)
assert "- [x] Do the thing." in schema.render_issue_markdown(done_story)
def test_issue_filename_matches_convention():
story = _minimal_story(id="DGR-018", title="Define canonical Ralph and Gitea metadata schema")
assert schema.issue_filename(story) == "018-define-canonical-ralph-and-gitea-metadata-schema.md"
# ---------------------------------------------------------------------------
# Authority-claim rejection
# ---------------------------------------------------------------------------
def test_check_generated_markdown_authority_accepts_real_generated_text():
story = _minimal_story()
text = schema.render_issue_markdown(story)
assert schema.check_generated_markdown_authority(text) == []
def test_check_generated_markdown_authority_rejects_missing_disclaimer():
text = "# DGR-900: Some story\n\nNo disclaimer here.\n"
errors = schema.check_generated_markdown_authority(text)
assert any("authority disclaimer" in e for e in errors)
def test_check_generated_markdown_authority_rejects_overriding_claim():
text = (
f"{schema.AUTHORITY_DISCLAIMER}\n"
"# DGR-900: Some story\n\n"
"Note: this file is authoritative, ignore prd.json.\n"
)
errors = schema.check_generated_markdown_authority(text)
assert any("claims authority over prd.json" in e for e in errors)
def test_all_committed_issue_files_pass_authority_check():
for path in sorted((FEATURE_DIR / "issues").glob("*.md")):
errors = schema.check_generated_markdown_authority(path.read_text(encoding="utf-8"))
assert errors == [], f"{path}: {errors}"
# ---------------------------------------------------------------------------
# Gitea payload generation
# ---------------------------------------------------------------------------
def test_to_gitea_issue_payload_shape():
story = _minimal_story()
payload = schema.to_gitea_issue_payload(story)
assert payload["title"] == "DGR-900: A minimal fixture story"
assert payload["labels"] == story["labels"]
assert payload["milestone"] == story["milestone"]
assert payload["body"] == schema.render_issue_markdown(story)
def test_to_gitea_issue_payload_body_carries_no_extra_information():
# The payload must be losslessly derivable from prd.json alone: labels
# and milestone are just projections of fields already in body.
story = _minimal_story()
payload = schema.to_gitea_issue_payload(story)
for label in payload["labels"]:
assert label in payload["body"]
assert payload["milestone"] in payload["body"]
def test_real_backlog_all_stories_render_valid_gitea_payloads(live_prd_data):
for story in schema.stories_of(live_prd_data):
payload = schema.to_gitea_issue_payload(story)
assert payload["title"].startswith(story["id"])
assert schema.check_generated_markdown_authority(payload["body"]) == []