#!/usr/bin/env python3 """Fixture tests for nAvid run state transitions.""" from __future__ import annotations import json from pathlib import Path import subprocess import sys import tempfile import unittest SCRIPT = Path(__file__).with_name("run_state.py") STAGES = ("source", "script", "voice", "composition", "render", "qa") VIETNAMESE = "Ti\u1ebfng Vi\u1ec7t" class RunStateTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.project = Path(self.temp.name) / "demo-video" self.project.mkdir() self.run_cli( "init", "--project-dir", str(self.project), "--profile", "ai-news-247", "--language", VIETNAMESE, "--template-strategy", "use-existing", "--template-set", "clean-tech", "--mode", "review-first", "--reporting", "quiet", ) def tearDown(self) -> None: self.temp.cleanup() def run_cli(self, *args: str) -> dict: result = subprocess.run( [sys.executable, "-X", "utf8", str(SCRIPT), *args], check=True, capture_output=True, text=True, encoding="utf-8", ) return json.loads(result.stdout) def artifact(self, name: str, text: str = "ok") -> str: path = self.project / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return name def registry(self) -> Path: path = self.project / "template-registry.json" path.write_text(json.dumps({ "sets": [ {"id": "repo-update-fullframe", "tier": "approved", "domain": "developer-tools"}, {"id": "developer-tool", "tier": "approved", "domain": "developer-tools"}, {"id": "candidate-only", "tier": "candidate", "domain": "developer-tools"} ] }), encoding="utf-8") return path def complete_all(self) -> None: files = { "source": self.artifact("capture/source.json"), "script": self.artifact("SCRIPT.md"), "voice": self.artifact("narration.wav"), "composition": self.artifact("index.html"), "render": self.artifact("renders/demo.mp4"), "qa": self.artifact("capture/qa.json"), } previous = None for stage in STAGES: args = [ "complete-stage", "--project-dir", str(self.project), "--stage", stage, "--artifact", files[stage], ] if previous: args += ["--depends", previous] if stage == "qa": args += ["--meta", "result=pass"] self.run_cli(*args) previous = files[stage] def test_init_preserves_intake_and_modes(self) -> None: state = self.run_cli("inspect", "--project-dir", str(self.project)) self.assertEqual(state["intake"]["profile"], "ai-news-247") self.assertEqual(state["intake"]["language"], VIETNAMESE) self.assertEqual(state["mode"], {"execution": "review-first", "reporting": "quiet"}) self.assertEqual(list(state["stages"]), list(STAGES)) self.assertEqual(state["resume"]["stage"], "source") def test_all_stages_complete_and_qa_tracks_render(self) -> None: self.complete_all() result = self.run_cli("resume-check", "--project-dir", str(self.project)) self.assertEqual(result["resume"]["status"], "complete") self.assertEqual(result["stages"]["qa"]["metadata"]["result"], "pass") self.assertTrue(result["stages"]["qa"]["dependencies"]) def test_missing_artifact_invalidates_earliest_affected_stage(self) -> None: self.complete_all() (self.project / "narration.wav").unlink() result = self.run_cli("resume-check", "--project-dir", str(self.project)) self.assertEqual(result["resume"], {"stage": "voice", "status": "invalid"}) self.assertEqual(result["stages"]["render"]["status"], "invalid") def test_change_cascade_boundaries(self) -> None: cases = ( ("source", "source"), ("narration", "voice"), ("voice", "voice"), ("template", "composition"), ) for change, expected in cases: with self.subTest(change=change): project = Path(self.temp.name) / f"cascade-{change}" project.mkdir() old = self.project try: self.project = project self.run_cli( "init", "--project-dir", str(project), "--profile", "ai-news-247", "--language", VIETNAMESE, "--template-strategy", "new", ) self.complete_all() result = self.run_cli( "invalidate", "--project-dir", str(project), "--change", change ) self.assertEqual(result["resume"]["stage"], expected) self.assertEqual(result["events"][-1]["type"], "stages_invalidated") finally: self.project = old def test_retry_exhaustion_is_failed(self) -> None: for number in range(3): result = self.run_cli( "record-retry", "--project-dir", str(self.project), "--stage", "voice", "--error", f"timeout-{number}", "--same-output", ) self.assertEqual(result["attempt_count"], 3) self.assertEqual(result["status"], "failed") self.assertFalse(result["can_retry"]) def test_fallback_is_blocked_then_run_scoped_approval_invalidates(self) -> None: blocked = self.run_cli( "block-fallback", "--project-dir", str(self.project), "--stage", "voice", "--original", "OmniVoice", "--proposal", "Edge TTS", "--reason", "preferred provider unavailable", "--impact", "voice quality changes", ) self.assertEqual(blocked["status"], "blocked") approved = self.run_cli( "approve-fallback", "--project-dir", str(self.project), "--stage", "voice", "--decision", "use Edge TTS for this run", "--scope", "voice through QA", ) self.assertEqual(approved["status"], "invalid") state = self.run_cli("inspect", "--project-dir", str(self.project)) self.assertTrue(any(event["type"] == "fallback_approved" for event in state["events"])) def test_selected_approved_template_persists_with_rationale(self) -> None: result = self.run_cli( "select-template", "--project-dir", str(self.project), "--registry", str(self.registry()), "--template-id", "repo-update-fullframe", "--rationale", "Matches GitHub changelog proof and readable crop treatment.", ) self.assertEqual("repo-update-fullframe", result["template_selection"]["id"]) state = self.run_cli("inspect", "--project-dir", str(self.project)) self.assertEqual("approved", state["template_selection"]["tier"]) self.assertIn("GitHub changelog", state["template_selection"]["rationale"]) def test_candidate_selection_is_rejected(self) -> None: result = subprocess.run( [sys.executable, "-X", "utf8", str(SCRIPT), "select-template", "--project-dir", str(self.project), "--registry", str(self.registry()), "--template-id", "candidate-only", "--rationale", "Not allowed."], check=False, capture_output=True, text=True, encoding="utf-8", ) self.assertNotEqual(0, result.returncode) def test_changing_selected_template_invalidates_composition_downstream(self) -> None: self.complete_all() registry = self.registry() self.run_cli( "select-template", "--project-dir", str(self.project), "--registry", str(registry), "--template-id", "repo-update-fullframe", "--rationale", "Initial proof fit.", ) changed = self.run_cli( "select-template", "--project-dir", str(self.project), "--registry", str(registry), "--template-id", "developer-tool", "--rationale", "Different approved layout requested.", ) self.assertEqual(["composition", "render", "qa"], changed["invalidated"]) state = self.run_cli("inspect", "--project-dir", str(self.project)) self.assertEqual("complete", state["stages"]["voice"]["status"]) self.assertEqual("invalid", state["stages"]["composition"]["status"]) if __name__ == "__main__": unittest.main(verbosity=2)