Spaces:
Paused
Paused
| """Server integration test: the full wizard flow over HTTP with mock adapters. | |
| Covers the job lifecycle (submit -> poll -> result), artifact endpoints, | |
| static file serving of the rendered MP4, and the clone route. | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import time | |
| import pytest | |
| # Mock adapters must be selected BEFORE server.app builds its Pipeline. | |
| os.environ.update( | |
| ADAPTER_LLM="mock", ADAPTER_TTS="mock", ADAPTER_STOCK="mock", | |
| ADAPTER_SFX="mock", ADAPTER_MUSIC="mock", ADAPTER_REFDATA="mock", | |
| ADAPTER_TRANSCRIPT="mock", | |
| ) | |
| from tests.mocks import register_mocks # noqa: E402 | |
| register_mocks() | |
| from fastapi.testclient import TestClient # noqa: E402 | |
| from core.stages import topics as topics_stage # noqa: E402 | |
| from server.app import app # noqa: E402 | |
| client = TestClient(app) | |
| def wait_job(job_id: str, timeout: float = 120.0) -> dict: | |
| deadline = time.time() + timeout | |
| while time.time() < deadline: | |
| job = client.get(f"/api/jobs/{job_id}").json() | |
| if job["status"] == "done": | |
| return job["result"] | |
| if job["status"] == "error": | |
| raise AssertionError(f"job {job['kind']} failed: {job['error']}") | |
| time.sleep(0.2) | |
| raise AssertionError("job timed out") | |
| def project_id(request): | |
| resp = client.post("/api/projects", json={ | |
| "name": "Server Smoke", | |
| "niche": {"topic": "deep sea creatures", "keywords": ["ocean"], | |
| "subreddits": ["ocean"], "audience": "everyone", "tone": "fun"}, | |
| "format": "youtube_short", | |
| "references": [ | |
| {"kind": "youtube", "name": "@deepseadaily"}, | |
| {"kind": "instagram_style", "name": "oceanreels", | |
| "url": "https://instagram.com/oceanreels", "notes": "big captions"}, | |
| ], | |
| }) | |
| assert resp.status_code == 200, resp.text | |
| pid = resp.json()["id"] | |
| request.addfinalizer(lambda: client.delete(f"/api/projects/{pid}")) | |
| return pid | |
| def test_health(): | |
| body = client.get("/api/health").json() | |
| assert body["ok"] and "caption_presets" in body | |
| assert "tts_providers" in body | |
| def test_wizard_flow(project_id, monkeypatch): | |
| monkeypatch.setattr(topics_stage, "_news_signals", lambda p: []) | |
| monkeypatch.setattr(topics_stage, "_reddit_signals", lambda p: []) | |
| # style: prompt prepare (job) then refresh, then edit-and-save the output | |
| job = client.post(f"/api/projects/{project_id}/style/prompt").json() | |
| assert "STYLE CARD" in wait_job(job["job_id"])["prompt"] | |
| job = client.post(f"/api/projects/{project_id}/style/refresh", json={}).json() | |
| style = wait_job(job["job_id"]) | |
| assert style["style_card"] and style["prompt_used"] | |
| style["style_card"] = "EDITED style card" | |
| assert client.put(f"/api/projects/{project_id}/style", json=style).json()["style_card"] == "EDITED style card" | |
| # suggest references (job-based) | |
| job = client.post(f"/api/projects/{project_id}/references/suggest", json={"n": 3}).json() | |
| assert isinstance(wait_job(job["job_id"]), list) | |
| # topics: prompt prepare + generate + inline edit | |
| job = client.post(f"/api/projects/{project_id}/topics/prompt", json={"n": 5}).json() | |
| assert wait_job(job["job_id"])["prompt"] | |
| job = client.post(f"/api/projects/{project_id}/topics/generate", json={"n": 5}).json() | |
| topics = wait_job(job["job_id"])["topics"] | |
| assert len(topics) == 5 | |
| topics[0]["title"] = "Edited topic title" | |
| resp = client.put(f"/api/projects/{project_id}/topics", json={"topics": topics}) | |
| assert resp.json()["topics"][0]["title"] == "Edited topic title" | |
| # create a reel for the chosen topic | |
| reel = client.post(f"/api/projects/{project_id}/reels", | |
| json={"topic": "Edited topic title"}).json() | |
| rid = reel["id"] | |
| base = f"/api/projects/{project_id}/reels/{rid}" | |
| # script: prompt prepare + generate + patch + scene regen | |
| assert "scripts" in client.post(f"{base}/script/prompt", json={"topic": "Edited topic title"}).json()["prompt"].lower() | |
| job = client.post(f"{base}/script/generate", json={"topic": "Edited topic title"}).json() | |
| script = wait_job(job["job_id"]) | |
| assert len(script["hook_options"]) == 3 and script["prompt_used"] | |
| script["selected_hook"] = 1 | |
| assert client.patch(f"{base}/script", json=script).json()["selected_hook"] == 1 | |
| job = client.post(f"{base}/script/regenerate-scene", json={"scene": 1}).json() | |
| assert "Rewritten" in wait_job(job["job_id"])["scenes"][0]["narration"] | |
| # voices listing (with providers) + voice synthesis | |
| vinfo = client.get("/api/voices").json() | |
| assert vinfo["voices"] and "providers" in vinfo | |
| job = client.post(f"{base}/voice", json={"voice": "", "provider": "mock"}).json() | |
| assert wait_job(job["job_id"])["total_duration_sec"] > 3 | |
| # media + swap | |
| job = client.post(f"{base}/media").json() | |
| media = wait_job(job["job_id"]) | |
| assert media["scenes"] and media["scenes"][0]["candidates"] | |
| before = media["scenes"][0]["selected"] | |
| swapped = client.post(f"{base}/media/swap", json={"line_index": 0}).json() | |
| assert swapped["scenes"][0]["selected"] != before | |
| # render one variant, then fetch the file over HTTP | |
| job = client.post(f"{base}/render", json={ | |
| "variants": [{"name": "Variant A", "caption_preset": "bold-center-karaoke", "clip_offset": 0}], | |
| }).json() | |
| renders = wait_job(job["job_id"], timeout=300) | |
| assert renders["results"] | |
| url = renders["results"][0]["url"] | |
| assert url.startswith("/files/") | |
| video = client.get(url) | |
| assert video.status_code == 200 and len(video.content) > 50_000 | |
| # reel payload includes everything | |
| payload = client.get(base).json() | |
| assert payload["script"] and payload["voice"] and payload["media"] and payload["renders"] | |
| # project payload lists the reel | |
| proj = client.get(f"/api/projects/{project_id}").json() | |
| assert any(r["id"] == rid for r in proj["reels"]) | |
| def test_multiple_reels_via_api(project_id): | |
| r1 = client.post(f"/api/projects/{project_id}/reels", json={"topic": "alpha"}).json() | |
| r2 = client.post(f"/api/projects/{project_id}/reels", json={"topic": "beta"}).json() | |
| reels = client.get(f"/api/projects/{project_id}/reels").json() | |
| ids = {r["id"] for r in reels} | |
| assert {r1["id"], r2["id"]} <= ids | |
| assert client.delete(f"/api/projects/{project_id}/reels/{r1['id']}").json()["ok"] | |
| remaining = {r["id"] for r in client.get(f"/api/projects/{project_id}/reels").json()} | |
| assert r1["id"] not in remaining and r2["id"] in remaining | |
| def test_clone_route(project_id): | |
| job = client.post(f"/api/projects/{project_id}/clone", | |
| json={"url": "https://www.youtube.com/shorts/abcdefghijk"}).json() | |
| result = wait_job(job["job_id"]) | |
| assert result["reel"]["clone_source"].endswith("abcdefghijk") | |
| assert result["script"]["clone_source"].endswith("abcdefghijk") | |
| job = client.post(f"/api/projects/{project_id}/clone", | |
| json={"url": "https://example.com/nope"}).json() | |
| status = None | |
| for _ in range(50): | |
| status = client.get(f"/api/jobs/{job['job_id']}").json() | |
| if status["status"] in ("done", "error"): | |
| break | |
| time.sleep(0.1) | |
| assert status["status"] == "error" and "YouTube" in status["error"] | |
| def test_bad_render_preset(project_id): | |
| reel = client.post(f"/api/projects/{project_id}/reels", json={"topic": "x"}).json() | |
| resp = client.post(f"/api/projects/{project_id}/reels/{reel['id']}/render", json={ | |
| "variants": [{"name": "X", "caption_preset": "nope", "clip_offset": 0}], | |
| }) | |
| assert resp.status_code == 400 | |