"""Regression tests for SeedDanceClient post-refactor. Validates: - Legacy public surface (constructor, _resolve_endpoint, is_available, _VALID_TIERS, _VALID_PROVIDERS, QUEUE_BASE, ATLAS_BASE) still present. - submit() -> wait_for_job() flow runs end-to-end against mock adapters. - Observability row is written on terminal events. - Env-var based provider/tier routing (SEEDANCE_PROVIDER / SEEDANCE_TIER) via get_client() continues to produce a functional client. CP-2 Phase 8 (2026-04-26): SeedDanceClient was migrated to execution/providers/fal.py + atlas.py. Provider-level behavior is exercised by tests/execution/providers/. This module is skipped at collect time and kept for git history; CP-3 will delete it. """ from __future__ import annotations import json import os import sys from pathlib import Path import pytest pytest.skip( "CP-2 Phase 8 — SeedDanceClient migrated to execution/providers/. " "Provider tests live under tests/execution/providers/.", allow_module_level=True, ) _RECOIL = Path(__file__).resolve().parents[2] if str(_RECOIL) not in sys.path: sys.path.insert(0, str(_RECOIL)) @pytest.fixture(autouse=True) def _env(monkeypatch, tmp_path): monkeypatch.setenv("RECOIL_PROVIDER_MODE", "test") monkeypatch.setenv("FAL_KEY", "x") monkeypatch.setenv("ATLAS_CLOUD_API_KEY", "x") monkeypatch.setenv("PIAPI_API_KEY", "x") monkeypatch.delenv("RECOIL_PROVIDER_OVERRIDE", raising=False) monkeypatch.delenv("SEEDANCE_PROVIDER", raising=False) monkeypatch.delenv("SEEDANCE_TIER", raising=False) # Redirect observability DB to a temp location. from recoil.execution.providers import observability as obs monkeypatch.setattr(obs, "_DB_PATH", tmp_path / "obs.sqlite", raising=True) from recoil.execution.providers.registry import reset_caches_for_tests reset_caches_for_tests() yield reset_caches_for_tests() def _patch_http(monkeypatch, responses): """Replace execution.video_model_client._http with a scripted responder. `responses` is a list of dicts returned in order. Each element is the parsed JSON the real `_http` would return. """ it = iter(responses) def fake_http(method, url, headers, body=None, timeout=30): return next(it) from recoil.execution import video_model_client as vmc monkeypatch.setattr(vmc, "_http", fake_http, raising=True) # Also neuter _download_video to avoid real network. import recoil.execution.api_client as ac monkeypatch.setattr(ac, "_download_video", lambda url, timeout=60: b"MP4", raising=True) def test_seeddance_client_public_surface_preserved(): from recoil.execution.api_client import SeedDanceClient c = SeedDanceClient(provider="fal", tier="pro") assert c._resolve_endpoint("t2v") == "bytedance/seedance-2.0/text-to-video" assert c._resolve_endpoint("video") == "bytedance/seedance-2.0/image-to-video" assert c._resolve_endpoint("r2v") == "bytedance/seedance-2.0/reference-to-video" assert c._VALID_PROVIDERS == ("fal", "atlas") assert c._VALID_TIERS == ("pro", "fast") assert c.QUEUE_BASE == "https://queue.fal.run" assert c.ATLAS_BASE == "https://api.atlascloud.ai/api/v1/model" assert c.is_available() is True def test_seeddance_client_fast_endpoint_paths(): from recoil.execution.api_client import SeedDanceClient c_fal_fast = SeedDanceClient(provider="fal", tier="fast") assert c_fal_fast._resolve_endpoint("video") == "bytedance/seedance-2.0/fast/image-to-video" c_atlas_fast = SeedDanceClient(provider="atlas", tier="fast") assert c_atlas_fast._resolve_endpoint("video") == "bytedance/seedance-2.0-fast/image-to-video" def test_seeddance_client_rejects_bad_args(): from recoil.execution.api_client import SeedDanceClient with pytest.raises(ValueError): SeedDanceClient(provider="nope") with pytest.raises(ValueError): SeedDanceClient(tier="turbo") def test_submit_and_wait_happy_path_via_mocks(monkeypatch): """Default routing (fal primary) completes end-to-end.""" responses = [ {"request_id": "req-1", "status_url": "http://mock/status", "response_url": "http://mock/res"}, # submit {"status": "COMPLETED"}, # poll -> COMPLETED {"video": {"url": "https://mock.example/out.mp4"}}, # result fetch ] _patch_http(monkeypatch, responses) # Patch time.sleep to avoid delays import recoil.execution.video_model_client as vmc monkeypatch.setattr(vmc.time, "sleep", lambda s: None, raising=True) from recoil.execution.api_client import SeedDanceClient c = SeedDanceClient(provider="fal", tier="pro") job = c.submit({"prompt": "hello", "duration": 5, "resolution": "720p"}) assert job.status == "submitted" # wait_for_job drives poll + result fetch result = c.wait_for_job(job, timeout_s=30) assert result.success is True assert result.video_url.endswith(".mp4") assert result.video_data == b"MP4" def test_piapi_route_via_override(monkeypatch): """Force piapi via RECOIL_PROVIDER_OVERRIDE and exercise PiAPI-shaped responses.""" monkeypatch.setenv("RECOIL_PROVIDER_OVERRIDE", "piapi") responses = [ {"data": {"task_id": "t-piapi"}}, {"data": { "status": "completed", "output": {"video": {"url": "https://mock.example/piapi.mp4"}}, "meta": {"usage": {"consume": 50.0}}, }}, ] _patch_http(monkeypatch, responses) import recoil.execution.video_model_client as vmc monkeypatch.setattr(vmc.time, "sleep", lambda s: None, raising=True) from recoil.execution.video_model_client import VideoModelClient c = VideoModelClient(model_id="seeddance-2.0") job = c.submit({"prompt": "text only"}) r = c.wait_for_job(job, timeout_s=30) assert r.success is True assert r.metadata["provider"] == "piapi" assert r.metadata["cost_usd_billed"] == pytest.approx(0.50) def test_observability_row_written_on_complete(monkeypatch): monkeypatch.setenv("RECOIL_PROVIDER_OVERRIDE", "piapi") responses = [ {"data": {"task_id": "t2"}}, {"data": { "status": "completed", "output": {"video": {"url": "https://mock.example/o.mp4"}}, "meta": {"usage": {"consume": 40.0}}, }}, ] _patch_http(monkeypatch, responses) import recoil.execution.video_model_client as vmc monkeypatch.setattr(vmc.time, "sleep", lambda s: None, raising=True) from recoil.execution.video_model_client import VideoModelClient from recoil.execution.providers import observability as obs c = VideoModelClient(model_id="seeddance-2.0") r = c.wait_for_job(c.submit({"prompt": "x"})) assert r.success rows = obs.query_drift(since_days=1, min_samples=1) assert any(row["provider"] == "piapi" for row in rows) def test_get_client_factory_still_returns_seeddance_client(): from recoil.execution.api_client import get_client, SeedDanceClient c = get_client("seeddance-2.0") assert isinstance(c, SeedDanceClient) def test_model_profile_schema_unchanged(): """Regression: we didn't accidentally mutate model_profiles.json. PiAPI was intentionally removed from the seeddance-2.0 providers block (REC-46, 2026-06-03 — never run, unverified pricing). The live providers are fal.ai (fallback), flora (primary dispatch), and atlas (deprecated). """ profile_path = _RECOIL / "config" / "model_profiles.json" data = json.loads(profile_path.read_text()) sd = data["seeddance-2.0"] assert "providers" in sd assert set(sd["providers"].keys()) >= {"fal.ai", "flora", "atlas"} assert "piapi" not in sd["providers"]