"""Phase 3 — tests for Seedance V2V edit dispatcher. Verifies: - source video validation - tier propagation (standard / fast) - image-ref view resolution - image-path escape hatch - mutex enforcement (image-refs XOR image-paths) - reference_videos cap (source + extras <= 3) - dry-run no-submit - generate_audio default ON - dispatch() routing with modality="video_i2v" + reference_videos - duration / aspect-ratio whitelist enforcement - provider routing locked to fal via provider_strategy.json (no piapi drift) All tests stub `dispatch()` so no fal.ai calls happen. """ from __future__ import annotations import json import sys from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest _REPO_ROOT = Path(__file__).resolve().parents[4] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from recoil.pipeline.tools.dispatch_cli import ( # noqa: E402 _dispatch_seedance_v2v_edit, ) def _make_args(**overrides): """Construct a SimpleNamespace mimicking argparse.Namespace defaults.""" base = { "project": "_probes", "source_video": None, "prompt": "Change kitchen to gothic", "image_refs": None, "image_paths": None, "ref_video": None, "duration": 5, "aspect_ratio": "9:16", "tier": "standard", "no_audio": False, "dry_run": False, "pass_id": None, "no_validate_frames": True, } base.update(overrides) return SimpleNamespace(**base) def _make_mock_receipt(success=True, cost=1.51, output="/tmp/out.mp4"): """Build a mock receipt with the minimum fields read by the dispatcher.""" run_result = MagicMock() run_result.success = success run_result.output_path = output run_result.error = None if success else "stub failure" run_result.metadata = {"cost_usd": cost} receipt = MagicMock() receipt.run_result = run_result return receipt @pytest.fixture def mock_dispatch(): """Patch `dispatch` inside dispatch_cli's namespace so the function under test sees the mock when it calls dispatch(...).""" with patch("recoil.pipeline.tools.dispatch_cli.dispatch") as m: m.return_value = _make_mock_receipt() yield m @pytest.fixture def mock_store(): return MagicMock() @pytest.fixture def mock_runner(): """Patch StepRunner / ProjectPaths / _register_test_dispatch_with_passstore / read_cost_from_result so the dispatcher runs end-to-end without filesystem or PassStore side effects.""" with patch("recoil.pipeline.tools.dispatch_cli.read_cost_from_result", return_value=1.51), \ patch("recoil.pipeline.tools.dispatch_cli._register_test_dispatch_with_passstore"), \ patch("recoil.execution.step_runner.StepRunner"), \ patch("recoil.execution.step_types.ProjectPaths") as paths_cls: paths_cls.for_episode.return_value = SimpleNamespace( video_dir=Path("/tmp/_test_seedance_v2v"), ) Path("/tmp/_test_seedance_v2v").mkdir(parents=True, exist_ok=True) yield def test_source_video_required(mock_store, mock_dispatch, mock_runner): args = _make_args(source_video=None) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 2 mock_dispatch.assert_not_called() def test_source_video_missing_raises(tmp_path, mock_store, mock_dispatch, mock_runner): bogus = tmp_path / "nope.mp4" args = _make_args(source_video=str(bogus)) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 1 mock_dispatch.assert_not_called() def test_prompt_required(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), prompt=None) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 2 mock_dispatch.assert_not_called() def test_tier_invalid_raises(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), tier="ultra") rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 2 mock_dispatch.assert_not_called() def test_duration_out_of_range_raises(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), duration=20) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 2 mock_dispatch.assert_not_called() def test_aspect_ratio_whitelist(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), aspect_ratio="2.39:1") rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 2 mock_dispatch.assert_not_called() def test_mutex_image_refs_and_paths_raises(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") ref = tmp_path / "ref.png" ref.write_bytes(b"fake") args = _make_args( source_video=str(src), image_refs="DRIVER", image_paths=str(ref), ) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 2 mock_dispatch.assert_not_called() def test_max_reference_videos_enforced(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") extras = [] for i in range(3): # 1 source + 3 extras = 4 total — must fail e = tmp_path / f"extra{i}.mp4" e.write_bytes(b"fake") extras.append(str(e)) args = _make_args( source_video=str(src), ref_video=",".join(extras), ) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 1 mock_dispatch.assert_not_called() def test_dry_run_does_not_submit(tmp_path, mock_store, mock_dispatch, mock_runner): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), dry_run=True) rc = _dispatch_seedance_v2v_edit(args, mock_store) assert rc == 0 mock_dispatch.assert_not_called() def test_tier_standard_routes_to_standard_720p( tmp_path, mock_store, mock_dispatch, mock_runner, ): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), tier="standard") _dispatch_seedance_v2v_edit(args, mock_store) mock_dispatch.assert_called_once() payload = mock_dispatch.call_args.args[1] assert payload["provider_hints"]["tier"] == "standard_720p" def test_tier_fast_routes_to_fast_720p( tmp_path, mock_store, mock_dispatch, mock_runner, ): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), tier="fast") _dispatch_seedance_v2v_edit(args, mock_store) mock_dispatch.assert_called_once() payload = mock_dispatch.call_args.args[1] assert payload["provider_hints"]["tier"] == "fast_720p" def test_payload_routes_via_dispatch_modality_video_i2v( tmp_path, mock_store, mock_dispatch, mock_runner, ): """Per the locked architectural facts, V2V routes through modality 'video_i2v' — the fal adapter's _infer_action discriminates R2V from I2V by inspecting payload.reference_videos, not by modality string.""" src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src)) _dispatch_seedance_v2v_edit(args, mock_store) mock_dispatch.assert_called_once() modality_arg = mock_dispatch.call_args.args[0] assert modality_arg == "video_i2v" def test_source_video_is_first_in_reference_videos( tmp_path, mock_store, mock_dispatch, mock_runner, ): src = tmp_path / "src.mp4" src.write_bytes(b"fake") extra = tmp_path / "extra.mp4" extra.write_bytes(b"fake") args = _make_args(source_video=str(src), ref_video=str(extra)) _dispatch_seedance_v2v_edit(args, mock_store) payload = mock_dispatch.call_args.args[1] rv = payload["reference_videos"] assert len(rv) == 2 assert rv[0] == str(src.resolve()) assert rv[1] == str(extra.resolve()) def test_image_paths_escape_hatch( tmp_path, mock_store, mock_dispatch, mock_runner, ): src = tmp_path / "src.mp4" src.write_bytes(b"fake") ref = tmp_path / "ref.png" ref.write_bytes(b"fake") args = _make_args(source_video=str(src), image_paths=str(ref)) _dispatch_seedance_v2v_edit(args, mock_store) payload = mock_dispatch.call_args.args[1] assert payload["reference_images"] == [str(ref)] def test_generate_audio_default_on( tmp_path, mock_store, mock_dispatch, mock_runner, ): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), no_audio=False) _dispatch_seedance_v2v_edit(args, mock_store) payload = mock_dispatch.call_args.args[1] assert payload["generate_audio"] is True def test_no_audio_flag_disables_audio( tmp_path, mock_store, mock_dispatch, mock_runner, ): src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src), no_audio=True) _dispatch_seedance_v2v_edit(args, mock_store) payload = mock_dispatch.call_args.args[1] assert payload["generate_audio"] is False def test_model_forced_to_seeddance( tmp_path, mock_store, mock_dispatch, mock_runner, ): """The dispatcher hardcodes model='seeddance-2.0' regardless of what the user passed in --model. The mode-detection routing only fires for seed* models, so this is safe.""" src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src)) _dispatch_seedance_v2v_edit(args, mock_store) payload = mock_dispatch.call_args.args[1] assert payload["model"] == "seeddance-2.0" def test_caller_id_is_seedance_v2v( tmp_path, mock_store, mock_dispatch, mock_runner, ): """Receipts log caller_id='seedance_v2v' so Phase 5 Trace E can discriminate Seedance V2V dispatches from other modes.""" src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src)) _dispatch_seedance_v2v_edit(args, mock_store) ctx = mock_dispatch.call_args.kwargs["context"] assert ctx.caller_id == "seedance_v2v" def test_provider_routing_resolves_to_fal_adapter( tmp_path, mock_store, mock_dispatch, mock_runner, ): """Locked fact #9 (post-Flora-pivot): provider routing for 'seeddance-2.0' is governed by provider_strategy.json, which now declares primary='flora' with fallback='fal' (the Flora pivot made Flora the primary execution target; fal remains the documented fallback adapter in the routing chain). The dispatcher's payload includes model='seeddance-2.0', so resolve_adapter() picks the SSOT-defined provider regardless of model_profiles.json's informational piapi 'status' field. This test reads provider_strategy.json directly and asserts the entry we depend on holds the expected primary/fallback contract. If this fails, the locked architectural fact is broken and the dispatcher could route to an undefined provider. """ strategy_path = _REPO_ROOT / "recoil" / "config" / "provider_strategy.json" strategy = json.loads(strategy_path.read_text(encoding="utf-8")) entry = strategy.get("seeddance-2.0") assert entry is not None, "provider_strategy.json missing seeddance-2.0 entry" assert entry.get("primary") == "flora", ( f"seeddance-2.0 primary provider drifted from 'flora' " f"(got {entry.get('primary')!r}); Flora-pivot SSOT contract violated." ) assert entry.get("fallback") == "fal", ( f"seeddance-2.0 fallback drifted from 'fal' " f"(got {entry.get('fallback')!r}); fal must remain the documented fallback." ) # Verify the dispatcher does NOT inject any provider-override key into the # payload — routing must be SSOT-driven via provider_strategy.json, not # caller-driven. src = tmp_path / "src.mp4" src.write_bytes(b"fake") args = _make_args(source_video=str(src)) _dispatch_seedance_v2v_edit(args, mock_store) payload = mock_dispatch.call_args.args[1] assert "provider" not in payload, ( "Payload must not contain a top-level 'provider' key — routing is " "SSOT-driven via provider_strategy.json." ) assert "provider_override" not in payload.get("provider_hints", {}), ( "provider_hints must not contain 'provider_override' — locked fact #9." )