#!/usr/bin/env python3 """Fixture tests for nAvid batch queue coordination.""" from __future__ import annotations import json from pathlib import Path import subprocess import sys import tempfile import unittest BATCH_SCRIPT = Path(__file__).with_name("batch_queue.py") RUN_SCRIPT = Path(__file__).with_name("run_state.py") STAGES = ("source", "script", "voice", "composition", "render", "qa") VIETNAMESE = "Ti\u1ebfng Vi\u1ec7t" class BatchQueueFoundationTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.queue_dir = self.root / "videos" / "_queues" / "news-morning" self.project_one = self.root / "videos" / "repo-alpha" self.project_two = self.root / "videos" / "repo-beta" self.project_three = self.root / "videos" / "repo-gamma" self.project_one.mkdir(parents=True) self.project_two.mkdir(parents=True) self.project_three.mkdir(parents=True) def tearDown(self) -> None: self.temp.cleanup() def run_batch(self, *args: str) -> dict: result = subprocess.run( [sys.executable, "-X", "utf8", str(BATCH_SCRIPT), *args], check=True, capture_output=True, text=True, encoding="utf-8", ) return json.loads(result.stdout) def run_state(self, *args: str) -> dict: result = subprocess.run( [sys.executable, "-X", "utf8", str(RUN_SCRIPT), *args], check=True, capture_output=True, text=True, encoding="utf-8", ) return json.loads(result.stdout) def init_queue(self) -> dict: return self.run_batch( "init", "--queue-dir", str(self.queue_dir), "--batch-id", "news-morning", "--item", f"https://github.com/a/alpha::{self.project_one}", "--item", f"https://github.com/a/alpha::{self.project_one}", "--item", f"https://github.com/b/beta::{self.project_two}", "--item", f"https://github.com/c/gamma::{self.project_three}", "--profile", "ai-news-247", "--language", VIETNAMESE, "--template-strategy", "use-existing", "--template-set", "repo-update-fullframe", "--mode", "auto-run", "--reporting", "quiet", ) def test_init_preserves_order_shared_intake_and_state_links(self) -> None: queue = self.init_queue() self.assertEqual(queue["batch"]["id"], "news-morning") self.assertEqual(queue["intake"]["profile"], "ai-news-247") self.assertEqual(queue["intake"]["language"], VIETNAMESE) self.assertEqual(queue["intake"]["template_set"], "repo-update-fullframe") self.assertEqual( [item["source_url"] for item in queue["items"]], ["https://github.com/a/alpha", "https://github.com/b/beta", "https://github.com/c/gamma"], ) self.assertEqual(queue["items"][0]["state_path"], str(self.project_one.resolve() / "run-state.json")) self.assertNotIn("stages", queue["items"][0]) self.assertTrue(any(event["type"] == "exact_duplicates_suppressed" for event in queue["events"])) def test_item_status_update_does_not_change_sibling(self) -> None: self.init_queue() updated = self.run_batch( "set-status", "--queue-dir", str(self.queue_dir), "--item-id", "item-001", "--status", "failed", "--error", "render timed out after bounded retries", ) self.assertEqual(updated["item"]["status"], "failed") inspected = self.run_batch("inspect", "--queue-dir", str(self.queue_dir)) self.assertEqual(inspected["items"][0]["status"], "failed") self.assertEqual(inspected["items"][1]["status"], "pending") def init_state(self, project: Path) -> None: self.run_state( "init", "--project-dir", str(project), "--profile", "ai-news-247", "--language", VIETNAMESE, "--template-strategy", "use-existing", "--template-set", "repo-update-fullframe", ) def artifact(self, project: Path, name: str, text: str = "ok") -> str: path = 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.root / "registry.json" path.write_text(json.dumps({ "sets": [ {"id": "repo-update-fullframe", "tier": "approved", "domain": "developer-tools"} ] }), encoding="utf-8") return path def complete_state(self, project: Path) -> None: self.init_state(project) self.artifact(project, "PUBLISH.md", "## Publish Title\nRepo update\n\n## Post Caption\nTin công nghệ.\n\n## Hashtags\n#AINews247 #OpenSource #DeveloperTools #AIAgents #RAG\n") files = { "source": self.artifact(project, "capture/source.json"), "script": self.artifact(project, "SCRIPT.md"), "voice": self.artifact(project, "narration.wav"), "composition": self.artifact(project, "index.html"), "render": self.artifact(project, "renders/final.mp4"), "qa": self.artifact(project, "capture/qa.json"), } previous = None for stage in STAGES: args = ["complete-stage", "--project-dir", str(project), "--stage", stage, "--artifact", files[stage]] if previous: args += ["--depends", previous] if stage == "voice": args += ["--meta", "provider=OmniVoice", "--meta", "profile=main"] if stage == "qa": args += ["--meta", "result=pass"] self.run_state(*args) previous = files[stage] self.run_state( "select-template", "--project-dir", str(project), "--registry", str(self.registry()), "--template-id", "repo-update-fullframe", "--rationale", "Shared approved batch proof layout.", ) def test_resume_skips_valid_complete_item_and_selects_next(self) -> None: self.init_queue() self.complete_state(self.project_one) result = self.run_batch("next-item", "--queue-dir", str(self.queue_dir)) self.assertEqual(result["items"][0]["status"], "complete") self.assertEqual(result["item"]["id"], "item-002") def test_changed_complete_artifact_resumes_only_its_item(self) -> None: self.init_queue() self.complete_state(self.project_one) self.complete_state(self.project_two) (self.project_one / "narration.wav").write_text("changed", encoding="utf-8") result = self.run_batch("next-item", "--queue-dir", str(self.queue_dir)) self.assertEqual(result["item"]["id"], "item-001") self.assertEqual(result["item"]["resume_stage"], "voice") synced_second = self.run_batch( "sync-item", "--queue-dir", str(self.queue_dir), "--item-id", "item-002" ) self.assertEqual(synced_second["item"]["status"], "complete") def test_failed_and_blocked_items_do_not_prevent_later_selection(self) -> None: self.init_queue() self.init_state(self.project_one) for number in range(3): self.run_state( "record-retry", "--project-dir", str(self.project_one), "--stage", "voice", "--error", f"timeout-{number}", "--same-output", ) failed = self.run_batch("next-item", "--queue-dir", str(self.queue_dir)) self.assertEqual(failed["items"][0]["status"], "failed") self.assertEqual(failed["item"]["id"], "item-002") self.init_state(self.project_two) self.run_state( "block-fallback", "--project-dir", str(self.project_two), "--stage", "voice", "--original", "OmniVoice", "--proposal", "Edge TTS", "--reason", "preferred voice unavailable", "--impact", "voice changes", ) blocked = self.run_batch("next-item", "--queue-dir", str(self.queue_dir)) self.assertEqual(blocked["items"][1]["status"], "blocked") self.assertEqual(blocked["item"]["id"], "item-003") def test_selected_retry_only_reopens_requested_failed_item(self) -> None: self.init_queue() self.run_batch("set-status", "--queue-dir", str(self.queue_dir), "--item-id", "item-001", "--status", "failed", "--error", "timeout") result = self.run_batch("retry-item", "--queue-dir", str(self.queue_dir), "--item-id", "item-001") self.assertEqual(result["item"]["status"], "pending") inspected = self.run_batch("inspect", "--queue-dir", str(self.queue_dir)) self.assertEqual(inspected["items"][1]["status"], "pending") def test_summary_discloses_ready_blocked_and_failed_items(self) -> None: self.init_queue() self.complete_state(self.project_one) self.init_state(self.project_two) self.run_state( "block-fallback", "--project-dir", str(self.project_two), "--stage", "voice", "--original", "OmniVoice", "--proposal", "Edge TTS", "--reason", "approval needed", "--impact", "voice changes", ) self.init_state(self.project_three) for number in range(3): self.run_state( "record-retry", "--project-dir", str(self.project_three), "--stage", "render", "--error", f"render-timeout-{number}", "--same-output", ) report = self.run_batch("summary", "--queue-dir", str(self.queue_dir), "--refresh") text = (self.queue_dir / "SUMMARY.md").read_text(encoding="utf-8") self.assertEqual(report["ready_to_publish"], ["item-001"]) self.assertIn("## Ready to publish", text) self.assertIn("## Blocked", text) self.assertIn("## Failed", text) self.assertIn("repo-update-fullframe (approved)", text) self.assertIn("OmniVoice", text) self.assertIn("approval needed", text) self.assertIn("Publish Title: Repo update", text) self.assertIn("Hashtags: #AINews247 #OpenSource #DeveloperTools #AIAgents #RAG", text) def test_ready_item_requires_publish_package(self) -> None: self.init_queue() self.complete_state(self.project_one) (self.project_one / "PUBLISH.md").unlink() report = self.run_batch("summary", "--queue-dir", str(self.queue_dir), "--refresh") self.assertNotIn("item-001", report["ready_to_publish"]) def test_candidate_template_never_appears_publish_ready(self) -> None: self.init_queue() self.complete_state(self.project_one) state_path = self.project_one / "run-state.json" state = json.loads(state_path.read_text(encoding="utf-8")) state["template_selection"]["tier"] = "candidate" state_path.write_text(json.dumps(state), encoding="utf-8") report = self.run_batch("summary", "--queue-dir", str(self.queue_dir), "--refresh") self.assertNotIn("item-001", report["ready_to_publish"]) self.assertEqual(report["items"][0]["status"], "blocked") if __name__ == "__main__": unittest.main(verbosity=2)