Initial release — OpenMontage: the first open-source agentic video production system

11 production pipelines, 47 tools, 124 agent skills.
Supports cloud APIs (fal.ai, OpenAI, ElevenLabs, Suno, HeyGen, Runway) and
free local providers (diffusers, Piper TTS, WAN 2.1, Hunyuan, CogVideo).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
calesthio
2026-03-29 08:25:17 -07:00
commit a3e735cc7a
1147 changed files with 240221 additions and 0 deletions
View File
View File
+493
View File
@@ -0,0 +1,493 @@
"""Phase 0 contract tests — infrastructure layer.
Tests config, schemas, checkpoints, pipeline manifests, tools, cost tracker,
and media profiles. The intelligence layer (orchestrator, reviewer, checkpoint
policy, handlers) has been replaced by instruction-driven architecture:
pipeline manifests + stage director skills + meta skills.
"""
import importlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
import pytest
# Add project root to path
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from lib.config_model import OpenMontageConfig
from lib.checkpoint import (
CheckpointValidationError,
STAGES,
get_next_stage,
read_checkpoint,
write_checkpoint,
)
from lib.media_profiles import get_profile, ffmpeg_output_args, ALL_PROFILES
from lib.pipeline_loader import (
get_required_tools,
get_stage_order,
get_stage_skill,
get_stage_review_focus,
list_pipelines,
load_pipeline,
)
from tools.base_tool import BaseTool, ToolResult, ToolTier, ToolStatus
from tools.tool_registry import ToolRegistry
from tools.cost_tracker import CostTracker, BudgetMode, BudgetExceededError, ApprovalRequiredError
from schemas.artifacts import load_schema, validate_artifact, list_schemas
def sample_artifact(name: str) -> dict:
"""Return a minimal schema-valid artifact for tests."""
if name == "research_brief":
return {
"version": "1.0",
"topic": "Test Topic",
"research_date": "2026-03-27",
"landscape": {
"existing_content": [
{"title": "Existing Video 1", "source": "youtube", "angle": "tutorial", "what_it_covers": "basics"},
{"title": "Existing Video 2", "source": "blog", "angle": "deep dive", "what_it_covers": "advanced"},
{"title": "Existing Video 3", "source": "youtube", "angle": "comparison", "what_it_covers": "alternatives"},
],
"saturated_angles": ["basic tutorial"],
"underserved_gaps": ["misconceptions about topic"],
},
"data_points": [
{"claim": "73% of users prefer X", "source_url": "https://example.com/study", "credibility": "primary_source"},
{"claim": "Market grew 40% in 2025", "source_url": "https://example.com/report", "credibility": "secondary_source"},
{"claim": "Most experts agree on Y", "source_url": "https://example.com/survey", "credibility": "primary_source"},
],
"audience_insights": {
"common_questions": ["What is X?", "How does X work?", "Why is X important?"],
"misconceptions": [{"myth": "X is slow", "reality": "X is fast"}],
"knowledge_level": "Beginner to intermediate",
},
"angles_discovered": [
{"name": "The Surprising Truth", "hook": "You think X is slow. It's not.", "type": "contrarian", "why_now": "New benchmark data", "grounded_in": ["data_point_1"]},
{"name": "X From Scratch", "hook": "Build X in 5 minutes.", "type": "evergreen", "why_now": "Audience demand", "grounded_in": ["audience_q1"]},
{"name": "Why X Matters Now", "hook": "X just changed everything.", "type": "trending", "why_now": "Recent announcement", "grounded_in": ["trending_1"]},
],
"sources": [
{"url": "https://example.com/study", "title": "Study on X", "used_for": "data_points"},
{"url": "https://example.com/report", "title": "Market Report", "used_for": "data_points"},
{"url": "https://example.com/survey", "title": "Expert Survey", "used_for": "data_points"},
{"url": "https://example.com/reddit", "title": "Reddit Discussion", "used_for": "audience_insights"},
{"url": "https://example.com/blog", "title": "Tech Blog", "used_for": "landscape"},
],
}
if name == "proposal_packet":
return {
"version": "1.0",
"concept_options": [
{
"id": "c1", "title": "The Surprising Truth About X", "hook": "You think X is slow.",
"narrative_structure": "myth_busting", "visual_approach": "animated diagrams",
"target_duration_seconds": 60, "why_this_works": "Strong misconception found in research",
},
{
"id": "c2", "title": "X From Scratch", "hook": "Build X in 5 minutes.",
"narrative_structure": "tutorial", "visual_approach": "code walkthrough",
"target_duration_seconds": 90, "why_this_works": "High demand in audience questions",
},
{
"id": "c3", "title": "Why X Matters Now", "hook": "X just changed everything.",
"narrative_structure": "timeline", "visual_approach": "motion graphics",
"target_duration_seconds": 75, "why_this_works": "Recent announcement creates timeliness",
},
],
"selected_concept": {"concept_id": "c1", "rationale": "Strongest research backing"},
"production_plan": {
"pipeline": "animated-explainer",
"stages": [
{"stage": "script", "tools": [], "approach": "Write from research"},
{"stage": "assets", "tools": [{"tool_name": "tts_selector", "role": "narration", "available": True}], "approach": "Generate assets"},
],
},
"cost_estimate": {
"total_estimated_usd": 0.52,
"line_items": [{"tool": "elevenlabs_tts", "operation": "narration", "estimated_usd": 0.18}],
"budget_verdict": "within_budget",
},
"approval": {"status": "approved"},
}
if name == "brief":
return {
"version": "1.0",
"title": "Test Brief",
"hook": "Did you know?",
"key_points": ["point 1"],
"tone": "casual",
"style": "clean-professional",
"target_platform": "youtube",
"target_duration_seconds": 60,
}
if name == "script":
return {
"version": "1.0",
"title": "Test Script",
"total_duration_seconds": 60,
"sections": [
{
"id": "s1",
"text": "Hello world",
"start_seconds": 0,
"end_seconds": 10,
}
],
}
if name == "scene_plan":
return {
"version": "1.0",
"scenes": [
{
"id": "scene-1",
"type": "talking_head",
"description": "Host on camera",
"start_seconds": 0,
"end_seconds": 10,
}
],
}
if name == "asset_manifest":
return {
"version": "1.0",
"assets": [
{
"id": "asset-1",
"type": "video",
"path": "assets/clip.mp4",
"source_tool": "ffmpeg",
"scene_id": "scene-1",
}
],
}
if name == "edit_decisions":
return {
"version": "1.0",
"cuts": [
{
"id": "cut-1",
"source": "asset-1",
"in_seconds": 0,
"out_seconds": 10,
}
],
}
if name == "render_report":
return {
"version": "1.0",
"outputs": [
{
"path": "renders/output.mp4",
"format": "mp4",
"resolution": "1920x1080",
"duration_seconds": 60,
}
],
}
if name == "publish_log":
return {
"version": "1.0",
"entries": [
{
"platform": "youtube",
"status": "draft",
"timestamp": datetime.now(timezone.utc).isoformat(),
}
],
}
raise KeyError(f"Unknown artifact sample: {name}")
# ---- Config ----
class TestConfig:
def test_load_defaults(self):
config = OpenMontageConfig()
assert config.llm.provider == "anthropic"
assert config.budget.mode.value == "warn"
assert config.checkpoint.policy.value == "guided"
def test_load_from_yaml(self):
config = OpenMontageConfig.load()
assert config.budget.total_usd == 10.0
# ---- Schemas ----
class TestSchemas:
def test_all_schemas_loadable(self):
names = list_schemas()
assert len(names) >= 7
for name in names:
schema = load_schema(name)
assert "$schema" in schema
def test_brief_validates(self):
validate_artifact("brief", sample_artifact("brief"))
def test_brief_rejects_invalid(self):
with pytest.raises(Exception):
validate_artifact("brief", {"version": "1.0"})
# ---- Checkpoint ----
class TestCheckpoint:
def test_write_read_roundtrip(self, tmp_path):
write_checkpoint(
tmp_path, "test_project", "research", "completed",
{"research_brief": sample_artifact("research_brief")},
)
cp = read_checkpoint(tmp_path, "test_project", "research")
assert cp is not None
assert cp["stage"] == "research"
assert cp["status"] == "completed"
assert cp["artifacts"]["research_brief"]["topic"] == "Test Topic"
def test_get_next_stage(self, tmp_path):
assert get_next_stage(tmp_path, "proj") == "research"
write_checkpoint(
tmp_path,
"proj",
"research",
"completed",
{"research_brief": sample_artifact("research_brief")},
)
assert get_next_stage(tmp_path, "proj") == "proposal"
def test_invalid_stage_rejected(self, tmp_path):
with pytest.raises(ValueError):
write_checkpoint(tmp_path, "proj", "invalid_stage", "completed", {})
def test_invalid_canonical_artifact_rejected(self, tmp_path):
with pytest.raises(CheckpointValidationError):
write_checkpoint(
tmp_path,
"proj",
"research",
"completed",
{"research_brief": {"topic": "missing schema fields"}},
)
def test_missing_canonical_artifact_rejected(self, tmp_path):
with pytest.raises(CheckpointValidationError):
write_checkpoint(tmp_path, "proj", "research", "completed", {})
def test_invalid_status_rejected(self, tmp_path):
with pytest.raises(CheckpointValidationError):
write_checkpoint(
tmp_path,
"proj",
"research",
"mystery",
{"research_brief": sample_artifact("research_brief")},
)
# ---- Pipeline manifests ----
class TestPipelineManifests:
def test_framework_smoke_manifest_loads(self):
manifest = load_pipeline("framework-smoke")
assert manifest["name"] == "framework-smoke"
assert get_stage_order(manifest) == ["research", "script"]
assert get_required_tools(manifest) == set()
def test_framework_smoke_manifest_listed(self):
assert "framework-smoke" in list_pipelines()
# ---- BaseTool ----
class DummyTool(BaseTool):
name = "dummy"
version = "0.1.0"
tier = ToolTier.CORE
capabilities = ["test"]
dependencies = []
def execute(self, inputs):
return ToolResult(success=True, data={"echo": inputs})
class TestBaseTool:
def test_get_info(self):
tool = DummyTool()
info = tool.get_info()
assert info["name"] == "dummy"
assert info["tier"] == "core"
assert info["status"] == "available"
def test_execute(self):
tool = DummyTool()
result = tool.execute({"msg": "hello"})
assert result.success
def test_unavailable_when_deps_missing(self):
class MissingDepTool(BaseTool):
name = "missing"
dependencies = ["cmd:nonexistent_binary_xyz"]
def execute(self, inputs):
return ToolResult(success=True)
tool = MissingDepTool()
assert tool.get_status() == ToolStatus.UNAVAILABLE
# ---- ToolRegistry ----
class TestToolRegistry:
def test_register_and_find(self):
reg = ToolRegistry()
reg.register(DummyTool())
assert reg.get("dummy") is not None
assert "dummy" in reg.list_all()
assert len(reg.get_by_tier(ToolTier.CORE)) == 1
assert len(reg.find_by_capability("test")) == 1
def test_support_envelope(self):
reg = ToolRegistry()
reg.register(DummyTool())
envelope = reg.support_envelope()
assert "dummy" in envelope
assert envelope["dummy"]["status"] == "available"
def test_discovers_concrete_tools_from_package(self, tmp_path, monkeypatch):
package_dir = tmp_path / "demo_tools"
package_dir.mkdir()
(package_dir / "__init__.py").write_text("", encoding="utf-8")
(package_dir / "demo_tool.py").write_text(
"\n".join(
[
"from tools.base_tool import BaseTool, ToolResult, ToolTier",
"",
"class DiscoveredTool(BaseTool):",
" name = 'discovered'",
" tier = ToolTier.CORE",
" capabilities = ['discover']",
" dependencies = []",
"",
" def execute(self, inputs):",
" return ToolResult(success=True, data=inputs)",
"",
]
),
encoding="utf-8",
)
monkeypatch.syspath_prepend(str(tmp_path))
importlib.invalidate_caches()
reg = ToolRegistry()
discovered = reg.discover("demo_tools")
assert discovered == ["discovered"]
assert reg.get("discovered") is not None
assert reg.find_by_capability("discover")[0].name == "discovered"
# ---- CostTracker ----
class TestCostTracker:
def test_estimate_reserve_reconcile(self):
tracker = CostTracker(budget_total_usd=10.0, mode=BudgetMode.OBSERVE)
entry_id = tracker.estimate("image_selector", "generate", 0.05)
tracker.reserve(entry_id)
assert tracker.budget_reserved_usd == 0.05
tracker.reconcile(entry_id, 0.04, success=True)
assert tracker.budget_spent_usd == 0.04
assert tracker.budget_reserved_usd == 0.0
def test_cap_mode_blocks_overspend(self):
tracker = CostTracker(
budget_total_usd=1.0,
mode=BudgetMode.CAP,
single_action_approval_usd=10.0, # raise threshold so budget check triggers
)
tracker.approve_tool("expensive")
eid = tracker.estimate("expensive", "op", 5.0)
with pytest.raises(BudgetExceededError):
tracker.reserve(eid)
def test_persistence(self, tmp_path):
log_path = tmp_path / "cost_log.json"
t1 = CostTracker(budget_total_usd=10.0, mode=BudgetMode.OBSERVE, cost_log_path=log_path)
eid = t1.estimate("tool", "op", 0.10)
t1.reserve(eid)
t1.reconcile(eid, 0.08)
t2 = CostTracker(cost_log_path=log_path)
assert t2.budget_spent_usd == 0.08
# ---- Pipeline Instruction Architecture ----
class TestPipelineInstructionArchitecture:
"""Verify that the instruction-driven architecture is in place:
manifests reference skills, not Python handlers."""
def test_animated_explainer_stages_have_skills(self):
try:
manifest = load_pipeline("animated-explainer")
except FileNotFoundError:
pytest.skip("animated-explainer manifest not yet created")
for stage in manifest["stages"]:
assert "skill" in stage, f"Stage {stage['name']} missing skill field"
def test_stage_skill_lookup(self):
manifest = load_pipeline("framework-smoke")
# framework-smoke may not have skills yet — just verify the function works
result = get_stage_skill(manifest, "idea")
assert result is None or isinstance(result, str)
def test_stage_review_focus_lookup(self):
manifest = load_pipeline("framework-smoke")
result = get_stage_review_focus(manifest, "idea")
assert isinstance(result, list)
# ---- Agent context files ----
class TestAgentContextFiles:
def test_agent_guide_contains_canonical_sections(self):
contents = (PROJECT_ROOT / "AGENT_GUIDE.md").read_text(encoding="utf-8")
for header in (
"## Orchestrator",
"## Stage Agents",
"## Reviewer Protocol",
"## Communication Protocol",
"## Human Checkpoint Protocol",
):
assert header in contents
def test_platform_wrappers_reference_agent_guide(self):
for path in ("CLAUDE.md", "CODEX.md", "CURSOR.md", "COPILOT.md", "AGENTS.md"):
contents = (PROJECT_ROOT / path).read_text(encoding="utf-8")
assert "AGENT_GUIDE.md" in contents
# ---- Media Profiles ----
class TestMediaProfiles:
def test_all_profiles_exist(self):
assert len(ALL_PROFILES) >= 9
def test_get_profile(self):
p = get_profile("youtube_landscape")
assert p.width == 1920
assert p.height == 1080
def test_ffmpeg_args(self):
args = ffmpeg_output_args(get_profile("tiktok"))
assert "-c:v" in args
assert "1080" in args[-1]
def test_unknown_profile_raises(self):
with pytest.raises(ValueError):
get_profile("nonexistent")
+362
View File
@@ -0,0 +1,362 @@
"""Phase 1 contract tests — Core Talking-Head Pipeline tools."""
import json
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from tools.base_tool import BaseTool, ToolResult, ToolTier, ToolStatus, DependencyError
from tools.tool_registry import ToolRegistry
from lib.pipeline_loader import load_pipeline, get_stage_order, get_required_tools, list_pipelines
# ---- Tool imports ----
from tools.analysis.transcriber import Transcriber
from tools.video.video_trimmer import VideoTrimmer
from tools.subtitle.subtitle_gen import SubtitleGen
from tools.analysis.frame_sampler import FrameSampler
from tools.audio.audio_mixer import AudioMixer
from tools.video.video_compose import VideoCompose
# ---- Contract: every tool inherits BaseTool and has required fields ----
PHASE1_TOOLS = [
Transcriber,
VideoTrimmer,
SubtitleGen,
FrameSampler,
AudioMixer,
VideoCompose,
]
class TestPhase1ToolContracts:
"""Verify all Phase 1 tools satisfy the ToolContract."""
@pytest.mark.parametrize("tool_cls", PHASE1_TOOLS)
def test_inherits_base_tool(self, tool_cls):
assert issubclass(tool_cls, BaseTool)
@pytest.mark.parametrize("tool_cls", PHASE1_TOOLS)
def test_has_required_identity(self, tool_cls):
tool = tool_cls()
assert tool.name, f"{tool_cls.__name__} must have a non-empty name"
assert tool.version, f"{tool_cls.__name__} must have a version"
assert tool.tier in ToolTier
assert len(tool.capabilities) > 0, f"{tool_cls.__name__} must declare capabilities"
@pytest.mark.parametrize("tool_cls", PHASE1_TOOLS)
def test_get_info_returns_valid_dict(self, tool_cls):
tool = tool_cls()
info = tool.get_info()
assert isinstance(info, dict)
assert info["name"] == tool.name
assert info["tier"] in [t.value for t in ToolTier]
assert info["status"] in ["available", "unavailable", "degraded"]
@pytest.mark.parametrize("tool_cls", PHASE1_TOOLS)
def test_has_input_schema(self, tool_cls):
tool = tool_cls()
assert isinstance(tool.input_schema, dict)
assert "properties" in tool.input_schema or "type" in tool.input_schema
@pytest.mark.parametrize("tool_cls", PHASE1_TOOLS)
def test_execute_is_implemented(self, tool_cls):
"""Verify execute() is not the abstract stub."""
tool = tool_cls()
# Should not raise TypeError — it's implemented
assert callable(tool.execute)
@pytest.mark.parametrize("tool_cls", PHASE1_TOOLS)
def test_dry_run_returns_dict(self, tool_cls):
tool = tool_cls()
result = tool.dry_run({})
assert isinstance(result, dict)
assert "tool" in result
assert result["tool"] == tool.name
# ---- Contract: tools report correct status based on dependencies ----
class TestPhase1ToolStatus:
def test_subtitle_gen_always_available(self):
"""SubtitleGen has no external dependencies — always available."""
tool = SubtitleGen()
assert tool.get_status() == ToolStatus.AVAILABLE
def test_transcriber_reports_status_correctly(self):
"""Transcriber should report unavailable if faster_whisper not installed."""
tool = Transcriber()
status = tool.get_status()
assert status in (ToolStatus.AVAILABLE, ToolStatus.UNAVAILABLE)
def test_ffmpeg_tools_report_status(self):
"""FFmpeg-dependent tools should report based on ffmpeg availability."""
for cls in [VideoTrimmer, FrameSampler, AudioMixer, VideoCompose]:
tool = cls()
status = tool.get_status()
assert status in (ToolStatus.AVAILABLE, ToolStatus.UNAVAILABLE)
# ---- Contract: tool names are unique ----
class TestPhase1ToolNames:
def test_unique_names(self):
names = [cls().name for cls in PHASE1_TOOLS]
assert len(names) == len(set(names)), f"Duplicate tool names: {names}"
def test_expected_names(self):
names = {cls().name for cls in PHASE1_TOOLS}
expected = {"transcriber", "video_trimmer", "subtitle_gen", "frame_sampler", "audio_mixer", "video_compose"}
assert names == expected
# ---- Contract: tools are discoverable via registry ----
class TestPhase1ToolDiscovery:
def test_all_phase1_tools_discoverable(self):
"""Registry.discover() should find all Phase 1 tools."""
reg = ToolRegistry()
discovered = reg.discover("tools")
for cls in PHASE1_TOOLS:
name = cls().name
assert name in discovered or reg.get(name) is not None, (
f"Tool {name!r} not discovered by registry"
)
def test_phase1_tools_are_core_tier(self):
"""All Phase 1 tools should be in the CORE tier."""
for cls in PHASE1_TOOLS:
tool = cls()
assert tool.tier == ToolTier.CORE, f"{tool.name} should be CORE tier"
# ---- Contract: SubtitleGen produces valid output without FFmpeg ----
class TestSubtitleGenUnit:
def test_srt_generation(self):
segments = [
{
"text": "Hello world",
"start": 0.0,
"end": 1.5,
"words": [
{"word": "Hello", "start": 0.0, "end": 0.5},
{"word": "world", "start": 0.6, "end": 1.5},
],
},
{
"text": "This is a test",
"start": 2.0,
"end": 4.0,
"words": [
{"word": "This", "start": 2.0, "end": 2.3},
{"word": "is", "start": 2.4, "end": 2.5},
{"word": "a", "start": 2.6, "end": 2.7},
{"word": "test", "start": 2.8, "end": 4.0},
],
},
]
tool = SubtitleGen()
result = tool.execute({
"segments": segments,
"format": "srt",
"output_path": "test_output.srt",
})
assert result.success
assert len(result.artifacts) == 1
content = Path(result.artifacts[0]).read_text()
assert "-->" in content
assert "Hello world" in content
# Cleanup
Path(result.artifacts[0]).unlink(missing_ok=True)
def test_vtt_generation(self):
segments = [
{
"text": "Test cue",
"start": 1.0,
"end": 3.0,
"words": [
{"word": "Test", "start": 1.0, "end": 1.5},
{"word": "cue", "start": 1.6, "end": 3.0},
],
},
]
tool = SubtitleGen()
result = tool.execute({
"segments": segments,
"format": "vtt",
"output_path": "test_output.vtt",
})
assert result.success
content = Path(result.artifacts[0]).read_text()
assert content.startswith("WEBVTT")
Path(result.artifacts[0]).unlink(missing_ok=True)
def test_json_generation(self):
segments = [
{
"text": "JSON test",
"start": 0.0,
"end": 2.0,
"words": [
{"word": "JSON", "start": 0.0, "end": 0.8},
{"word": "test", "start": 0.9, "end": 2.0},
],
},
]
tool = SubtitleGen()
result = tool.execute({
"segments": segments,
"format": "json",
"output_path": "test_output.caption.json",
})
assert result.success
data = json.loads(Path(result.artifacts[0]).read_text())
assert "cues" in data
assert len(data["cues"]) >= 1
Path(result.artifacts[0]).unlink(missing_ok=True)
def test_word_grouping_respects_max_words(self):
words = [{"word": f"w{i}", "start": i * 0.5, "end": i * 0.5 + 0.4} for i in range(20)]
segments = [{"text": " ".join(w["word"] for w in words), "start": 0.0, "end": 10.0, "words": words}]
tool = SubtitleGen()
result = tool.execute({
"segments": segments,
"format": "json",
"max_words_per_cue": 4,
"output_path": "test_grouping.caption.json",
})
assert result.success
data = json.loads(Path(result.artifacts[0]).read_text())
for cue in data["cues"]:
assert len(cue["words"]) <= 4
Path(result.artifacts[0]).unlink(missing_ok=True)
def test_segment_fallback_without_words(self):
"""Segments without word-level timestamps use segment-level timing."""
segments = [
{"text": "No word data", "start": 0.0, "end": 2.0},
]
tool = SubtitleGen()
result = tool.execute({
"segments": segments,
"format": "srt",
"output_path": "test_fallback.srt",
})
assert result.success
content = Path(result.artifacts[0]).read_text()
assert "No word data" in content
Path(result.artifacts[0]).unlink(missing_ok=True)
# ---- Contract: missing input file returns error, not crash ----
class TestPhase1ErrorHandling:
def test_transcriber_missing_file(self):
tool = Transcriber()
result = tool.execute({"input_path": "/nonexistent/file.mp4"})
assert not result.success
assert "not found" in result.error.lower() or "not installed" in result.error.lower()
def test_video_trimmer_missing_file(self):
tool = VideoTrimmer()
result = tool.execute({
"operation": "cut",
"input_path": "/nonexistent/file.mp4",
})
assert not result.success
def test_frame_sampler_missing_file(self):
tool = FrameSampler()
result = tool.execute({
"input_path": "/nonexistent/file.mp4",
"strategy": "interval",
})
assert not result.success
def test_audio_mixer_missing_tracks(self):
tool = AudioMixer()
result = tool.execute({"operation": "mix", "tracks": []})
assert not result.success
def test_video_compose_missing_decisions(self):
tool = VideoCompose()
result = tool.execute({"operation": "compose"})
assert not result.success
# ---- Contract: talking-head pipeline manifest ----
class TestTalkingHeadManifest:
def test_manifest_loads(self):
manifest = load_pipeline("talking-head")
assert manifest["name"] == "talking-head"
assert manifest["category"] == "talking_head"
def test_manifest_has_all_stages(self):
manifest = load_pipeline("talking-head")
stages = get_stage_order(manifest)
assert stages == ["idea", "script", "scene_plan", "assets", "edit", "compose", "publish"]
def test_manifest_references_phase1_tools(self):
manifest = load_pipeline("talking-head")
tools = get_required_tools(manifest)
phase1_tools = {"transcriber", "video_trimmer", "subtitle_gen", "frame_sampler", "audio_mixer", "video_compose"}
# At least some Phase 1 tools should be referenced
assert len(tools & phase1_tools) > 0
def test_manifest_listed(self):
assert "talking-head" in list_pipelines()
def test_manifest_has_required_skills(self):
manifest = load_pipeline("talking-head")
skills = manifest.get("required_skills", [])
# Instruction-driven architecture: skills are stage director + meta skills
assert any("talking-head" in s for s in skills)
assert any("reviewer" in s for s in skills)
assert any("checkpoint-protocol" in s for s in skills)
def test_idea_and_publish_require_approval(self):
manifest = load_pipeline("talking-head")
for stage in manifest["stages"]:
if stage["name"] in ("idea", "publish"):
assert stage.get("human_approval_default") is True
# ---- Contract: skill files exist ----
class TestPhase1Skills:
@pytest.mark.parametrize("skill_path", [
"skills/core/ffmpeg.md",
"skills/core/whisperx.md",
"skills/core/subtitle-sync.md",
"skills/creative/video-editing.md",
"skills/creative/enhancement-strategy.md",
])
def test_skill_file_exists(self, skill_path):
full_path = PROJECT_ROOT / skill_path
assert full_path.exists(), f"Skill file missing: {skill_path}"
@pytest.mark.parametrize("skill_path", [
"skills/core/ffmpeg.md",
"skills/core/whisperx.md",
"skills/core/subtitle-sync.md",
"skills/creative/video-editing.md",
"skills/creative/enhancement-strategy.md",
])
def test_skill_has_content(self, skill_path):
full_path = PROJECT_ROOT / skill_path
content = full_path.read_text(encoding="utf-8")
assert len(content) > 100, f"Skill file too short: {skill_path}"
assert "## When to Use" in content, f"Skill missing 'When to Use' section: {skill_path}"
assert "## Quality Checklist" in content, f"Skill missing 'Quality Checklist' section: {skill_path}"
+62
View File
@@ -0,0 +1,62 @@
"""Phase 1 golden scenario test — validates the talking-head pipeline
manifest and skill architecture are in place.
The old test ran the Python orchestrator pipeline end-to-end. That layer
has been removed in favor of instruction-driven architecture: the agent
reads pipeline manifests + stage director skills and drives the pipeline
itself. These tests verify the infrastructure is correctly wired.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from lib.checkpoint import STAGES
from lib.pipeline_loader import (
load_pipeline,
get_stage_order,
get_stage_skill,
get_stage_review_focus,
list_pipelines,
)
from schemas.artifacts import load_schema, validate_artifact, list_schemas
class TestTalkingHeadManifest:
"""Verify the talking-head pipeline manifest is well-formed."""
def test_manifest_loads(self):
manifest = load_pipeline("talking-head")
assert manifest["name"] == "talking-head"
def test_all_stages_present(self):
manifest = load_pipeline("talking-head")
stage_names = get_stage_order(manifest)
expected = ["idea", "script", "scene_plan", "assets", "edit", "compose", "publish"]
assert stage_names == expected
def test_manifest_listed(self):
assert "talking-head" in list_pipelines()
class TestGoldenScenarioArtifacts:
"""Validate golden scenario artifact samples against schemas."""
GOLDEN_PATH = PROJECT_ROOT / "eval" / "golden_scenarios" / "talking_head_basic.json"
@pytest.fixture
def golden(self):
if not self.GOLDEN_PATH.exists():
pytest.skip("Golden scenario file not found")
return json.loads(self.GOLDEN_PATH.read_text())
def test_golden_file_structure(self, golden):
assert "inputs" in golden
assert "expected_artifacts" in golden
+170
View File
@@ -0,0 +1,170 @@
"""Phase 2 side-by-side comparison: Phase 1 vs Phase 2 enhanced output.
Runs the talking-head compose handler twice — once with enhance=False
(Phase 1 baseline) and once with enhance=True (Phase 2 enhanced) —
then asserts both outputs exist and reports file size / duration for
manual comparison.
"""
import json
import subprocess
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
FOOTAGE = Path(
r"C:\Users\ishan\Documents\SocialMedia\IshanAIVideos\pipeline\2026-03-08\seg1_intro.mp4"
)
def _has_ffmpeg() -> bool:
import shutil
return shutil.which("ffmpeg") is not None
def _get_duration(path: str) -> float:
result = subprocess.run(
["ffprobe", "-v", "quiet", "-show_entries", "format=duration", "-of", "json", path],
capture_output=True, text=True, check=True,
)
data = json.loads(result.stdout)
return float(data.get("format", {}).get("duration", 0))
def _get_file_size_mb(path: str) -> float:
return Path(path).stat().st_size / (1024 * 1024)
@pytest.fixture(scope="module")
def comparison_outputs(tmp_path_factory):
"""Run compose handler twice: baseline and enhanced."""
if not FOOTAGE.exists():
pytest.skip("Test footage not available")
if not _has_ffmpeg():
pytest.skip("FFmpeg not available")
from tools.video.video_compose import VideoCompose
from tools.enhancement.face_enhance import FaceEnhance
from tools.enhancement.color_grade import ColorGrade
from tools.audio.audio_enhance import AudioEnhance
tmp = tmp_path_factory.mktemp("comparison")
results = {}
# Phase 1 baseline: just encode
composer = VideoCompose()
baseline_path = str(tmp / "phase1_baseline.mp4")
r = composer.execute({
"operation": "encode",
"input_path": str(FOOTAGE),
"output_path": baseline_path,
"codec": "libx264",
"crf": 20,
})
assert r.success, f"Baseline encode failed: {r.error}"
results["baseline"] = baseline_path
# Phase 2 enhanced: face -> color -> audio
current = str(FOOTAGE)
enhancements = []
face = FaceEnhance()
face_out = str(tmp / "step_face.mp4")
r = face.execute({
"input_path": current,
"output_path": face_out,
"presets": ["talking_head_standard"],
})
if r.success:
current = r.data["output"]
enhancements.append("face_enhance:talking_head_standard")
color = ColorGrade()
color_out = str(tmp / "step_color.mp4")
r = color.execute({
"input_path": current,
"output_path": color_out,
"profile": "cinematic_warm",
"intensity": 0.85,
})
if r.success:
current = r.data["output"]
enhancements.append("color_grade:cinematic_warm@0.85")
audio = AudioEnhance()
audio_out = str(tmp / "step_audio.mp4")
r = audio.execute({
"input_path": current,
"output_path": audio_out,
"preset": "clean_speech",
})
if r.success:
current = r.data["output"]
enhancements.append("audio_enhance:clean_speech")
# Final encode
enhanced_path = str(tmp / "phase2_enhanced.mp4")
r = composer.execute({
"operation": "encode",
"input_path": current,
"output_path": enhanced_path,
"codec": "libx264",
"crf": 20,
})
assert r.success, f"Enhanced encode failed: {r.error}"
results["enhanced"] = enhanced_path
results["enhancements"] = enhancements
return results
class TestPhase2Comparison:
def test_baseline_exists(self, comparison_outputs):
assert Path(comparison_outputs["baseline"]).exists()
def test_enhanced_exists(self, comparison_outputs):
assert Path(comparison_outputs["enhanced"]).exists()
def test_both_have_valid_duration(self, comparison_outputs):
base_dur = _get_duration(comparison_outputs["baseline"])
enh_dur = _get_duration(comparison_outputs["enhanced"])
assert base_dur > 0
assert enh_dur > 0
# Durations should be within 1 second of each other
assert abs(base_dur - enh_dur) < 1.0, (
f"Duration mismatch: baseline={base_dur:.1f}s, enhanced={enh_dur:.1f}s"
)
def test_enhancements_were_applied(self, comparison_outputs):
enhancements = comparison_outputs["enhancements"]
assert len(enhancements) > 0, "No enhancements were applied"
def test_report(self, comparison_outputs):
"""Print comparison report for manual review."""
baseline = comparison_outputs["baseline"]
enhanced = comparison_outputs["enhanced"]
base_size = _get_file_size_mb(baseline)
enh_size = _get_file_size_mb(enhanced)
base_dur = _get_duration(baseline)
enh_dur = _get_duration(enhanced)
enhancements = comparison_outputs["enhancements"]
report = (
f"\n{'='*60}\n"
f" Phase 1 vs Phase 2 Comparison\n"
f"{'='*60}\n"
f" Baseline: {base_size:.2f} MB, {base_dur:.1f}s\n"
f" Enhanced: {enh_size:.2f} MB, {enh_dur:.1f}s\n"
f" Enhancements: {', '.join(enhancements)}\n"
f" Baseline path: {baseline}\n"
f" Enhanced path: {enhanced}\n"
f"{'='*60}\n"
)
print(report)
# Always passes — this test is for reporting
assert True
+266
View File
@@ -0,0 +1,266 @@
"""Phase 2 contract tests — Enhancement Layer tools."""
import json
import shutil
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from tools.base_tool import BaseTool, ToolResult, ToolTier, ToolStatus
from tools.tool_registry import ToolRegistry
from tools.enhancement.face_enhance import FaceEnhance, PRESETS as FACE_PRESETS
from tools.analysis.scene_detect import SceneDetect
from tools.enhancement.color_grade import ColorGrade, PROFILES as COLOR_PROFILES
from tools.audio.audio_enhance import AudioEnhance, PRESETS as AUDIO_PRESETS
from tools.graphics.image_selector import ImageSelector
from tools.graphics.code_snippet import CodeSnippet, THEMES as CODE_THEMES
from tools.graphics.diagram_gen import DiagramGen
PHASE2_TOOLS = [
FaceEnhance,
SceneDetect,
ColorGrade,
AudioEnhance,
ImageSelector,
CodeSnippet,
DiagramGen,
]
# ---- Contract: every tool inherits BaseTool and has required fields ----
class TestPhase2ToolContracts:
@pytest.mark.parametrize("tool_cls", PHASE2_TOOLS)
def test_inherits_base_tool(self, tool_cls):
assert issubclass(tool_cls, BaseTool)
@pytest.mark.parametrize("tool_cls", PHASE2_TOOLS)
def test_has_required_identity(self, tool_cls):
tool = tool_cls()
assert tool.name
assert tool.version
assert tool.tier in ToolTier
assert len(tool.capabilities) > 0
@pytest.mark.parametrize("tool_cls", PHASE2_TOOLS)
def test_get_info_returns_valid_dict(self, tool_cls):
tool = tool_cls()
info = tool.get_info()
assert isinstance(info, dict)
assert info["name"] == tool.name
assert info["status"] in ["available", "unavailable", "degraded"]
@pytest.mark.parametrize("tool_cls", PHASE2_TOOLS)
def test_execute_is_implemented(self, tool_cls):
tool = tool_cls()
assert callable(tool.execute)
@pytest.mark.parametrize("tool_cls", PHASE2_TOOLS)
def test_dry_run_returns_dict(self, tool_cls):
tool = tool_cls()
result = tool.dry_run({})
assert isinstance(result, dict)
assert result["tool"] == tool.name
# ---- Contract: unique names, all CORE tier ----
class TestPhase2ToolNames:
def test_unique_names(self):
names = [cls().name for cls in PHASE2_TOOLS]
assert len(names) == len(set(names))
def test_expected_names(self):
names = {cls().name for cls in PHASE2_TOOLS}
expected = {
"face_enhance", "scene_detect", "color_grade", "audio_enhance",
"image_selector", "code_snippet", "diagram_gen",
}
assert names == expected
def test_expected_tiers(self):
for cls in PHASE2_TOOLS:
tool = cls()
if tool.name == "image_selector":
assert tool.tier == ToolTier.GENERATE
else:
assert tool.tier == ToolTier.CORE, f"{tool.name} should be CORE"
# ---- Contract: discoverable via registry ----
class TestPhase2ToolDiscovery:
def test_all_phase2_tools_discoverable(self):
reg = ToolRegistry()
reg.discover("tools")
for cls in PHASE2_TOOLS:
name = cls().name
assert reg.get(name) is not None, f"{name} not discovered"
# ---- Contract: presets/profiles are well-formed ----
class TestPresets:
def test_face_presets_have_vf(self):
for name, preset in FACE_PRESETS.items():
assert "vf" in preset, f"Face preset {name} missing 'vf'"
assert "description" in preset
def test_color_profiles_have_vf(self):
for name, profile in COLOR_PROFILES.items():
assert "vf" in profile, f"Color profile {name} missing 'vf'"
assert "description" in profile
def test_audio_presets_have_af(self):
for name, preset in AUDIO_PRESETS.items():
assert "af" in preset, f"Audio preset {name} missing 'af'"
assert "description" in preset
def test_code_themes_have_required_fields(self):
for name, theme in CODE_THEMES.items():
assert "pygments_style" in theme
assert "bg_color" in theme
assert "text_color" in theme
def test_face_list_presets(self):
presets = FaceEnhance.list_presets()
assert len(presets) >= 8
assert "talking_head_standard" in presets
def test_color_list_profiles(self):
profiles = ColorGrade.list_profiles()
assert len(profiles) >= 7
assert "cinematic_warm" in profiles
def test_audio_list_presets(self):
presets = AudioEnhance.list_presets()
assert len(presets) >= 5
assert "clean_speech" in presets
# ---- Contract: error handling for missing inputs ----
class TestPhase2ErrorHandling:
def test_face_enhance_missing_file(self):
tool = FaceEnhance()
r = tool.execute({"input_path": "/nonexistent.mp4"})
assert not r.success
def test_scene_detect_missing_file(self):
tool = SceneDetect()
r = tool.execute({"input_path": "/nonexistent.mp4"})
assert not r.success
def test_color_grade_missing_file(self):
tool = ColorGrade()
r = tool.execute({"input_path": "/nonexistent.mp4"})
assert not r.success
def test_audio_enhance_missing_file(self):
tool = AudioEnhance()
r = tool.execute({"input_path": "/nonexistent.mp4"})
assert not r.success
def test_image_selector_no_provider(self):
tool = ImageSelector()
# Will fail if no API key or local model
r = tool.execute({"prompt": "test"})
# Either succeeds (provider available) or fails gracefully
assert isinstance(r, ToolResult)
def test_diagram_gen_empty_boxes(self):
tool = DiagramGen()
if tool.get_status() == ToolStatus.AVAILABLE:
r = tool.execute({"diagram_type": "boxes", "boxes": []})
assert isinstance(r, ToolResult)
# ---- Contract: code_snippet renders valid images ----
class TestCodeSnippetUnit:
@pytest.fixture
def has_deps(self):
tool = CodeSnippet()
if tool.get_status() != ToolStatus.AVAILABLE:
pytest.skip("Pygments/Pillow not installed")
def test_render_python(self, has_deps, tmp_path):
tool = CodeSnippet()
r = tool.execute({
"code": "def hello():\n print('Hello, world!')\n",
"language": "python",
"theme": "monokai",
"output_path": str(tmp_path / "test.png"),
})
assert r.success
assert Path(r.data["output"]).exists()
assert r.data["line_count"] == 3
def test_render_with_title(self, has_deps, tmp_path):
tool = CodeSnippet()
r = tool.execute({
"code": "console.log('test');",
"language": "javascript",
"theme": "dracula",
"title": "example.js",
"output_path": str(tmp_path / "titled.png"),
})
assert r.success
def test_render_different_themes(self, has_deps, tmp_path):
tool = CodeSnippet()
for theme_name in ["monokai", "github_dark", "light"]:
r = tool.execute({
"code": "x = 42",
"language": "python",
"theme": theme_name,
"output_path": str(tmp_path / f"{theme_name}.png"),
})
assert r.success, f"Theme {theme_name} failed"
# ---- Contract: diagram_gen renders valid images ----
class TestDiagramGenUnit:
@pytest.fixture
def has_deps(self):
tool = DiagramGen()
if tool.get_status() != ToolStatus.AVAILABLE:
pytest.skip("No diagram renderer available")
def test_render_box_diagram(self, has_deps, tmp_path):
tool = DiagramGen()
r = tool.execute({
"diagram_type": "boxes",
"title": "Test Flow",
"boxes": [
{"label": "Input", "color": "#2563eb"},
{"label": "Process", "color": "#7c3aed"},
{"label": "Output", "color": "#059669"},
],
"connections": [
{"from": 0, "to": 1, "label": "data"},
{"from": 1, "to": 2, "label": "result"},
],
"theme": "dark",
"output_path": str(tmp_path / "boxes.png"),
})
assert r.success
assert Path(r.data["output"]).exists()
assert r.data["box_count"] == 3
def test_render_mermaid_fallback(self, has_deps, tmp_path):
"""If mmdc not installed, falls back to text card."""
tool = DiagramGen()
r = tool.execute({
"diagram_type": "mermaid",
"definition": "graph TD\n A[Start] --> B[End]",
"output_path": str(tmp_path / "mermaid.png"),
})
assert r.success
+342
View File
@@ -0,0 +1,342 @@
"""Phase 3 contract tests — instruction-driven architecture.
Tests the new tools (TTS, music gen), pipeline manifests, style playbooks,
stage director skills, meta skills, and the animated-explainer pipeline.
"""
import sys
from pathlib import Path
import pytest
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from lib.pipeline_loader import (
load_pipeline,
get_stage_order,
get_required_tools,
get_stage_skill,
get_stage_review_focus,
list_pipelines,
)
from lib.checkpoint import STAGES
from schemas.artifacts import list_schemas
from styles.playbook_loader import load_playbook, list_playbooks, validate_playbook
from tools.base_tool import ToolTier
from tools.audio.music_gen import MusicGen
from tools.tool_registry import ToolRegistry
from tools.audio.elevenlabs_tts import ElevenLabsTTS
from tools.audio.openai_tts import OpenAITTS
from tools.audio.piper_tts import PiperTTS
from tools.audio.tts_selector import TTSSelector
# ---- TTS Provider Tools ----
class TestElevenLabsTTS:
def test_identity(self):
tool = ElevenLabsTTS()
info = tool.get_info()
assert info["name"] == "elevenlabs_tts"
assert info["tier"] == "voice"
assert info["capability"] == "tts"
assert info["provider"] == "elevenlabs"
def test_cost_estimate(self):
tool = ElevenLabsTTS()
cost = tool.estimate_cost({"text": "Hello world, this is a test."})
assert cost > 0
assert cost < 0.01 # short text should be cheap
def test_capabilities(self):
tool = ElevenLabsTTS()
assert "text_to_speech" in tool.capabilities
assert "voice_selection" in tool.capabilities
class TestPiperTTS:
def test_identity(self):
tool = PiperTTS()
info = tool.get_info()
assert info["name"] == "piper_tts"
assert info["tier"] == "voice"
assert info["capability"] == "tts"
assert info["provider"] == "piper"
def test_cost_is_free(self):
tool = PiperTTS()
assert tool.estimate_cost({"text": "anything"}) == 0.0
def test_capabilities(self):
tool = PiperTTS()
assert "text_to_speech" in tool.capabilities
assert "offline_generation" in tool.capabilities
class TestMusicGen:
def test_identity(self):
tool = MusicGen()
info = tool.get_info()
assert info["name"] == "music_gen"
assert info["tier"] == "generate"
def test_cost_estimate_scales_with_duration(self):
tool = MusicGen()
cost_30 = tool.estimate_cost({"prompt": "ambient", "duration_seconds": 30})
cost_60 = tool.estimate_cost({"prompt": "ambient", "duration_seconds": 60})
assert cost_60 > cost_30
def test_capabilities(self):
tool = MusicGen()
assert "generate_background_music" in tool.capabilities
class TestNewToolsRegistry:
def test_all_register(self):
reg = ToolRegistry()
reg.register(ElevenLabsTTS())
reg.register(PiperTTS())
reg.register(MusicGen())
assert len(reg.list_all()) == 3
def test_voice_tier_tools(self):
reg = ToolRegistry()
reg.register(ElevenLabsTTS())
reg.register(OpenAITTS())
reg.register(PiperTTS())
voice_tools = reg.get_by_tier(ToolTier.VOICE)
assert len(voice_tools) == 3
names = {t.name for t in voice_tools}
assert names == {"elevenlabs_tts", "openai_tts", "piper_tts"}
class TestCapabilityMetadata:
def test_tts_tools_expose_capability_provider_and_location(self):
tool = ElevenLabsTTS()
info = tool.get_info()
assert info["capability"] == "tts"
assert info["provider"] == "elevenlabs"
assert info["usage_location"].endswith("tools\\audio\\elevenlabs_tts.py") or info["usage_location"].endswith("tools/audio/elevenlabs_tts.py")
assert "related_skills" in info
assert "fallback_tools" in info
def test_provider_specific_tts_tools_register(self):
reg = ToolRegistry()
reg.register(ElevenLabsTTS())
reg.register(OpenAITTS())
reg.register(PiperTTS())
reg.register(TTSSelector())
assert {tool.name for tool in reg.get_by_capability("tts")} == {
"elevenlabs_tts",
"openai_tts",
"piper_tts",
"tts_selector",
}
assert {tool.name for tool in reg.get_by_provider("elevenlabs")} == {"elevenlabs_tts"}
def test_registry_catalog_views(self):
reg = ToolRegistry()
reg.register(ElevenLabsTTS())
reg.register(OpenAITTS())
reg.register(PiperTTS())
catalog = reg.capability_catalog()
assert "tts" in catalog
providers = {item["provider"] for item in catalog["tts"] if item["provider"] != "selector"}
assert providers == {"elevenlabs", "openai", "piper"}
# ---- Animated Explainer Pipeline ----
class TestAnimatedExplainerManifest:
def test_loads(self):
manifest = load_pipeline("animated-explainer")
assert manifest["name"] == "animated-explainer"
assert manifest["version"] == "2.0"
def test_all_stages_present(self):
manifest = load_pipeline("animated-explainer")
stage_names = get_stage_order(manifest)
expected = ["research", "proposal", "script", "scene_plan", "assets", "edit", "compose", "publish"]
assert stage_names == expected
def test_every_stage_has_skill(self):
manifest = load_pipeline("animated-explainer")
for stage in manifest["stages"]:
assert "skill" in stage, f"Stage {stage['name']} missing skill"
skill = get_stage_skill(manifest, stage["name"])
assert skill is not None
assert skill.startswith("pipelines/explainer/")
def test_every_stage_has_review_focus(self):
manifest = load_pipeline("animated-explainer")
for stage in manifest["stages"]:
focus = get_stage_review_focus(manifest, stage["name"])
assert len(focus) >= 3, f"Stage {stage['name']} needs more review focus items"
def test_required_tools_complete(self):
manifest = load_pipeline("animated-explainer")
tools = get_required_tools(manifest)
expected = {"tts_selector", "image_selector", "video_compose", "audio_mixer"}
for t in expected:
assert t in tools, f"Missing required tool: {t}"
def test_creative_stages_require_human_approval(self):
manifest = load_pipeline("animated-explainer")
approval_stages = {"proposal", "script", "scene_plan", "publish"}
for stage in manifest["stages"]:
if stage["name"] in approval_stages:
assert stage.get("human_approval_default") is True, (
f"Stage {stage['name']} should require human approval"
)
def test_listed(self):
assert "animated-explainer" in list_pipelines()
# ---- Style Playbooks ----
class TestStylePlaybooks:
def test_all_listed(self):
playbooks = list_playbooks()
assert "clean-professional" in playbooks
assert "flat-motion-graphics" in playbooks
assert "minimalist-diagram" in playbooks
@pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
def test_loads_and_validates(self, name):
pb = load_playbook(name)
assert pb["identity"]["name"]
assert pb["identity"]["category"]
@pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
def test_has_required_sections(self, name):
pb = load_playbook(name)
assert "visual_language" in pb
assert "typography" in pb
assert "motion" in pb
assert "audio" in pb
assert "asset_generation" in pb
assert "quality_rules" in pb
assert len(pb["quality_rules"]) >= 3
@pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
def test_color_palette_complete(self, name):
pb = load_playbook(name)
palette = pb["visual_language"]["color_palette"]
assert "primary" in palette
assert "accent" in palette
assert "background" in palette
assert "text" in palette
@pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
def test_pacing_rules_present(self, name):
pb = load_playbook(name)
pacing = pb["motion"]["pacing_rules"]
assert "min_scene_hold_seconds" in pacing
assert "max_scene_hold_seconds" in pacing
def test_compatible_with_manifest(self):
manifest = load_pipeline("animated-explainer")
available = list_playbooks()
for name in manifest.get("compatible_playbooks", []):
assert name in available, f"Manifest references unavailable playbook: {name}"
# ---- Skills Existence ----
class TestSkillsExist:
SKILLS_DIR = PROJECT_ROOT / "skills"
@pytest.mark.parametrize("skill_path", [
"pipelines/explainer/idea-director.md",
"pipelines/explainer/script-director.md",
"pipelines/explainer/scene-director.md",
"pipelines/explainer/asset-director.md",
"pipelines/explainer/edit-director.md",
"pipelines/explainer/compose-director.md",
"pipelines/explainer/publish-director.md",
])
def test_director_skills_exist(self, skill_path):
full_path = self.SKILLS_DIR / skill_path
assert full_path.exists(), f"Missing director skill: {skill_path}"
content = full_path.read_text(encoding="utf-8")
assert len(content) > 500, f"Skill too short to be useful: {skill_path}"
@pytest.mark.parametrize("skill_path", [
"meta/reviewer.md",
"meta/checkpoint-protocol.md",
"meta/skill-creator.md",
])
def test_meta_skills_exist(self, skill_path):
full_path = self.SKILLS_DIR / skill_path
assert full_path.exists(), f"Missing meta skill: {skill_path}"
content = full_path.read_text(encoding="utf-8")
assert len(content) > 500, f"Skill too short to be useful: {skill_path}"
@pytest.mark.parametrize("skill_path", [
"pipelines/explainer/idea-director.md",
"pipelines/explainer/script-director.md",
"pipelines/explainer/scene-director.md",
"pipelines/explainer/asset-director.md",
"pipelines/explainer/edit-director.md",
"pipelines/explainer/compose-director.md",
"pipelines/explainer/publish-director.md",
])
def test_director_skills_have_required_sections(self, skill_path):
content = (self.SKILLS_DIR / skill_path).read_text(encoding="utf-8")
assert "## When to Use" in content
assert "## Process" in content or "## Protocol" in content
assert "Self-Evaluate" in content or "self-evaluate" in content.lower()
@pytest.mark.parametrize("skill_path", [
"meta/reviewer.md",
"meta/checkpoint-protocol.md",
"meta/skill-creator.md",
])
def test_meta_skills_have_required_sections(self, skill_path):
content = (self.SKILLS_DIR / skill_path).read_text(encoding="utf-8")
assert "## When to Use" in content
assert "## Protocol" in content or "## Process" in content
# ---- Remotion Scaffold ----
class TestRemotionScaffold:
REMOTION_DIR = PROJECT_ROOT / "remotion-composer"
def test_package_json_exists(self):
assert (self.REMOTION_DIR / "package.json").exists()
def test_entry_point_exists(self):
assert (self.REMOTION_DIR / "src" / "index.tsx").exists()
def test_root_composition_exists(self):
assert (self.REMOTION_DIR / "src" / "Root.tsx").exists()
def test_explainer_component_exists(self):
assert (self.REMOTION_DIR / "src" / "Explainer.tsx").exists()
def test_text_card_component_exists(self):
assert (self.REMOTION_DIR / "src" / "components" / "TextCard.tsx").exists()
def test_stat_card_component_exists(self):
assert (self.REMOTION_DIR / "src" / "components" / "StatCard.tsx").exists()
# ---- Video Compose Operations ----
class TestVideoComposeOperations:
def test_render_operation_exists(self):
from tools.video.video_compose import VideoCompose
tool = VideoCompose()
ops = tool.input_schema["properties"]["operation"]["enum"]
assert "render" in ops
assert "remotion_render" in ops
def test_render_rejects_missing_inputs(self):
from tools.video.video_compose import VideoCompose
tool = VideoCompose()
result = tool.execute({"operation": "render"})
assert not result.success
assert "edit_decisions" in result.error
View File
View File
View File
@@ -0,0 +1,67 @@
{
"name": "talking_head_basic",
"pipeline_type": "talking-head",
"description": "Basic talking-head pipeline: 15s vertical clip with burned subtitles",
"inputs": {
"footage_path": "C:/Users/ishan/Documents/SocialMedia/IshanAIVideos/pipeline/2026-03-08/seg1_intro.mp4",
"model_size": "base",
"target_platform": "youtube"
},
"expected_artifacts": {
"brief": {
"version": "1.0",
"required_fields": ["title", "hook", "key_points", "tone", "style", "target_platform", "target_duration_seconds"]
},
"script": {
"version": "1.0",
"required_fields": ["title", "total_duration_seconds", "sections"],
"assertions": {
"sections_not_empty": true,
"total_duration_range": [10, 20]
}
},
"scene_plan": {
"version": "1.0",
"required_fields": ["scenes"],
"assertions": {
"scenes_not_empty": true,
"all_scenes_are_talking_head": true
}
},
"asset_manifest": {
"version": "1.0",
"required_fields": ["assets"],
"assertions": {
"has_subtitle_asset": true,
"has_source_footage_asset": true
}
},
"edit_decisions": {
"version": "1.0",
"required_fields": ["cuts"],
"assertions": {
"subtitles_enabled": true,
"cuts_cover_full_duration": true
}
},
"render_report": {
"version": "1.0",
"required_fields": ["outputs"],
"assertions": {
"output_file_exists": true,
"output_is_playable": true,
"resolution_matches_source": true
}
},
"publish_log": {
"version": "1.0",
"required_fields": ["entries"],
"assertions": {
"has_at_least_one_entry": true
}
}
},
"eval_mode": "stochastic",
"tolerance": 0.1,
"tags": ["phase1", "talking_head", "subtitles", "vertical"]
}
+154
View File
@@ -0,0 +1,154 @@
"""Replay evaluation harness.
Provides the scaffold for replaying golden scenarios from saved checkpoints
and comparing outputs against reference results. Supports both deterministic
(exact match) and stochastic (threshold-based) evaluation paths.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Optional
class EvalMode(str, Enum):
DETERMINISTIC = "deterministic"
STOCHASTIC = "stochastic"
@dataclass
class GoldenScenario:
"""A known-good input -> output pair for regression testing."""
name: str
pipeline_type: str
inputs: dict[str, Any]
expected_artifacts: dict[str, Any]
eval_mode: EvalMode = EvalMode.DETERMINISTIC
tolerance: float = 0.0 # For stochastic comparisons
tags: list[str] = field(default_factory=list)
@classmethod
def load(cls, path: Path) -> "GoldenScenario":
with open(path) as f:
data = json.load(f)
return cls(
name=data["name"],
pipeline_type=data["pipeline_type"],
inputs=data["inputs"],
expected_artifacts=data["expected_artifacts"],
eval_mode=EvalMode(data.get("eval_mode", "deterministic")),
tolerance=data.get("tolerance", 0.0),
tags=data.get("tags", []),
)
def save(self, path: Path) -> None:
data = {
"name": self.name,
"pipeline_type": self.pipeline_type,
"inputs": self.inputs,
"expected_artifacts": self.expected_artifacts,
"eval_mode": self.eval_mode.value,
"tolerance": self.tolerance,
"tags": self.tags,
}
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(data, f, indent=2)
@dataclass
class EvalResult:
"""Result of evaluating a single scenario."""
scenario_name: str
passed: bool
details: dict[str, Any] = field(default_factory=dict)
errors: list[str] = field(default_factory=list)
# Type for custom comparison functions
Comparator = Callable[[Any, Any, float], bool]
def default_comparator(expected: Any, actual: Any, tolerance: float) -> bool:
"""Default comparator: exact match for deterministic, threshold for stochastic."""
if tolerance == 0.0:
return expected == actual
if isinstance(expected, (int, float)) and isinstance(actual, (int, float)):
return abs(expected - actual) <= tolerance
return expected == actual
class ReplayHarness:
"""Loads golden scenarios and replays them through the pipeline."""
def __init__(
self,
scenarios_dir: Optional[Path] = None,
outputs_dir: Optional[Path] = None,
) -> None:
self.scenarios_dir = scenarios_dir or Path("tests/eval/golden_scenarios")
self.outputs_dir = outputs_dir or Path("tests/eval/golden_outputs")
self._comparators: dict[str, Comparator] = {}
def register_comparator(self, artifact_name: str, comparator: Comparator) -> None:
"""Register a custom comparator for a specific artifact type."""
self._comparators[artifact_name] = comparator
def load_scenarios(self, tags: Optional[list[str]] = None) -> list[GoldenScenario]:
"""Load all golden scenarios, optionally filtering by tags."""
scenarios = []
for path in self.scenarios_dir.glob("*.json"):
scenario = GoldenScenario.load(path)
if tags and not any(t in scenario.tags for t in tags):
continue
scenarios.append(scenario)
return scenarios
def evaluate(
self,
scenario: GoldenScenario,
actual_artifacts: dict[str, Any],
) -> EvalResult:
"""Compare actual artifacts against expected for a scenario."""
errors = []
details: dict[str, Any] = {}
for artifact_name, expected in scenario.expected_artifacts.items():
actual = actual_artifacts.get(artifact_name)
if actual is None:
errors.append(f"Missing artifact: {artifact_name}")
continue
comparator = self._comparators.get(artifact_name, default_comparator)
passed = comparator(expected, actual, scenario.tolerance)
details[artifact_name] = {
"passed": passed,
"expected_type": type(expected).__name__,
"actual_type": type(actual).__name__,
}
if not passed:
errors.append(f"Mismatch in {artifact_name}")
return EvalResult(
scenario_name=scenario.name,
passed=len(errors) == 0,
details=details,
errors=errors,
)
def run_all(
self,
runner: Callable[[GoldenScenario], dict[str, Any]],
tags: Optional[list[str]] = None,
) -> list[EvalResult]:
"""Run all scenarios through a runner function and evaluate results."""
scenarios = self.load_scenarios(tags)
results = []
for scenario in scenarios:
actual = runner(scenario)
result = self.evaluate(scenario, actual)
results.append(result)
return results
View File
+72
View File
@@ -0,0 +1,72 @@
# Quality Validation Plan — Phase 3.5 + G3.11
## Purpose
Run every tool with real API keys, inspect outputs (see images, listen to audio, watch video), find gaps, fix them. This is the gate before calling Phase 3.5 "Verified."
## Test Scripts (all ready to run)
| Script | Tools Tested | API Keys Used | Est. Cost |
|--------|-------------|---------------|-----------|
| `test_01_tts.py` | `elevenlabs_tts` (ElevenLabs) | ELEVENLABS_API_KEY | ~$0.02 |
| `test_02_image_gen.py` | `image_gen` (DALL-E 3 + FLUX) | OPENAI_API_KEY, FAL_AI_API_KEY | ~$0.15 |
| `test_03_music.py` | `music_gen` (ElevenLabs) | ELEVENLABS_API_KEY | ~$0.10 |
| `test_04_audio_mix.py` | `audio_mixer` | None (ffmpeg only) | $0 |
| `test_05_video_compose.py` | `video_compose` | None (ffmpeg only) | $0 |
| `test_06_video_stitch.py` | `video_stitch` | None (ffmpeg only) | $0 |
| `test_07_playbook_intelligence.py` | `playbook_loader.py` functions | None (pure Python) | $0 |
| `test_08_end_to_end.py` | Full animated-explainer pipeline | None (ffmpeg fixtures) | $0 |
## Inspection Protocol
For each output:
1. **Audio files**: Use `ffprobe` for format/duration/channels, then LISTEN (play in media player or use Whisper to verify content matches prompt)
2. **Image files**: Use `ffprobe` for dimensions, then VIEW (open image, check composition, text readability, style match)
3. **Video files**: Use `ffprobe` for resolution/fps/duration/codec, then WATCH (check A/V sync, transitions, subtitle timing)
4. **Design intelligence**: Run against all 3 playbooks, verify contrast ratios match manual calculation, check CVD warnings are accurate
## Known Risk Areas
| Area | Risk | How to Validate |
|------|------|-----------------|
| TTS voice selection | Default voice may not match playbook mood | Test with multiple voice IDs, compare against playbook `voice_style` |
| Image gen consistency | DALL-E/FLUX outputs vary wildly per prompt | Test with playbook `image_prompt_prefix` prepended |
| Music duration alignment | Music may not match narration duration | Compare `music.duration` vs `tts.duration`, check padding/looping |
| Audio ducking timing | Ducking may cut music too aggressively | Inspect waveform: music should duck ~6dB under speech, recover smoothly |
| Video stitch transitions | Crossfade may flicker with mismatched codecs | Test with both matching and mismatched clips, check `auto_normalize` |
| Subtitle burn-in | Font size/position may clip on mobile formats | Test with 9:16 (TikTok) and 16:9 (YouTube) profiles |
| Remotion render | Components may fail with real data | Build a test composition with all 8 components, render at 1080p |
| Playbook contrast | Edge cases in dark-on-dark or light-on-light themes | Test with all 3 playbooks + a deliberately low-contrast custom one |
## Run Order
```bash
cd C:/Users/ishan/Documents/OpenMontage
# Phase 1: Individual tools (can run in parallel)
python tests/qa/test_01_tts.py
python tests/qa/test_02_image_gen.py
python tests/qa/test_03_music.py
# Phase 2: Composition (depends on Phase 1 outputs)
python tests/qa/test_04_audio_mix.py
python tests/qa/test_05_video_compose.py
python tests/qa/test_06_video_stitch.py
# Phase 3: Intelligence validation (no API calls)
python tests/qa/test_07_playbook_intelligence.py
# Phase 4: Full pipeline
python tests/qa/test_08_end_to_end.py
```
## Success Criteria
- [ ] All 3 TTS samples: clear speech, correct content, no artifacts, ≥44.1kHz
- [ ] All 4 images: match prompt intent, correct dimensions, no watermarks, good composition
- [ ] Both music tracks: match mood prompt, correct duration (±2s), no abrupt cuts
- [ ] Audio mix: speech clearly above music, ducking smooth, no clipping
- [ ] Video compose: A/V sync within 50ms, correct resolution, playable in VLC
- [ ] Video stitch: smooth transitions, no frame drops, PIP correctly positioned
- [ ] Playbook intelligence: all 3 playbooks pass a11y, contrast ratios within 0.1 of manual calc
- [ ] End-to-end: 60-second explainer renders without errors, all stages checkpoint correctly
+80
View File
@@ -0,0 +1,80 @@
#!/usr/bin/env python3
"""QA Test 01: TTS voice generation via ElevenLabs."""
import sys, os, json, time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.env_loader import load_env
load_env()
from tools.audio.elevenlabs_tts import ElevenLabsTTS
OUT = os.path.join(os.path.dirname(__file__), "output")
os.makedirs(OUT, exist_ok=True)
tool = ElevenLabsTTS()
print(f"Tool status: {tool.get_status()}")
# Test 1: Short narration
print("\n--- Test 1: Short narration ---")
r1 = tool.execute({
"text": "Welcome to OpenMontage. Let's build something amazing together.",
"output_path": os.path.join(OUT, "tts_short.mp3"),
})
print(f"Success: {r1.success}, Cost: ${r1.cost_usd:.4f}, Duration: {r1.duration_seconds:.2f}s")
if r1.error: print(f"Error: {r1.error}")
if r1.artifacts: print(f"Artifacts: {r1.artifacts}")
# Test 2: Longer paragraph with technical content
print("\n--- Test 2: Technical narration ---")
r2 = tool.execute({
"text": (
"Quantum computing leverages quantum mechanical phenomena like superposition and entanglement "
"to process information in fundamentally different ways than classical computers. "
"While a classical bit can be either zero or one, a quantum bit, or qubit, can exist in "
"a superposition of both states simultaneously. This allows quantum computers to explore "
"many possible solutions at once, making them exceptionally powerful for certain types of problems."
),
"output_path": os.path.join(OUT, "tts_technical.mp3"),
})
print(f"Success: {r2.success}, Cost: ${r2.cost_usd:.4f}, Duration: {r2.duration_seconds:.2f}s")
if r2.error: print(f"Error: {r2.error}")
if r2.artifacts: print(f"Artifacts: {r2.artifacts}")
# Test 3: Emotional / storytelling narration
print("\n--- Test 3: Storytelling narration ---")
r3 = tool.execute({
"text": (
"Picture this. You wake up one morning, check your phone, and discover that overnight, "
"your side project went viral. Thousands of people are using it. Messages are flooding in. "
"This isn't a dream. This is what happens when you build something people actually need."
),
"output_path": os.path.join(OUT, "tts_story.mp3"),
})
print(f"Success: {r3.success}, Cost: ${r3.cost_usd:.4f}, Duration: {r3.duration_seconds:.2f}s")
if r3.error: print(f"Error: {r3.error}")
if r3.artifacts: print(f"Artifacts: {r3.artifacts}")
# Probe outputs with ffprobe
import subprocess
for name in ["tts_short.mp3", "tts_technical.mp3", "tts_story.mp3"]:
path = os.path.join(OUT, name)
if os.path.exists(path):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", path],
capture_output=True, text=True
)
info = json.loads(probe.stdout)
fmt = info.get("format", {})
streams = info.get("streams", [{}])
audio = streams[0] if streams else {}
print(f"\n[{name}] Duration: {fmt.get('duration', '?')}s, "
f"Sample rate: {audio.get('sample_rate', '?')}Hz, "
f"Channels: {audio.get('channels', '?')}, "
f"Codec: {audio.get('codec_name', '?')}, "
f"Size: {os.path.getsize(path)} bytes")
else:
print(f"\n[{name}] FILE NOT FOUND")
print("\n=== TTS TEST COMPLETE ===")
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env python3
"""QA Test 02: Image generation via OpenAI (DALL-E 3) and fal.ai (FLUX)."""
import sys, os, json, time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.env_loader import load_env
load_env()
from tools.graphics.image_gen import ImageGen
OUT = os.path.join(os.path.dirname(__file__), "output")
os.makedirs(OUT, exist_ok=True)
tool = ImageGen()
print(f"Tool status: {tool.get_status()}")
# Test 1: DALL-E — professional diagram
print("\n--- Test 1: DALL-E professional diagram ---")
r1 = tool.execute({
"prompt": "A clean, professional infographic showing the 5 stages of a video production pipeline: Idea, Script, Assets, Edit, Publish. Flat design, blue and amber color scheme, white background, no text.",
"width": 1280,
"height": 720,
"provider": "openai",
"output_path": os.path.join(OUT, "img_dalle_diagram.png"),
})
print(f"Success: {r1.success}, Cost: ${r1.cost_usd:.4f}")
if r1.error: print(f"Error: {r1.error}")
if r1.artifacts: print(f"Artifacts: {r1.artifacts}")
# Test 2: DALL-E — cinematic scene
print("\n--- Test 2: DALL-E cinematic scene ---")
r2 = tool.execute({
"prompt": "A futuristic control room with holographic displays showing data visualizations, cinematic lighting, wide angle, film grain, warm tones",
"width": 1280,
"height": 720,
"provider": "openai",
"output_path": os.path.join(OUT, "img_dalle_cinematic.png"),
})
print(f"Success: {r2.success}, Cost: ${r2.cost_usd:.4f}")
if r2.error: print(f"Error: {r2.error}")
if r2.artifacts: print(f"Artifacts: {r2.artifacts}")
# Test 3: FLUX via fal.ai — abstract tech
print("\n--- Test 3: FLUX abstract tech ---")
r3 = tool.execute({
"prompt": "Abstract visualization of neural network connections, glowing nodes and edges, dark background, neon blue and purple, high detail, 8k render",
"width": 1280,
"height": 720,
"provider": "flux",
"output_path": os.path.join(OUT, "img_flux_abstract.png"),
})
print(f"Success: {r3.success}, Cost: ${r3.cost_usd:.4f}")
if r3.error: print(f"Error: {r3.error}")
if r3.artifacts: print(f"Artifacts: {r3.artifacts}")
# Test 4: FLUX — character/mascot
print("\n--- Test 4: FLUX character illustration ---")
r4 = tool.execute({
"prompt": "Friendly robot mascot character, simple geometric design, holding a film clapboard, isometric view, clean white background, flat illustration style",
"width": 1024,
"height": 1024,
"provider": "flux",
"output_path": os.path.join(OUT, "img_flux_mascot.png"),
})
print(f"Success: {r4.success}, Cost: ${r4.cost_usd:.4f}")
if r4.error: print(f"Error: {r4.error}")
if r4.artifacts: print(f"Artifacts: {r4.artifacts}")
# Probe outputs
import subprocess
for name in ["img_dalle_diagram.png", "img_dalle_cinematic.png", "img_flux_abstract.png", "img_flux_mascot.png"]:
path = os.path.join(OUT, name)
if os.path.exists(path):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", path],
capture_output=True, text=True
)
info = json.loads(probe.stdout)
stream = info.get("streams", [{}])[0]
print(f"\n[{name}] {stream.get('width', '?')}x{stream.get('height', '?')}, "
f"Format: {stream.get('codec_name', '?')}, "
f"Size: {os.path.getsize(path)} bytes")
else:
print(f"\n[{name}] FILE NOT FOUND")
print("\n=== IMAGE GEN TEST COMPLETE ===")
+62
View File
@@ -0,0 +1,62 @@
#!/usr/bin/env python3
"""QA Test 03: Music generation via ElevenLabs."""
import sys, os, json
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.env_loader import load_env
load_env()
from tools.audio.music_gen import MusicGen
OUT = os.path.join(os.path.dirname(__file__), "output")
os.makedirs(OUT, exist_ok=True)
tool = MusicGen()
print(f"Tool status: {tool.get_status()}")
# Test 1: Upbeat tech background
print("\n--- Test 1: Upbeat tech background ---")
r1 = tool.execute({
"prompt": "Upbeat electronic background music, 120 BPM, energetic but not overwhelming, suitable for a tech explainer video",
"duration_seconds": 30,
"output_path": os.path.join(OUT, "music_upbeat.mp3"),
})
print(f"Success: {r1.success}, Cost: ${r1.cost_usd:.4f}")
if r1.error: print(f"Error: {r1.error}")
if r1.artifacts: print(f"Artifacts: {r1.artifacts}")
# Test 2: Calm ambient
print("\n--- Test 2: Calm ambient ---")
r2 = tool.execute({
"prompt": "Calm ambient background music, soft piano and strings, 80 BPM, reflective mood, suitable for documentary narration",
"duration_seconds": 30,
"output_path": os.path.join(OUT, "music_calm.mp3"),
})
print(f"Success: {r2.success}, Cost: ${r2.cost_usd:.4f}")
if r2.error: print(f"Error: {r2.error}")
if r2.artifacts: print(f"Artifacts: {r2.artifacts}")
# Probe outputs
import subprocess
for name in ["music_upbeat.mp3", "music_calm.mp3"]:
path = os.path.join(OUT, name)
if os.path.exists(path):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", path],
capture_output=True, text=True
)
info = json.loads(probe.stdout)
fmt = info.get("format", {})
streams = info.get("streams", [{}])
audio = streams[0] if streams else {}
print(f"\n[{name}] Duration: {fmt.get('duration', '?')}s, "
f"Sample rate: {audio.get('sample_rate', '?')}Hz, "
f"Channels: {audio.get('channels', '?')}, "
f"Codec: {audio.get('codec_name', '?')}, "
f"Size: {os.path.getsize(path)} bytes")
else:
print(f"\n[{name}] FILE NOT FOUND")
print("\n=== MUSIC GEN TEST COMPLETE ===")
+153
View File
@@ -0,0 +1,153 @@
#!/usr/bin/env python3
"""QA Test 04: Audio mixing — mix TTS + music with ducking, verify levels.
Depends on test_01 and test_03 outputs (TTS + music files).
If those don't exist, generates minimal test fixtures via ffmpeg.
"""
import sys, os, json, subprocess
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.env_loader import load_env
load_env()
from tools.audio.audio_mixer import AudioMixer
OUT = os.path.join(os.path.dirname(__file__), "output")
os.makedirs(OUT, exist_ok=True)
# --- Fixture generation (if test_01/test_03 outputs don't exist) ---
SPEECH_FILE = os.path.join(OUT, "tts_short.mp3")
MUSIC_FILE = os.path.join(OUT, "music_calm.mp3")
def generate_fixture(path, description, duration=5):
"""Generate a minimal audio fixture with ffmpeg if the file doesn't exist."""
if os.path.exists(path):
print(f" [fixture] Using existing: {path}")
return
print(f" [fixture] Generating {description}: {path}")
# Sine wave for speech stand-in, pink noise for music stand-in
if "speech" in description or "tts" in description:
src = f"sine=frequency=440:duration={duration}"
else:
src = f"anoisesrc=d={duration}:c=pink"
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i", src, "-ar", "44100", "-ac", "1", path],
capture_output=True, check=True,
)
generate_fixture(SPEECH_FILE, "speech/tts fixture", duration=8)
generate_fixture(MUSIC_FILE, "music fixture", duration=15)
# --- Tool setup ---
tool = AudioMixer()
print(f"Tool status: {tool.get_status()}")
# --- Test 1: Basic mix (speech + music, no ducking) ---
print("\n--- Test 1: Basic mix (speech + music) ---")
r1 = tool.execute({
"operation": "mix",
"tracks": [
{"path": SPEECH_FILE, "role": "speech", "volume": 1.0},
{"path": MUSIC_FILE, "role": "music", "volume": 0.3},
],
"normalize": True,
"output_path": os.path.join(OUT, "mix_basic.wav"),
})
print(f"Success: {r1.success}, Duration: {r1.duration_seconds:.2f}s")
if r1.error: print(f"Error: {r1.error}")
if r1.artifacts: print(f"Artifacts: {r1.artifacts}")
# --- Test 2: Mix with fades ---
print("\n--- Test 2: Mix with fades ---")
r2 = tool.execute({
"operation": "mix",
"tracks": [
{"path": SPEECH_FILE, "role": "speech", "volume": 1.0, "fade_in_seconds": 0.5},
{"path": MUSIC_FILE, "role": "music", "volume": 0.25, "fade_in_seconds": 1.0, "fade_out_seconds": 2.0},
],
"normalize": True,
"output_path": os.path.join(OUT, "mix_fades.wav"),
})
print(f"Success: {r2.success}, Duration: {r2.duration_seconds:.2f}s")
if r2.error: print(f"Error: {r2.error}")
if r2.artifacts: print(f"Artifacts: {r2.artifacts}")
# --- Test 3: Ducking (sidechain compress music under speech) ---
print("\n--- Test 3: Ducking ---")
r3 = tool.execute({
"operation": "duck",
"tracks": [
{"path": SPEECH_FILE, "role": "speech"},
{"path": MUSIC_FILE, "role": "music"},
],
"ducking": {
"enabled": True,
"music_volume_during_speech": 0.15,
"attack_ms": 200,
"release_ms": 500,
},
"output_path": os.path.join(OUT, "mix_ducked.wav"),
})
print(f"Success: {r3.success}, Duration: {r3.duration_seconds:.2f}s")
if r3.error: print(f"Error: {r3.error}")
if r3.artifacts: print(f"Artifacts: {r3.artifacts}")
# --- Test 4: Mix with delayed music start ---
print("\n--- Test 4: Delayed music start ---")
r4 = tool.execute({
"operation": "mix",
"tracks": [
{"path": SPEECH_FILE, "role": "speech", "volume": 1.0},
{"path": MUSIC_FILE, "role": "music", "volume": 0.2, "start_seconds": 3.0},
],
"normalize": False,
"output_path": os.path.join(OUT, "mix_delayed.wav"),
})
print(f"Success: {r4.success}, Duration: {r4.duration_seconds:.2f}s")
if r4.error: print(f"Error: {r4.error}")
if r4.artifacts: print(f"Artifacts: {r4.artifacts}")
# --- Probe all outputs ---
print("\n--- Output inspection ---")
for name in ["mix_basic.wav", "mix_fades.wav", "mix_ducked.wav", "mix_delayed.wav"]:
path = os.path.join(OUT, name)
if os.path.exists(path):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", path],
capture_output=True, text=True,
)
info = json.loads(probe.stdout)
fmt = info.get("format", {})
streams = info.get("streams", [{}])
audio = streams[0] if streams else {}
print(f"\n[{name}] Duration: {fmt.get('duration', '?')}s, "
f"Sample rate: {audio.get('sample_rate', '?')}Hz, "
f"Channels: {audio.get('channels', '?')}, "
f"Codec: {audio.get('codec_name', '?')}, "
f"Size: {os.path.getsize(path)} bytes")
# Check for clipping via loudnorm stats
loud = subprocess.run(
["ffmpeg", "-i", path, "-af", "loudnorm=print_format=json", "-f", "null", "-"],
capture_output=True, text=True,
)
# Parse loudnorm JSON from stderr (ffmpeg writes it there)
stderr = loud.stderr
json_start = stderr.rfind("{")
json_end = stderr.rfind("}") + 1
if json_start >= 0 and json_end > json_start:
try:
loudness = json.loads(stderr[json_start:json_end])
print(f" Loudness: I={loudness.get('input_i', '?')} LUFS, "
f"TP={loudness.get('input_tp', '?')} dBTP, "
f"LRA={loudness.get('input_lra', '?')} LU")
except json.JSONDecodeError:
print(" (Could not parse loudness stats)")
else:
print(f"\n[{name}] FILE NOT FOUND")
print("\n=== AUDIO MIX TEST COMPLETE ===")
+230
View File
@@ -0,0 +1,230 @@
#!/usr/bin/env python3
"""QA Test 05: Video composition — image + mixed audio → video, verify A/V sync.
Creates a video from static images with audio and optional subtitles.
Uses ffmpeg-generated fixtures if prior test outputs don't exist.
"""
import sys, os, json, subprocess
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.env_loader import load_env
load_env()
from tools.video.video_compose import VideoCompose
OUT = os.path.join(os.path.dirname(__file__), "output")
os.makedirs(OUT, exist_ok=True)
# --- Fixture generation ---
def ensure_image(path, width=1280, height=720, color="blue"):
"""Generate a test image with ffmpeg if it doesn't exist."""
if os.path.exists(path):
print(f" [fixture] Using existing: {path}")
return
print(f" [fixture] Generating {color} image: {path}")
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i",
f"color=c={color}:s={width}x{height}:d=1",
"-frames:v", "1", path],
capture_output=True, check=True,
)
def ensure_video(path, duration=5, width=1280, height=720, color="blue"):
"""Generate a test video clip with ffmpeg if it doesn't exist.
Uses forced keyframes every 1s so copy-mode trimming retains the video stream.
"""
if os.path.exists(path):
print(f" [fixture] Using existing: {path}")
return
print(f" [fixture] Generating {duration}s {color} video: {path}")
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i",
f"color=c={color}:s={width}x{height}:d={duration}:r=30",
"-f", "lavfi", "-i", f"sine=frequency=440:duration={duration}",
"-c:v", "libx264", "-crf", "23", "-pix_fmt", "yuv420p",
"-g", "30", "-keyint_min", "30",
"-c:a", "aac", "-shortest", path],
capture_output=True, check=True,
)
def ensure_audio(path, duration=5):
"""Generate a test audio file if it doesn't exist."""
if os.path.exists(path):
print(f" [fixture] Using existing: {path}")
return
print(f" [fixture] Generating {duration}s audio: {path}")
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i",
f"sine=frequency=440:duration={duration}",
"-ar", "44100", "-ac", "2", path],
capture_output=True, check=True,
)
def ensure_subtitle(path):
"""Write a minimal SRT subtitle file."""
if os.path.exists(path):
print(f" [fixture] Using existing: {path}")
return
print(f" [fixture] Generating subtitle: {path}")
with open(path, "w", encoding="utf-8") as f:
f.write("1\n00:00:00,000 --> 00:00:03,000\nWelcome to OpenMontage\n\n")
f.write("2\n00:00:03,000 --> 00:00:06,000\nBuilding amazing videos with AI\n\n")
f.write("3\n00:00:06,000 --> 00:00:10,000\nLet's see what we can create\n\n")
# Create fixtures
CLIP_A = os.path.join(OUT, "compose_clip_a.mp4")
CLIP_B = os.path.join(OUT, "compose_clip_b.mp4")
AUDIO_MIX = os.path.join(OUT, "compose_audio.wav")
SUBTITLE = os.path.join(OUT, "compose_subs.srt")
# Use 10s clips so -c copy has keyframe headroom (5s clips lose video stream)
ensure_video(CLIP_A, duration=10, color="darkblue")
ensure_video(CLIP_B, duration=10, color="darkgreen")
ensure_audio(AUDIO_MIX, duration=10)
ensure_subtitle(SUBTITLE)
# --- Tool setup ---
tool = VideoCompose()
print(f"Tool status: {tool.get_status()}")
# --- Test 1: Compose with cuts + audio ---
print("\n--- Test 1: Compose from edit_decisions ---")
r1 = tool.execute({
"operation": "compose",
"edit_decisions": {
"cuts": [
{"source": CLIP_A, "in_seconds": 0, "out_seconds": 5},
{"source": CLIP_B, "in_seconds": 0, "out_seconds": 5},
],
},
"audio_path": AUDIO_MIX,
"output_path": os.path.join(OUT, "compose_basic.mp4"),
})
print(f"Success: {r1.success}, Duration: {r1.duration_seconds:.2f}s")
if r1.error: print(f"Error: {r1.error}")
if r1.artifacts: print(f"Artifacts: {r1.artifacts}")
# --- Test 2: Compose with subtitles ---
print("\n--- Test 2: Compose with subtitles ---")
r2 = tool.execute({
"operation": "compose",
"edit_decisions": {
"cuts": [
{"source": CLIP_A, "in_seconds": 0, "out_seconds": 5},
{"source": CLIP_B, "in_seconds": 0, "out_seconds": 5},
],
},
"audio_path": AUDIO_MIX,
"subtitle_path": SUBTITLE,
"subtitle_style": {
"font": "Arial",
"font_size": 24,
"primary_color": "&HFFFFFF",
"outline_color": "&H000000",
"outline_width": 2,
"margin_v": 40,
},
"output_path": os.path.join(OUT, "compose_subtitled.mp4"),
})
print(f"Success: {r2.success}, Duration: {r2.duration_seconds:.2f}s")
if r2.error: print(f"Error: {r2.error}")
if r2.artifacts: print(f"Artifacts: {r2.artifacts}")
# --- Test 3: Burn subtitles onto existing video ---
print("\n--- Test 3: Burn subtitles standalone ---")
r3 = tool.execute({
"operation": "burn_subtitles",
"input_path": CLIP_A,
"subtitle_path": SUBTITLE,
"subtitle_style": {
"font": "Arial",
"font_size": 20,
"bold": True,
},
"output_path": os.path.join(OUT, "compose_burn_subs.mp4"),
})
print(f"Success: {r3.success}, Duration: {r3.duration_seconds:.2f}s")
if r3.error: print(f"Error: {r3.error}")
if r3.artifacts: print(f"Artifacts: {r3.artifacts}")
# --- Test 4: Encode with media profile ---
print("\n--- Test 4: Re-encode with profile ---")
r4 = tool.execute({
"operation": "encode",
"input_path": CLIP_A,
"profile": "YOUTUBE_LANDSCAPE",
"crf": 20,
"preset": "fast",
"output_path": os.path.join(OUT, "compose_encoded.mp4"),
})
print(f"Success: {r4.success}, Duration: {r4.duration_seconds:.2f}s")
if r4.error: print(f"Error: {r4.error}")
if r4.artifacts: print(f"Artifacts: {r4.artifacts}")
# --- Test 5: Overlay ---
print("\n--- Test 5: Overlay image on video ---")
OVERLAY_IMG = os.path.join(OUT, "compose_overlay.png")
ensure_image(OVERLAY_IMG, width=200, height=200, color="red")
r5 = tool.execute({
"operation": "overlay",
"input_path": CLIP_A,
"overlays": [
{
"asset_path": OVERLAY_IMG,
"x": 50, "y": 50,
"width": 150, "height": 150,
"start_seconds": 1,
"end_seconds": 4,
"opacity": 0.8,
},
],
"output_path": os.path.join(OUT, "compose_overlay.mp4"),
})
print(f"Success: {r5.success}, Duration: {r5.duration_seconds:.2f}s")
if r5.error: print(f"Error: {r5.error}")
if r5.artifacts: print(f"Artifacts: {r5.artifacts}")
# --- Probe all outputs ---
print("\n--- Output inspection ---")
outputs = [
"compose_basic.mp4",
"compose_subtitled.mp4",
"compose_burn_subs.mp4",
"compose_encoded.mp4",
"compose_overlay.mp4",
]
for name in outputs:
path = os.path.join(OUT, name)
if os.path.exists(path):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", path],
capture_output=True, text=True,
)
info = json.loads(probe.stdout)
fmt = info.get("format", {})
video = {}
audio = {}
for s in info.get("streams", []):
if s.get("codec_type") == "video" and not video:
video = s
elif s.get("codec_type") == "audio" and not audio:
audio = s
print(f"\n[{name}]"
f" Duration: {fmt.get('duration', '?')}s"
f" | Video: {video.get('width', '?')}x{video.get('height', '?')}"
f" {video.get('codec_name', '?')}@{video.get('r_frame_rate', '?')}fps"
f" | Audio: {audio.get('codec_name', '?')}"
f" {audio.get('sample_rate', '?')}Hz"
f" {audio.get('channels', '?')}ch"
f" | Size: {os.path.getsize(path)} bytes")
else:
print(f"\n[{name}] FILE NOT FOUND")
print("\n=== VIDEO COMPOSE TEST COMPLETE ===")
+225
View File
@@ -0,0 +1,225 @@
#!/usr/bin/env python3
"""QA Test 06: Video stitch — sequential concat, crossfade, fade, spatial PIP.
Tests the VideoStitch tool with both matching and mismatched clips.
Generates fixtures via ffmpeg — no API keys needed.
"""
import sys, os, json, subprocess
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from lib.env_loader import load_env
load_env()
from tools.video.video_stitch import VideoStitch
OUT = os.path.join(os.path.dirname(__file__), "output")
os.makedirs(OUT, exist_ok=True)
# --- Fixture generation ---
def ensure_video(path, duration=4, width=1280, height=720, fps=30, color="blue"):
"""Generate a test video clip with ffmpeg.
Uses forced keyframes every 1s so copy-mode operations retain the video stream.
"""
if os.path.exists(path):
print(f" [fixture] Using existing: {path}")
return
print(f" [fixture] Generating {duration}s {width}x{height}@{fps}fps {color}: {path}")
subprocess.run(
["ffmpeg", "-y",
"-f", "lavfi", "-i", f"color=c={color}:s={width}x{height}:d={duration}:r={fps}",
"-f", "lavfi", "-i", f"sine=frequency=440:duration={duration}",
"-c:v", "libx264", "-crf", "23", "-pix_fmt", "yuv420p",
"-g", str(fps), "-keyint_min", str(fps),
"-c:a", "aac", "-ar", "44100", "-ac", "2",
"-shortest", path],
capture_output=True, check=True,
)
# Matching clips (same resolution/fps/codec)
CLIP_1 = os.path.join(OUT, "stitch_clip1.mp4")
CLIP_2 = os.path.join(OUT, "stitch_clip2.mp4")
CLIP_3 = os.path.join(OUT, "stitch_clip3.mp4")
ensure_video(CLIP_1, duration=4, color="darkblue")
ensure_video(CLIP_2, duration=4, color="darkgreen")
ensure_video(CLIP_3, duration=4, color="darkred")
# Mismatched clip (different resolution + fps)
CLIP_MISMATCH = os.path.join(OUT, "stitch_clip_mismatch.mp4")
ensure_video(CLIP_MISMATCH, duration=4, width=640, height=480, fps=24, color="purple")
# --- Tool setup ---
tool = VideoStitch()
print(f"Tool status: {tool.get_status()}")
# --- Test 1: Validate matching clips ---
print("\n--- Test 1: Validate matching clips ---")
r1 = tool.execute({
"operation": "validate",
"clips": [CLIP_1, CLIP_2, CLIP_3],
})
print(f"Success: {r1.success}")
if r1.data:
print(f" Compatible: {r1.data.get('compatible')}")
print(f" Total duration: {r1.data.get('total_duration')}s")
print(f" Mismatches: {len(r1.data.get('mismatches', []))}")
if r1.error: print(f"Error: {r1.error}")
# --- Test 2: Validate mismatched clips ---
print("\n--- Test 2: Validate mismatched clips ---")
r2 = tool.execute({
"operation": "validate",
"clips": [CLIP_1, CLIP_MISMATCH],
})
print(f"Success: {r2.success}")
if r2.data:
print(f" Compatible: {r2.data.get('compatible')}")
mismatches = r2.data.get("mismatches", [])
for m in mismatches:
print(f" Clip[{m['clip_index']}]: {', '.join(m['differences'])}")
# --- Test 3: Simple cut stitch (matching clips) ---
print("\n--- Test 3: Cut stitch (2 clips) ---")
r3 = tool.execute({
"operation": "stitch",
"clips": [CLIP_1, CLIP_2],
"transition": "cut",
"output_path": os.path.join(OUT, "stitch_cut.mp4"),
})
print(f"Success: {r3.success}, Duration: {r3.duration_seconds:.2f}s")
if r3.data: print(f" Output duration: {r3.data.get('duration')}s, Method: {r3.data.get('method')}")
if r3.error: print(f"Error: {r3.error}")
# --- Test 4: Crossfade stitch ---
print("\n--- Test 4: Crossfade stitch (2 clips) ---")
r4 = tool.execute({
"operation": "stitch",
"clips": [CLIP_1, CLIP_2],
"transition": "crossfade",
"transition_duration": 1.0,
"output_path": os.path.join(OUT, "stitch_crossfade.mp4"),
})
print(f"Success: {r4.success}, Duration: {r4.duration_seconds:.2f}s")
if r4.data: print(f" Output duration: {r4.data.get('duration')}s, Method: {r4.data.get('method')}")
if r4.error: print(f"Error: {r4.error}")
# --- Test 5: Fade-through-black (3 clips) ---
print("\n--- Test 5: Fade through black (3 clips) ---")
r5 = tool.execute({
"operation": "stitch",
"clips": [CLIP_1, CLIP_2, CLIP_3],
"transition": "fade",
"transition_duration": 0.5,
"output_path": os.path.join(OUT, "stitch_fadeblack.mp4"),
})
print(f"Success: {r5.success}, Duration: {r5.duration_seconds:.2f}s")
if r5.data: print(f" Output duration: {r5.data.get('duration')}s, Method: {r5.data.get('method')}")
if r5.error: print(f"Error: {r5.error}")
# --- Test 6: Auto-normalize mismatched clips ---
print("\n--- Test 6: Stitch mismatched clips (auto_normalize) ---")
r6 = tool.execute({
"operation": "stitch",
"clips": [CLIP_1, CLIP_MISMATCH],
"transition": "cut",
"auto_normalize": True,
"output_path": os.path.join(OUT, "stitch_normalized.mp4"),
})
print(f"Success: {r6.success}, Duration: {r6.duration_seconds:.2f}s")
if r6.data: print(f" Output duration: {r6.data.get('duration')}s, Normalized: {r6.data.get('auto_normalized')}")
if r6.error: print(f"Error: {r6.error}")
# --- Test 7: Preview stitch (low-res) ---
print("\n--- Test 7: Preview stitch ---")
r7 = tool.execute({
"operation": "preview_stitch",
"clips": [CLIP_1, CLIP_2, CLIP_3],
"transition": "cut",
"output_path": os.path.join(OUT, "stitch_preview.mp4"),
})
print(f"Success: {r7.success}, Duration: {r7.duration_seconds:.2f}s")
if r7.data: print(f" Preview resolution: {r7.data.get('preview_resolution')}")
if r7.error: print(f"Error: {r7.error}")
# --- Test 8: Spatial — side by side ---
print("\n--- Test 8: Spatial side-by-side ---")
r8 = tool.execute({
"operation": "spatial",
"clips": [CLIP_1, CLIP_2],
"layout": "side_by_side",
"output_path": os.path.join(OUT, "stitch_side_by_side.mp4"),
})
print(f"Success: {r8.success}, Duration: {r8.duration_seconds:.2f}s")
if r8.data: print(f" Layout: {r8.data.get('layout')}, Duration: {r8.data.get('duration')}s")
if r8.error: print(f"Error: {r8.error}")
# --- Test 9: Spatial — picture-in-picture ---
print("\n--- Test 9: Spatial PIP (bottom-right) ---")
r9 = tool.execute({
"operation": "spatial",
"clips": [CLIP_1, CLIP_2],
"layout": "picture_in_picture",
"pip_position": "bottom_right",
"pip_scale": 0.3,
"pip_margin": 20,
"output_path": os.path.join(OUT, "stitch_pip.mp4"),
})
print(f"Success: {r9.success}, Duration: {r9.duration_seconds:.2f}s")
if r9.data: print(f" Layout: {r9.data.get('layout')}, Duration: {r9.data.get('duration')}s")
if r9.error: print(f"Error: {r9.error}")
# --- Test 10: Spatial — vertical stack ---
print("\n--- Test 10: Spatial vertical stack ---")
r10 = tool.execute({
"operation": "spatial",
"clips": [CLIP_1, CLIP_2],
"layout": "vertical_stack",
"output_path": os.path.join(OUT, "stitch_vstack.mp4"),
})
print(f"Success: {r10.success}, Duration: {r10.duration_seconds:.2f}s")
if r10.data: print(f" Layout: {r10.data.get('layout')}, Duration: {r10.data.get('duration')}s")
if r10.error: print(f"Error: {r10.error}")
# --- Probe all video outputs ---
print("\n--- Output inspection ---")
outputs = [
"stitch_cut.mp4",
"stitch_crossfade.mp4",
"stitch_fadeblack.mp4",
"stitch_normalized.mp4",
"stitch_preview.mp4",
"stitch_side_by_side.mp4",
"stitch_pip.mp4",
"stitch_vstack.mp4",
]
for name in outputs:
path = os.path.join(OUT, name)
if os.path.exists(path):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", path],
capture_output=True, text=True,
)
info = json.loads(probe.stdout)
fmt = info.get("format", {})
video = {}
audio = {}
for s in info.get("streams", []):
if s.get("codec_type") == "video" and not video:
video = s
elif s.get("codec_type") == "audio" and not audio:
audio = s
print(f"\n[{name}]"
f" Duration: {fmt.get('duration', '?')}s"
f" | Video: {video.get('width', '?')}x{video.get('height', '?')}"
f" {video.get('codec_name', '?')}"
f" | Audio: {audio.get('codec_name', '?')}"
f" | Size: {os.path.getsize(path)} bytes")
else:
print(f"\n[{name}] FILE NOT FOUND")
print("\n=== VIDEO STITCH TEST COMPLETE ===")
+279
View File
@@ -0,0 +1,279 @@
#!/usr/bin/env python3
"""QA Test 07: Playbook design intelligence — no API calls.
Tests contrast validation, color harmony generation, color-blind safety,
type scale computation, type hierarchy validation, font pairing suggestions,
and full accessibility audit across all 3 playbooks.
"""
import sys, os, json
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from styles.playbook_loader import (
load_playbook,
validate_playbook,
list_playbooks,
validate_contrast,
check_color_blind_safety,
validate_palette,
generate_harmony,
compute_type_scale,
validate_type_hierarchy,
suggest_font_pairing,
validate_accessibility,
TYPE_SCALE_RATIOS,
)
PASS = 0
FAIL = 0
def check(name, condition, detail=""):
global PASS, FAIL
status = "PASS" if condition else "FAIL"
if condition:
PASS += 1
else:
FAIL += 1
suffix = f"{detail}" if detail else ""
print(f" [{status}] {name}{suffix}")
# ===================================================================
# Test 1: List and load all playbooks
# ===================================================================
print("--- Test 1: List and load playbooks ---")
playbooks_available = list_playbooks()
print(f" Found {len(playbooks_available)} playbooks: {playbooks_available}")
check("At least 3 playbooks exist", len(playbooks_available) >= 3)
loaded = {}
for name in ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]:
try:
pb = load_playbook(name)
loaded[name] = pb
check(f"Load + validate {name}", True)
except Exception as e:
loaded[name] = None
check(f"Load + validate {name}", False, str(e))
# ===================================================================
# Test 2: Contrast validation (manual spot-checks)
# ===================================================================
print("\n--- Test 2: Contrast validation ---")
# Known pair: black on white should be 21:1
r = validate_contrast("#000000", "#FFFFFF")
check("Black on white ~21:1", abs(r["ratio"] - 21.0) < 0.1, f"ratio={r['ratio']}")
check("Black on white passes AAA", r["normal_text"]["AAA"])
# Known pair: white on white should be 1:1
r = validate_contrast("#FFFFFF", "#FFFFFF")
check("White on white = 1:1", abs(r["ratio"] - 1.0) < 0.01, f"ratio={r['ratio']}")
check("White on white fails AA", not r["normal_text"]["AA"])
# Mid-gray on white (~4.5:1 boundary)
r = validate_contrast("#767676", "#FFFFFF")
check("#767676 on white passes AA normal", r["normal_text"]["AA"], f"ratio={r['ratio']}")
# Just-below threshold
r = validate_contrast("#777777", "#FFFFFF")
check("#777777 on white borderline", r["ratio"] >= 4.4, f"ratio={r['ratio']}")
# Dark-on-dark: low contrast
r = validate_contrast("#1A1A1A", "#2B2B2B")
check("Dark-on-dark fails AA", not r["normal_text"]["AA"], f"ratio={r['ratio']}")
# Check each playbook's text-on-bg contrast
for name, pb in loaded.items():
if pb is None:
continue
palette = pb.get("visual_language", {}).get("color_palette", {})
text = palette.get("text", "#000000")
bg = palette.get("background", "#FFFFFF")
r = validate_contrast(text, bg)
check(f"{name}: text on bg passes AA", r["normal_text"]["AA"], f"ratio={r['ratio']}")
# ===================================================================
# Test 3: Color harmony generation
# ===================================================================
print("\n--- Test 3: Color harmony generation ---")
for harmony_type in ["complementary", "analogous", "triadic", "split-complementary"]:
colors = generate_harmony("#3B82F6", harmony_type)
check(f"Harmony {harmony_type}", len(colors) >= 2, f"generated {len(colors)} colors: {colors}")
# First color should match base
check(f" Base preserved in {harmony_type}", colors[0].upper() == generate_harmony("#3B82F6", harmony_type)[0].upper())
# Edge case: pure red
colors = generate_harmony("#FF0000", "triadic")
check("Triadic from pure red", len(colors) == 3, f"{colors}")
# ===================================================================
# Test 4: Color-blind safety
# ===================================================================
print("\n--- Test 4: Color-blind safety ---")
# Safe palette (blue + orange — distinguishable by all CVD types due to lightness diff)
safe = check_color_blind_safety(["#2563EB", "#F59E0B"])
print(f" Blue + orange: safe={safe['safe']}, issues={len(safe.get('issues', []))}")
# Problematic palette (red + green, similar lightness)
risky = check_color_blind_safety(["#DC2626", "#16A34A"])
print(f" Red + green: safe={risky['safe']}, issues={len(risky.get('issues', []))}")
check("Red+green flagged as risky", not risky["safe"] or len(risky.get("issues", [])) > 0,
"should flag deuteranopia/protanopia")
# Single color = no pairs to check
single = check_color_blind_safety(["#FF0000"])
check("Single color is safe", single["safe"])
# Grays should be safe (low saturation)
grays = check_color_blind_safety(["#333333", "#999999", "#CCCCCC"])
check("Grays are safe", grays["safe"])
# ===================================================================
# Test 5: Full palette validation per playbook
# ===================================================================
print("\n--- Test 5: Palette validation (all playbooks) ---")
for name, pb in loaded.items():
if pb is None:
continue
issues = validate_palette(pb)
errors = [i for i in issues if i.get("severity") == "error"]
warnings = [i for i in issues if i.get("severity") == "warning"]
print(f" [{name}] {len(errors)} errors, {len(warnings)} warnings, {len(issues)} total issues")
check(f"{name}: no contrast errors", len(errors) == 0,
"; ".join(e["message"] for e in errors) if errors else "all clear")
for issue in issues:
sev = issue.get("severity", "?")
print(f" [{sev}] {issue.get('message', '')}")
# ===================================================================
# Test 6: Type scale computation
# ===================================================================
print("\n--- Test 6: Type scale computation ---")
for ratio_name, ratio_val in TYPE_SCALE_RATIOS.items():
scale = compute_type_scale(24, ratio_name)
sizes = scale["sizes"]
check(f"Scale {ratio_name}: display > heading > subheading > body > caption",
sizes["display"] > sizes["heading"] > sizes["subheading"] > sizes["body"] > sizes["caption"],
f"{sizes}")
check(f" Base preserved", sizes["body"] == 24)
# Custom numeric ratio
scale = compute_type_scale(24, "1.5")
check("Custom ratio 1.5", scale["ratio_value"] == 1.5, f"sizes={scale['sizes']}")
# ===================================================================
# Test 7: Type hierarchy validation
# ===================================================================
print("\n--- Test 7: Type hierarchy validation ---")
for name, pb in loaded.items():
if pb is None:
continue
issues = validate_type_hierarchy(pb)
print(f" [{name}] {len(issues)} type hierarchy issues")
for issue in issues:
print(f" [{issue.get('severity')}] {issue.get('message')}")
# No errors expected in shipped playbooks
errors = [i for i in issues if i.get("severity") == "error"]
check(f"{name}: no type hierarchy errors", len(errors) == 0)
# Deliberately bad playbook
bad_typography = {
"typography": {
"headings": {"font": "Inter", "weight": 400},
"body": {"font": "Inter", "weight": 400},
"stat_card": {"font": "Inter", "size_multiplier": 0.8},
}
}
issues = validate_type_hierarchy(bad_typography)
check("Bad typography flagged", len(issues) > 0, f"{len(issues)} issues found")
# ===================================================================
# Test 8: Font pairing suggestions
# ===================================================================
print("\n--- Test 8: Font pairing suggestions ---")
for font in ["Inter", "Space Grotesk", "IBM Plex Sans", "Lora", "JetBrains Mono"]:
pairings = suggest_font_pairing(font)
check(f"Pairings for {font}", len(pairings) >= 1, f"{len(pairings)} suggestions")
for p in pairings:
print(f"{p['font']} ({p['category']}): {p['rationale']}")
# Unknown font fallback
pairings = suggest_font_pairing("UnknownFont")
check("Unknown font gets fallback", len(pairings) >= 1)
# ===================================================================
# Test 9: Full accessibility audit (all playbooks)
# ===================================================================
print("\n--- Test 9: Accessibility audit (all playbooks) ---")
for name, pb in loaded.items():
if pb is None:
continue
result = validate_accessibility(pb)
status = "PASS" if result["pass"] else "FAIL"
print(f"\n [{name}] Overall: {status}"
f" | Errors: {result['error_count']}"
f" | Warnings: {result['warning_count']}"
f" | Total: {result['total_issues']}")
check(f"{name}: a11y audit passes", result["pass"])
for issue in result["issues"]:
cat = issue.get("category", "?")
sev = issue.get("severity", "?")
print(f" [{cat}/{sev}] {issue.get('message', '')}")
# ===================================================================
# Test 10: Deliberately low-contrast custom playbook
# ===================================================================
print("\n--- Test 10: Low-contrast custom playbook ---")
low_contrast_pb = {
"identity": {"name": "low-contrast-test", "category": "test", "mood": "test", "pace": "moderate", "best_for": ["testing"]},
"visual_language": {
"color_palette": {
"primary": ["#555555"],
"accent": ["#666666"],
"background": "#444444",
"text": "#555555",
"muted": "#4A4A4A",
},
"composition": "centered",
"texture": "none",
},
"typography": {
"headings": {"font": "Arial", "weight": 700, "size_multiplier": 1.5},
"body": {"font": "Arial", "weight": 400, "size_multiplier": 1.0},
"code": {"font": "Courier", "weight": 400, "size_multiplier": 0.9},
"stat_card": {"font": "Arial", "weight": 700, "size_multiplier": 2.5},
"scale_system": "major_third",
"weight_matrix": {"title": 800, "heading": 700, "body": 400, "caption": 300},
},
"motion": {"transitions": "cut", "animation_style": "none", "pacing_rules": {}},
"audio": {"voice_style": "neutral", "music_mood": "none"},
"asset_generation": {"image_prompt_prefix": "test", "negative_prompt": ""},
"overlays": {
"stat_card": {"bg": "#444444", "text": "#555555", "border": "#444444", "radius": 8, "shadow": "none"},
},
"quality_rules": [],
"chart_palette": ["#555555", "#666666", "#777777"],
}
# This should be flagged with errors
issues = validate_palette(low_contrast_pb)
errors = [i for i in issues if i.get("severity") == "error"]
check("Low-contrast playbook has errors", len(errors) > 0, f"{len(errors)} contrast errors")
for e in errors:
print(f" [error] {e.get('message')}")
# ===================================================================
# Summary
# ===================================================================
print(f"\n{'='*60}")
print(f"PLAYBOOK INTELLIGENCE TEST COMPLETE: {PASS} passed, {FAIL} failed")
print(f"{'='*60}")
+548
View File
@@ -0,0 +1,548 @@
#!/usr/bin/env python3
"""QA Test 08: End-to-end animated-explainer pipeline simulation.
Walks through all 7 stages (idea -> publish) with synthetic artifacts,
validating checkpoints, artifact schemas, and cost tracking at each step.
The compose stage runs real tools (audio_mixer + video_compose) to produce
an actual output video. All other stages use synthetic data.
No API keys needed -- uses ffmpeg-generated fixtures throughout.
"""
import sys, os, json, subprocess, shutil
from datetime import datetime, timezone
from pathlib import Path
PROJECT_ROOT = str(Path(__file__).resolve().parent.parent.parent)
sys.path.insert(0, PROJECT_ROOT)
from lib.env_loader import load_env
load_env()
from lib.checkpoint import (
write_checkpoint,
read_checkpoint,
get_completed_stages,
get_next_stage,
STAGES,
CANONICAL_STAGE_ARTIFACTS,
)
from tools.cost_tracker import CostTracker, BudgetMode
from schemas.artifacts import validate_artifact, list_schemas
from styles.playbook_loader import load_playbook, validate_accessibility
OUT = os.path.join(os.path.dirname(__file__), "output")
PIPELINE_DIR = Path(OUT) / "e2e_pipeline"
PROJECT_ID = "qa_e2e_test"
ASSETS_DIR = Path(OUT) / "e2e_assets"
# Clean previous run
if PIPELINE_DIR.exists():
shutil.rmtree(PIPELINE_DIR)
if ASSETS_DIR.exists():
shutil.rmtree(ASSETS_DIR)
ASSETS_DIR.mkdir(parents=True, exist_ok=True)
PASS = 0
FAIL = 0
def check(name, condition, detail=""):
global PASS, FAIL
if condition:
PASS += 1
print(f" [PASS] {name}" + (f" -- {detail}" if detail else ""))
else:
FAIL += 1
print(f" [FAIL] {name}" + (f" -- {detail}" if detail else ""))
def ensure_audio(path, duration=5):
if os.path.exists(path):
return
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i",
f"sine=frequency=440:duration={duration}",
"-ar", "44100", "-ac", "1", path],
capture_output=True, check=True,
)
def ensure_video(path, duration=5, width=1280, height=720, color="blue"):
if os.path.exists(path):
return
subprocess.run(
["ffmpeg", "-y",
"-f", "lavfi", "-i", f"color=c={color}:s={width}x{height}:d={duration}:r=30",
"-f", "lavfi", "-i", f"sine=frequency=440:duration={duration}",
"-c:v", "libx264", "-crf", "23", "-pix_fmt", "yuv420p",
"-g", "30", "-keyint_min", "30",
"-c:a", "aac", "-shortest", path],
capture_output=True, check=True,
)
# ===================================================================
# Setup: Cost tracker + playbook
# ===================================================================
print("--- Setup ---")
cost_log = PIPELINE_DIR / PROJECT_ID / "cost_log.json"
tracker = CostTracker(
budget_total_usd=5.0,
mode=BudgetMode.OBSERVE,
cost_log_path=cost_log,
)
print(f" Budget: ${tracker.budget_total_usd}")
print(f" Available schemas: {list_schemas()}")
playbook = load_playbook("clean-professional")
a11y = validate_accessibility(playbook)
print(f" Playbook a11y: pass={a11y['pass']}, errors={a11y['error_count']}, warnings={a11y['warning_count']}")
# ===================================================================
# Stage 1: idea -> brief
# ===================================================================
print("\n--- Stage 1: idea ---")
brief = {
"version": "1.0",
"title": "AI Video Production in 60 Seconds",
"hook": "What if you could create a professional video in 60 seconds with just a text prompt?",
"key_points": [
"Traditional video production takes days or weeks",
"AI can automate scripting, visuals, narration, and editing",
"OpenMontage orchestrates the full pipeline",
],
"tone": "confident, energetic",
"style": "clean-professional",
"target_platform": "youtube",
"target_duration_seconds": 60,
"target_audience": "content creators and developers",
"cta": "Try OpenMontage today",
"angle_options": [
{"name": "democratization", "description": "AI makes video creation accessible to everyone"},
{"name": "workflow", "description": "AI automates the tedious parts of video production"},
{"name": "quality", "description": "AI-generated content is reaching professional quality"},
],
"selected_angle": "workflow",
}
try:
validate_artifact("brief", brief)
check("Brief validates against schema", True)
except Exception as e:
check("Brief validates against schema", False, str(e))
cp_path = write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "idea", "completed",
artifacts={"brief": brief},
pipeline_type="animated-explainer",
style_playbook="clean-professional",
)
check("Idea checkpoint written", cp_path.exists())
# Next uncompleted stage in global STAGES order (research/proposal come before idea)
check("Next stage after idea", get_next_stage(PIPELINE_DIR, PROJECT_ID) == "research")
# ===================================================================
# Stage 2: script
# ===================================================================
print("\n--- Stage 2: script ---")
# Section timestamps (cumulative)
SECTIONS = [
("s1_hook", "Hook", 0, 8, "What if creating a professional video took less time than making your morning coffee?"),
("s2_setup", "Setup", 8, 20, "Traditional video production involves scripting, filming, editing, and post-production. It takes days, sometimes weeks."),
("s3_build", "Build", 20, 38, "Now imagine an AI that handles all of that. You type a topic, and it writes the script, generates visuals, creates narration, mixes audio, and delivers a finished video."),
("s4_climax", "Climax", 38, 50, "This is not science fiction. OpenMontage is an open-source platform that orchestrates AI tools into a complete video pipeline."),
("s5_landing", "Landing", 50, 60, "The future of video creation is open, automated, and available right now. Try it yourself."),
]
script = {
"version": "1.0",
"title": "AI Video Production in 60 Seconds",
"total_duration_seconds": 60,
"sections": [
{
"id": sid,
"label": label,
"text": text,
"start_seconds": start,
"end_seconds": end,
"speaker_directions": "Confident, engaging tone",
"enhancement_cues": [
{"type": "overlay", "description": f"Visual for {label} section", "timestamp_seconds": start + 2},
],
}
for sid, label, start, end, text in SECTIONS
],
}
try:
validate_artifact("script", script)
check("Script validates against schema", True)
except Exception as e:
check("Script validates against schema", False, str(e))
write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "script", "completed",
artifacts={"script": script},
pipeline_type="animated-explainer",
)
check("Completed stages", get_completed_stages(PIPELINE_DIR, PROJECT_ID) == ["idea", "script"]) # idea and script appear in STAGES order
# ===================================================================
# Stage 3: scene_plan
# ===================================================================
print("\n--- Stage 3: scene_plan ---")
SCENE_TYPES = ["text_card", "diagram", "animation", "generated", "text_card"]
scene_plan = {
"version": "1.0",
"style_playbook": "clean-professional",
"scenes": [
{
"id": f"sc{i+1}",
"type": SCENE_TYPES[i],
"description": f"Scene for {label} section",
"start_seconds": start,
"end_seconds": end,
"script_section_id": sid,
"required_assets": [
{"type": "narration", "description": f"TTS narration for {label}", "source": "generate"},
{"type": "image", "description": f"Visual for {label}", "source": "generate"},
],
}
for i, (sid, label, start, end, _) in enumerate(SECTIONS)
],
}
try:
validate_artifact("scene_plan", scene_plan)
check("Scene plan validates against schema", True)
except Exception as e:
check("Scene plan validates against schema", False, str(e))
write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "scene_plan", "completed",
artifacts={"scene_plan": scene_plan},
pipeline_type="animated-explainer",
)
# ===================================================================
# Stage 4: assets (generate fixtures)
# ===================================================================
print("\n--- Stage 4: assets ---")
# Generate TTS fixtures (one per section)
tts_files = {}
for sid, label, start, end, _ in SECTIONS:
path = str(ASSETS_DIR / f"tts_{sid}.mp3")
ensure_audio(path, duration=end - start)
tts_files[sid] = path
# Generate music fixture
music_path = str(ASSETS_DIR / "music_bg.mp3")
ensure_audio(music_path, duration=60)
# Build asset manifest (schema: id, type, path, source_tool, scene_id required)
clean_assets = []
for i, (sid, label, start, end, _) in enumerate(SECTIONS):
scene_id = f"sc{i+1}"
clean_assets.append({
"id": f"a_tts_{sid}",
"type": "narration",
"path": tts_files[sid],
"source_tool": "elevenlabs_tts",
"scene_id": scene_id,
})
# Image fixture
img_path = str(ASSETS_DIR / f"img_{scene_id}.png")
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i",
f"color=c=darkblue:s=1280x720:d=1", "-frames:v", "1", img_path],
capture_output=True, check=True,
)
clean_assets.append({
"id": f"a_img_{scene_id}",
"type": "image",
"path": img_path,
"source_tool": "image_selector",
"scene_id": scene_id,
})
clean_assets.append({
"id": "a_music",
"type": "music",
"path": music_path,
"source_tool": "music_gen",
"scene_id": "sc1",
})
asset_manifest = {
"version": "1.0",
"assets": clean_assets,
"total_cost_usd": 0.0,
}
try:
validate_artifact("asset_manifest", asset_manifest)
check("Asset manifest validates against schema", True)
except Exception as e:
check("Asset manifest validates against schema", False, str(e))
# Verify all files exist
all_exist = all(os.path.exists(a["path"]) for a in asset_manifest["assets"])
check("All asset files exist on disk", all_exist)
# Track costs
eid = tracker.estimate("image_selector", "generate", 0.15)
tracker.approve_tool("image_selector")
tracker.reserve(eid)
tracker.reconcile(eid, 0.0, success=True)
print(f" Cost snapshot: {tracker.cost_snapshot()}")
write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "assets", "completed",
artifacts={"asset_manifest": asset_manifest},
pipeline_type="animated-explainer",
cost_snapshot=tracker.cost_snapshot(),
)
# ===================================================================
# Stage 5: edit (edit_decisions)
# ===================================================================
print("\n--- Stage 5: edit ---")
# Create video clips for the compose step
colors = ["darkblue", "darkgreen", "darkorange", "darkred", "purple"]
scene_videos = {}
for i, scene in enumerate(scene_plan["scenes"]):
scid = scene["id"]
dur = scene["end_seconds"] - scene["start_seconds"]
vid_path = str(ASSETS_DIR / f"scene_{scid}.mp4")
ensure_video(vid_path, duration=dur, color=colors[i % len(colors)])
scene_videos[scid] = vid_path
edit_decisions = {
"version": "1.0",
"cuts": [
{
"id": f"cut_{scene['id']}",
"source": scene_videos[scene["id"]],
"in_seconds": 0,
"out_seconds": scene["end_seconds"] - scene["start_seconds"],
"speed": 1.0,
}
for scene in scene_plan["scenes"]
],
"music": {
"asset_id": "a_music",
"volume": 0.2,
"ducking": True,
"fade_in_seconds": 1.0,
"fade_out_seconds": 2.0,
},
"subtitles": {
"enabled": True,
"style": "clean-professional",
},
}
try:
validate_artifact("edit_decisions", edit_decisions)
check("Edit decisions validates against schema", True)
except Exception as e:
check("Edit decisions validates against schema", False, str(e))
write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "edit", "completed",
artifacts={"edit_decisions": edit_decisions},
pipeline_type="animated-explainer",
)
# ===================================================================
# Stage 6: compose (REAL tool execution)
# ===================================================================
print("\n--- Stage 6: compose (real tools) ---")
from tools.audio.audio_mixer import AudioMixer
from tools.video.video_compose import VideoCompose
# Step 1: Mix narration + music
print(" Mixing audio...")
mixer = AudioMixer()
mix_output = str(ASSETS_DIR / "final_mix.wav")
# Combine all narration into one track first
concat_narration = str(ASSETS_DIR / "narration_concat.wav")
narration_list = str(ASSETS_DIR / "narration_list.txt")
with open(narration_list, "w") as f:
for sid, _, _, _, _ in SECTIONS:
safe = tts_files[sid].replace("\\", "/")
f.write(f"file '{safe}'\n")
subprocess.run(
["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", narration_list,
"-c", "copy", concat_narration],
capture_output=True, check=True,
)
mix_result = mixer.execute({
"operation": "duck",
"tracks": [
{"path": concat_narration, "role": "speech"},
{"path": music_path, "role": "music"},
],
"ducking": {"enabled": True, "music_volume_during_speech": 0.15},
"output_path": mix_output,
})
check("Audio mix succeeded", mix_result.success, mix_result.error or "")
# Step 2: Compose video
print(" Composing video...")
composer = VideoCompose()
final_video = str(Path(OUT) / "e2e_final_output.mp4")
compose_result = composer.execute({
"operation": "compose",
"edit_decisions": {
"cuts": [
{"source": c["source"], "in_seconds": c["in_seconds"], "out_seconds": c["out_seconds"], "speed": c.get("speed", 1.0)}
for c in edit_decisions["cuts"]
],
},
"audio_path": mix_output,
"codec": "libx264",
"crf": 23,
"preset": "fast",
"output_path": final_video,
})
check("Video compose succeeded", compose_result.success, compose_result.error or "")
check("Output video exists", os.path.exists(final_video))
# Probe the output
duration = 0.0
video_stream = {}
audio_stream = {}
if os.path.exists(final_video):
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", "-show_streams", final_video],
capture_output=True, text=True,
)
info = json.loads(probe.stdout)
fmt = info.get("format", {})
duration = float(fmt.get("duration", 0))
for s in info.get("streams", []):
if s.get("codec_type") == "video" and not video_stream:
video_stream = s
elif s.get("codec_type") == "audio" and not audio_stream:
audio_stream = s
print(f" Output: {video_stream.get('width')}x{video_stream.get('height')}"
f" {video_stream.get('codec_name')} | {duration:.1f}s"
f" | Audio: {audio_stream.get('codec_name')}"
f" | Size: {os.path.getsize(final_video)} bytes")
check("Video has audio track", bool(audio_stream))
check("Video has video track", bool(video_stream))
check("Duration > 30s", duration > 30, f"{duration:.1f}s")
render_report = {
"version": "1.0",
"outputs": [
{
"path": final_video,
"format": "mp4",
"codec": video_stream.get("codec_name", "h264"),
"audio_codec": audio_stream.get("codec_name", "aac"),
"resolution": f"{video_stream.get('width', 1280)}x{video_stream.get('height', 720)}",
"fps": 30,
"duration_seconds": round(duration, 2),
"file_size_bytes": os.path.getsize(final_video) if os.path.exists(final_video) else 0,
"platform_target": "youtube",
}
],
"render_time_seconds": compose_result.duration_seconds,
}
try:
validate_artifact("render_report", render_report)
check("Render report validates against schema", True)
except Exception as e:
check("Render report validates against schema", False, str(e))
write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "compose", "completed",
artifacts={"render_report": render_report},
pipeline_type="animated-explainer",
cost_snapshot=tracker.cost_snapshot(),
)
# ===================================================================
# Stage 7: publish
# ===================================================================
print("\n--- Stage 7: publish ---")
publish_log = {
"version": "1.0",
"entries": [
{
"platform": "youtube",
"status": "exported",
"timestamp": datetime.now(timezone.utc).isoformat(),
"export_path": str(Path(OUT) / "e2e_export"),
"metadata_used": {
"title": "AI Video Production in 60 Seconds",
"description": "See how AI orchestrates an entire video production pipeline.",
"hashtags": ["#AI", "#VideoProduction", "#OpenMontage"],
"chapters": [
{"time": "0:00", "label": "Hook"},
{"time": "0:08", "label": "The Problem"},
{"time": "0:20", "label": "The Solution"},
{"time": "0:38", "label": "OpenMontage"},
{"time": "0:50", "label": "Try It"},
],
},
}
],
}
try:
validate_artifact("publish_log", publish_log)
check("Publish log validates against schema", True)
except Exception as e:
check("Publish log validates against schema", False, str(e))
write_checkpoint(
PIPELINE_DIR, PROJECT_ID, "publish", "completed",
artifacts={"publish_log": publish_log},
pipeline_type="animated-explainer",
)
# ===================================================================
# Final validation
# ===================================================================
print("\n--- Final validation ---")
completed = get_completed_stages(PIPELINE_DIR, PROJECT_ID)
check("All 7 stages completed", len(completed) == 7, f"completed={completed}")
check("Next stage is None (done)", get_next_stage(PIPELINE_DIR, PROJECT_ID) is None)
check("Stages in correct order", completed == STAGES, f"{completed}")
# Verify all checkpoints are readable
for stage in STAGES:
cp = read_checkpoint(PIPELINE_DIR, PROJECT_ID, stage)
check(f"Checkpoint {stage} readable", cp is not None)
if cp:
expected_artifact = CANONICAL_STAGE_ARTIFACTS[stage]
check(f" Has canonical artifact '{expected_artifact}'", expected_artifact in cp.get("artifacts", {}))
# Cost summary
print(f"\n Final cost: {tracker.cost_snapshot()}")
# ===================================================================
# Summary
# ===================================================================
print(f"\n{'='*60}")
print(f"END-TO-END TEST COMPLETE: {PASS} passed, {FAIL} failed")
print(f"{'='*60}")
if os.path.exists(final_video):
print(f"\nFinal video: {final_video}")
print("INSPECT: Open in VLC/media player to verify A/V sync, transitions, and content.")
View File
View File