Spaces:
Runtime error
Runtime error
Refactor VideoTemplateEditor component: remove AI editing features and related state management, streamline template loading, and update styles for improved layout. Enhance CSS for responsive design and visual consistency across various components, including dropdowns, sync stats, and quiz elements. Remove deprecated premium badge styles and adjust layout for chat components.
66de9c8 | """Videos through blog2video, with a fake blog2video: API key, denylist, the 3-step create (link, post, upload, | |
| multi-link), our own quotas and counters, premium gating, and workspace isolation for videos, templates, voices | |
| and styles (the integration guide's acceptance checklist).""" | |
| import json | |
| import json as jsonlib | |
| import logging | |
| import uuid | |
| from datetime import UTC, datetime, timedelta | |
| import httpx | |
| import pytest | |
| from sqlalchemy import select | |
| from app.config import settings | |
| from app.models import ( | |
| Artifact, | |
| B2VCustomVoice, | |
| B2VStyle, | |
| B2VTemplate, | |
| B2VVideo, | |
| Chat, | |
| Citation, | |
| Document, | |
| DocumentTopic, | |
| Job, | |
| Message, | |
| Notebook, | |
| Subscription, | |
| Topic, | |
| UsageCounter, | |
| User, | |
| UserSavedVoice, | |
| VoiceConsent, | |
| VoiceProfile, | |
| Workspace, | |
| ) | |
| from app.services import blog2video, video_focus, video_limits, video_quota | |
| from tests.conftest import last_code | |
| KEY = "b2v_live_test_key" | |
| API = "https://b2v.test" | |
| SARAH, BELLA = "EXAVITQu4vr4xnSDxMaL", "hpp4J3VqNfWAUOO0d1Us" # a free and a paid built-in voice | |
| DEFAULTS: dict[str, object] = { | |
| "GET /api/templates": [{"id": "default", "name": "Default"}, {"id": "geometric", "name": "Geometric", | |
| "preview_colors": {"accent": "#FF0000"}}], | |
| "GET /api/crafted-templates": [{"id": "crafted_neon", "name": "Neon"}], | |
| "GET /api/voices/prebuilt": {"voices": [ | |
| {"voice_id": SARAH, "name": "Sarah - Mature", "preview_url": "https://cdn.test/sarah.mp3", | |
| "labels": {"gender": "female", "accent": "american"}, "description": "Warm", "plan": "free"}, | |
| {"voice_id": BELLA, "name": "Bella - Bright", "preview_url": "https://cdn.test/bella.mp3", | |
| "labels": {"gender": "female", "accent": "british"}, "description": "Bright", "plan": "paid"}]}, | |
| "GET /api/background-music/tracks": [{"track_id": "corporate_upbeat", "display_name": "Corporate Upbeat"}], | |
| "GET /api/video-styles": {"styles": [ | |
| {"id": "explainer", "name": "Explainer", "kind": "builtin"}, | |
| {"id": "your_style", "name": "Your Style", "kind": "learned"}, | |
| {"id": "custom:7", "custom_id": 7, "name": "Mine", "kind": "custom", "guidance": "Short."}, | |
| {"id": "custom:9", "custom_id": 9, "name": "Theirs", "kind": "custom", "guidance": "Long."}]}, | |
| "POST /api/voices/custom": {"id": 31, "voice_id": "vc_designed", "name": "Narrator", "preview_url": None}, | |
| "POST /api/voices/clone": {"id": 32, "voice_id": "vc_clone", "name": "Me", "preview_url": None}, | |
| "POST /api/voices/design-from-prompt": {"previews": [{"generated_voice_id": "g1", "audio_base_64": "AAA", | |
| "media_type": "audio/mpeg", "duration_secs": 3}]}, | |
| "POST /api/video-styles/custom": {"id": "custom:41", "custom_id": 41, "name": "Punchy", "guidance": "Punchy."}, | |
| "POST /api/custom-templates": {"id": 58, "name": "Brand"}, | |
| "POST /api/custom-templates/extract-theme-from-prompt": {"extractable": True, "theme": {"colors": {}}}, | |
| "GET /api/auth/me": {"plan": "pro", "video_limit": 150, "videos_used_this_period": 145, "can_create_video": True, | |
| "ai_edit_allowance_remaining": 4000, "ai_edit_credits": 0, "custom_template_limit": 20, | |
| "custom_templates_created": 3}, | |
| } | |
| class FakeB2V: | |
| """Records every call; answers from a queue of (status, body) per "METHOD path", else a sensible default.""" | |
| def __init__(self): | |
| self.calls: list[dict] = [] | |
| self.queued: dict[str, list[tuple[int, object]]] = {} | |
| self.next_id = 812 | |
| def queue(self, method: str, path: str, *answers: tuple[int, object]): | |
| self.queued.setdefault(f"{method} {path}", []).extend(answers) | |
| def paths(self, method: str | None = None) -> list[str]: | |
| return [c["path"] for c in self.calls if method is None or c["method"] == method] | |
| def __call__(self, method, url, json=None, params=None, data=None, files=None, timeout=None, headers=None): | |
| if json is not None: | |
| jsonlib.dumps(json) # what httpx does: anything that is not plain JSON fails here, as it would for real | |
| path = url.removeprefix(API) | |
| key = f"{method} {path}" | |
| self.calls.append({"method": method, "path": path, "json": json, "data": data, "files": files, | |
| "headers": headers or {}, "params": params}) | |
| answers = self.queued.get(key) | |
| ctype = "application/json" | |
| if answers: | |
| status, body = answers.pop(0) | |
| if isinstance(body, Exception): | |
| raise body | |
| elif key in DEFAULTS: | |
| status, body = 200, DEFAULTS[key] | |
| elif key in ("POST /api/v1/videos", "POST /api/projects/upload"): | |
| vid, self.next_id = self.next_id, self.next_id + 1 | |
| status, body = (202, {"video_id": vid, "state": "queued"}) if "v1" in path else (200, {"id": vid}) | |
| elif path.startswith("/api/embed/token/"): | |
| status, body = 200, {"embed_token": "t", "preview_url": f"{API}/embed/t"} | |
| elif path.endswith("/status"): | |
| status, body = 200, {"status": "scripted", "step": 2, "running": True, "ready": False, "error": None, | |
| "video_url": None} | |
| elif method == "GET" and path.startswith("/api/projects/") and path.count("/") == 3: | |
| status, body = 200, {"id": int(path.rsplit("/", 1)[1]), "status": "generated", "r2_video_url": None, | |
| "scenes": [{"id": 2, "order": 2, "title": "B"}, {"id": 1, "order": 1, "title": "A"}]} | |
| elif path in ("/api/voice/preview",) or path.endswith("/download"): | |
| return httpx.Response(200, content=b"ID3audio", headers={"content-type": "audio/mpeg"}, | |
| request=httpx.Request(method, url)) | |
| else: | |
| status, body = 200, {"ok": True} | |
| return httpx.Response(status, content=_json(body), headers={"content-type": ctype}, | |
| request=httpx.Request(method, url)) | |
| def _json(body) -> bytes: | |
| return json.dumps(body).encode() | |
| def b2v(monkeypatch): | |
| fake = FakeB2V() | |
| monkeypatch.setattr(settings, "b2v_api_base_url", API) | |
| monkeypatch.setattr(settings, "b2v_api_key", KEY) | |
| monkeypatch.setattr(blog2video.httpx, "request", fake) | |
| monkeypatch.setattr(blog2video.time, "sleep", lambda s: None) | |
| blog2video.clear_cache() | |
| return fake | |
| def register(client, email: str) -> dict: | |
| client.post("/api/auth/email/register/start", json={"email": email, "password": "stardust-42", "name": "Ada"}) | |
| r = client.post("/api/auth/email/register/verify", json={"email": email, "code": last_code()}) | |
| return {"Authorization": f"Bearer {r.json()['access_token']}"} | |
| def owner(client, db_session): | |
| headers = register(client, "ada@example.com") | |
| return headers, db_session.scalar(select(Workspace)) | |
| def other(client, db_session, owner): | |
| headers = register(client, "grace@example.com") | |
| ws = db_session.scalar(select(Workspace).where(Workspace.id != owner[1].id)) | |
| return headers, ws | |
| def free_plan(monkeypatch): | |
| monkeypatch.setattr(settings, "billing_enabled", True) | |
| def add_doc(db, ws, url: str, text: str = "") -> Document: | |
| doc = Document(workspace_id=ws.id, title="On Pricing", url=url, clean_text=text, path="sources/x/pricing.md") | |
| db.add(doc) | |
| db.commit() | |
| return doc | |
| def creates(fake: FakeB2V) -> list[dict]: | |
| return [c for c in fake.calls if c["method"] == "POST" and c["path"] == "/api/v1/videos"] | |
| def used(db, ws) -> int: | |
| db.expire_all() | |
| return db.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)).videos_used | |
| def make_video(client, headers, **body) -> dict: | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a", **body}, headers=headers) | |
| assert r.status_code == 201, r.text | |
| return r.json() | |
| def age(db, row: B2VVideo) -> None: | |
| row.created_at = datetime.now(UTC) - timedelta(hours=1) | |
| db.commit() | |
| # Client | |
| def test_sends_our_api_key(b2v): | |
| blog2video.account() | |
| assert b2v.calls[0]["headers"]["Authorization"] == f"Bearer {KEY}" | |
| def test_create_retries_with_the_same_idempotency_key(b2v): | |
| b2v.queue("POST", "/api/v1/videos", (0, httpx.ConnectTimeout("slow")), (503, {"detail": "busy"})) | |
| assert blog2video.create_video({"url": "https://a.test"}, "ws:1")["video_id"] == 812 | |
| assert [c["headers"]["Idempotency-Key"] for c in b2v.calls] == ["ws:1"] * 3 | |
| def test_other_calls_are_not_retried(b2v): | |
| b2v.queue("GET", "/api/auth/me", (500, {"detail": "boom"})) | |
| with pytest.raises(blog2video.Upstream): | |
| blog2video.account() | |
| assert len(b2v.calls) == 1 | |
| def test_account_problems_alert_ops(b2v, caplog, status): | |
| b2v.queue("GET", "/api/auth/me", (status, {"detail": {"error": "x"}})) | |
| with caplog.at_level(logging.ERROR), pytest.raises(blog2video.Upstream if status != 402 | |
| else blog2video.AccountOutOfVideos): | |
| blog2video.account() | |
| assert "ALERT" in caplog.text and KEY not in caplog.text | |
| def test_account_wide_endpoints_are_never_called(b2v, method, path): | |
| with pytest.raises(blog2video.Denied): | |
| blog2video.request(method, path) | |
| assert b2v.calls == [] | |
| # Create: the 3 steps | |
| def test_link_create_forwards_every_wizard_field(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| body = {"stock_footage_enabled": False, "aspect_ratio": "portrait", | |
| "video_length": "medium", "logo_position": "top_left", "logo_opacity": 0.5, "template": "geometric", | |
| "video_style": "storytelling", "accent_color": "#FF5A1F", "bg_color": "#0B0B0F", "text_color": "#FFFFFF", | |
| "content_language": "es", "voice_gender": "female", "voice_accent": "british", | |
| "custom_voice_id": SARAH, | |
| "bgm_track_id": "corporate_upbeat", "bgm_volume": 0.2} | |
| art = make_video(client, headers, **body) | |
| call = creates(b2v)[0] | |
| for k, v in body.items(): | |
| assert call["json"][k] == v, k | |
| assert call["json"]["script_review_enabled"] is False | |
| assert call["headers"]["Idempotency-Key"].startswith(f"{ws.id}:") | |
| assert call["json"]["external_user_id"] == str(ws.id) | |
| row = db_session.scalar(select(B2VVideo)) | |
| assert (row.b2v_video_id, row.quota_state, row.created_via, row.template_ref) == (812, "charged", "v1", "geometric") | |
| assert art["content"]["b2v_video_id"] == 812 and used(db_session, ws) == 1 | |
| def test_script_review_is_always_off(client, b2v, owner): | |
| make_video(client, owner[0], script_review_enabled=True) # an old client asking for it is ignored | |
| assert creates(b2v)[0]["json"]["script_review_enabled"] is False | |
| def test_create_from_feed_post_and_uploaded_post(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/pricing") | |
| client.post("/api/videos", json={"document_id": str(doc.id)}, headers=headers) | |
| assert creates(b2v)[-1]["json"]["url"] == "https://ada.example.com/p/pricing" | |
| doc = add_doc(db_session, ws, "upload://notes.md", "Charging more made my readers take the work seriously. " * 3) | |
| client.post("/api/videos", json={"document_id": str(doc.id)}, headers=headers) | |
| sent = creates(b2v)[-1]["json"] | |
| assert "url" not in sent and sent["content"].startswith("Charging more") | |
| def test_notebook_posts_make_one_combined_video(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| new = add_doc(db_session, ws, "https://ada.example.com/p/new", "Raise prices once a year, and say why. " * 3) | |
| old = add_doc(db_session, ws, "https://ada.example.com/p/old", "Start cheaper than you think you should. " * 3) | |
| new.title, new.published_at = "Annual raises", datetime(2026, 5, 1, tzinfo=UTC) | |
| old.title, old.published_at = "Starting out", datetime(2025, 1, 1, tzinfo=UTC) | |
| nb = Notebook(workspace_id=ws.id, title="Pricing series") | |
| db_session.add(nb) | |
| db_session.commit() | |
| r = client.post("/api/videos", json={"document_ids": [str(new.id), str(old.id)], "notebook_id": str(nb.id)}, | |
| headers=headers) | |
| assert r.status_code == 201 and r.json()["content"]["title"] == "Pricing series" | |
| sent = creates(b2v)[-1]["json"] | |
| assert len(creates(b2v)) == 1 and "url" not in sent and sent["title"] == "Pricing series" | |
| assert not {"document_ids", "notebook_id", "document_id"} & set(sent) # the source is not a wizard option | |
| assert sent["content"].index("# Starting out") < sent["content"].index("# Annual raises") | |
| assert used(db_session, ws) == 1 | |
| def test_one_ticked_post_goes_by_its_link(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/pricing") | |
| client.post("/api/videos", json={"document_ids": [str(doc.id)]}, headers=headers) | |
| assert creates(b2v)[-1]["json"]["url"] == "https://ada.example.com/p/pricing" | |
| def test_locked_posts_are_refused_before_charging(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| ok = add_doc(db_session, ws, "https://ada.example.com/p/a", "Enough text to make a video from, really. " * 3) | |
| locked = add_doc(db_session, ws, "https://ada.example.com/p/b") | |
| locked.path, locked.metadata_json = None, {"locked": True} | |
| db_session.commit() | |
| for body in ({"document_id": str(locked.id)}, {"document_ids": [str(ok.id), str(locked.id)]}): | |
| r = client.post("/api/videos", json=body, headers=headers) | |
| assert r.status_code == 400 and "not indexed" in r.json()["detail"] | |
| assert not creates(b2v) and used(db_session, ws) == 0 | |
| def add_chat(db, ws, turns: list[tuple[str, str]], title: str | None = "Pricing chat") -> Chat: | |
| nb = Notebook(workspace_id=ws.id, title="Pricing series") | |
| db.add(nb) | |
| db.flush() | |
| chat = Chat(workspace_id=ws.id, notebook_id=nb.id, title=title) | |
| db.add(chat) | |
| db.flush() | |
| for i, (role, text) in enumerate(turns): | |
| db.add(Message(chat_id=chat.id, role=role, content=text, | |
| created_at=datetime(2026, 1, 1, tzinfo=UTC) + timedelta(minutes=i))) | |
| db.commit() | |
| return chat | |
| def test_chat_is_sent_as_its_transcript_after_the_context_prompt(client, b2v, owner, db_session): | |
| from app.routers.videos import CHAT_CONTEXT | |
| headers, ws = owner | |
| chat = add_chat(db_session, ws, [("user", "How often should I raise prices?"), | |
| ("assistant", "Once a year, and tell readers why [1]. Grandfather the early ones [2].")]) | |
| r = client.post("/api/videos", json={"chat_ids": [str(chat.id)]}, headers=headers) | |
| assert r.status_code == 201 and r.json()["content"]["title"] == "Pricing chat" | |
| sent = creates(b2v)[-1]["json"] | |
| assert sent["title"] == "Pricing chat" and "chat_ids" not in sent and "url" not in sent | |
| assert sent["content"].startswith(CHAT_CONTEXT) | |
| assert "User: How often should I raise prices?" in sent["content"] | |
| assert "Assistant: Once a year, and tell readers why. Grandfather the early ones." in sent["content"] | |
| assert used(db_session, ws) == 1 | |
| def test_several_chats_make_one_video(client, b2v, owner, other, db_session): | |
| from app.routers.videos import CHAT_CONTEXT | |
| headers, ws = owner | |
| a = add_chat(db_session, ws, [("user", "How often should I raise prices?"), | |
| ("assistant", "Once a year, and tell readers why it is happening.")]) | |
| b = add_chat(db_session, ws, [("user", "Should early readers keep their price?"), | |
| ("assistant", "Yes, grandfather them: they took the risk first.")], title="Early readers") | |
| r = client.post("/api/videos", json={"chat_ids": [str(a.id), str(b.id)]}, headers=headers) | |
| assert r.status_code == 201 and r.json()["content"]["title"] == "Pricing chat + 1 more" | |
| content = creates(b2v)[-1]["json"]["content"] | |
| assert content.startswith(CHAT_CONTEXT) | |
| assert content.index("# Pricing chat") < content.index("User: How often") < content.index("# Early readers") | |
| assert "Assistant: Yes, grandfather them" in content | |
| # Six is over the cap; one chat that is not ours makes it 404. | |
| six = [str(add_chat(db_session, ws, [("user", "Q?"), ("assistant", "A long enough answer.")]).id) for _ in range(6)] | |
| assert client.post("/api/videos", json={"chat_ids": six}, headers=headers).status_code == 422 | |
| theirs = add_chat(db_session, other[1], [("user", "Q?"), ("assistant", "A.")]) | |
| r = client.post("/api/videos", json={"chat_ids": [str(a.id), str(theirs.id)]}, headers=headers) | |
| assert r.status_code == 404 and len(creates(b2v)) == 1 | |
| def test_chats_list_across_notebooks_only_answered_ones(client, owner, other, db_session): | |
| headers, ws = owner | |
| answered = add_chat(db_session, ws, [("user", "Q?"), ("assistant", "A.")]) | |
| add_chat(db_session, ws, [("user", "Unanswered?")]) | |
| add_chat(db_session, other[1], [("user", "Q?"), ("assistant", "A.")]) | |
| r = client.get("/api/notebooks/chats", headers=headers) | |
| assert r.status_code == 200 | |
| assert [(c["id"], c["notebook_title"]) for c in r.json()] == [(str(answered.id), "Pricing series")] | |
| def test_chat_source_is_checked(client, b2v, owner, other, db_session): | |
| headers, ws = owner | |
| theirs = add_chat(db_session, other[1], [("user", "Q?"), ("assistant", "A long enough answer, really.")]) | |
| assert client.post("/api/videos", json={"chat_ids": [str(theirs.id)]}, headers=headers).status_code == 404 | |
| empty = add_chat(db_session, ws, [("user", "Anyone there?")]) | |
| assert client.post("/api/videos", json={"chat_ids": [str(empty.id)]}, headers=headers).status_code == 400 | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a") | |
| r = client.post("/api/videos", json={"chat_ids": [str(empty.id)], "document_id": str(doc.id)}, headers=headers) | |
| assert r.status_code == 422 | |
| assert not creates(b2v) and used(db_session, ws) == 0 | |
| def test_another_workspaces_post_is_404(client, b2v, owner, other, db_session): | |
| headers, ws = owner | |
| mine = add_doc(db_session, ws, "https://ada.example.com/p/a", "Enough text to make a video from, really. " * 3) | |
| theirs = add_doc(db_session, other[1], "https://bo.example.com/p/b") | |
| r = client.post("/api/videos", json={"document_ids": [str(mine.id), str(theirs.id)]}, headers=headers) | |
| assert r.status_code == 404 and not creates(b2v) | |
| def test_upload_goes_to_blog2video_and_generates(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| files = [("files", ("notes.md", b"# Notes\n" * 20, "text/markdown")), ("files", ("deck.pptx", b"x", "app/x"))] | |
| r = client.post("/api/videos/upload", files=files, data={"options": json.dumps({"video_length": "medium"})}, | |
| headers=headers) | |
| assert r.status_code == 201, r.text | |
| upload = next(c for c in b2v.calls if c["path"] == "/api/projects/upload") | |
| assert [f[1][0] for f in upload["files"]] == ["notes.md", "deck.pptx"] | |
| assert upload["data"]["video_length"] == "medium" and upload["data"]["script_review_enabled"] == "false" | |
| assert "/api/projects/812/generate" in b2v.paths("POST") | |
| row = db_session.scalar(select(B2VVideo)) | |
| assert row.created_via == "upload" and row.idempotency_key is None | |
| client.get(f"/api/videos/{r.json()['id']}/status", headers=headers) | |
| assert "/api/projects/812/status" in b2v.paths("GET") # uploads are not under /api/v1 | |
| def test_upload_rejects_wrong_files_before_charging(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| r = client.post("/api/videos/upload", files=[("files", ("a.exe", b"x", "app/x"))], headers=headers) | |
| assert r.status_code == 400 and used(db_session, ws) == 0 and b2v.calls == [] | |
| def test_multi_link_makes_one_video_per_link(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| r = client.post("/api/videos/batch", json={"urls": ["https://a.test/1", "https://a.test/2", "nope"], | |
| "bgm_track_id": "corporate_upbeat"}, headers=headers) | |
| assert r.status_code == 201 and len(r.json()) == 2 | |
| calls = creates(b2v) | |
| assert len({c["headers"]["Idempotency-Key"] for c in calls}) == 2 | |
| assert all("bgm_track_id" not in c["json"] for c in calls) # music is single-video only | |
| assert used(db_session, ws) == 2 | |
| def test_our_account_out_of_videos_refunds_the_user(client, b2v, owner, db_session, caplog): | |
| headers, ws = owner | |
| b2v.queue("POST", "/api/v1/videos", (402, {"detail": {"error": "quota_exceeded"}})) | |
| with caplog.at_level(logging.ERROR): | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a"}, headers=headers) | |
| assert r.status_code == 503 and "temporarily unavailable" in r.json()["detail"] and "ALERT" in caplog.text | |
| assert db_session.scalar(select(Artifact).where(Artifact.type == "video")) is None | |
| assert db_session.scalar(select(B2VVideo)).quota_state == "refunded" and used(db_session, ws) == 0 | |
| def test_validation_error_refunds_and_says_what_was_wrong(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| b2v.queue("POST", "/api/v1/videos", (422, {"detail": [{"loc": ["body", "url"], "msg": "could not fetch"}]})) | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a"}, headers=headers) | |
| assert r.status_code == 400 and "url: could not fetch" in r.json()["detail"] and used(db_session, ws) == 0 | |
| def test_slow_catalog_is_a_clean_error_not_a_500(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| b2v.queue("GET", "/api/templates", (0, httpx.ReadTimeout("slow"))) | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a", "template": "geometric"}, | |
| headers=headers) | |
| assert r.status_code == 502 and "being updated" in r.json()["detail"] | |
| assert not creates(b2v) and used(db_session, ws) == 0 | |
| def test_expired_catalog_is_used_when_blog2video_is_slow(client, b2v, owner): | |
| make_video(client, owner[0], template="geometric") # fills the catalog cache | |
| for key, (at, value) in list(blog2video._cache.items()): | |
| blog2video._cache[key] = (at - 2 * blog2video.CATALOG_TTL, value) # an hour and more ago | |
| b2v.queue("GET", "/api/templates", (0, httpx.ReadTimeout("slow"))) | |
| make_video(client, owner[0], template="geometric") | |
| assert len(creates(b2v)) == 2 | |
| def test_no_answer_after_retries_refunds(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| b2v.queue("POST", "/api/v1/videos", *[(0, httpx.ConnectError("down"))] * 3) | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a"}, headers=headers) | |
| assert r.status_code == 502 and len(creates(b2v)) == 3 and used(db_session, ws) == 0 | |
| def test_local_limit_blocks_without_calling_out(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| client.get("/api/videos/quota", headers=headers) # syncs the row | |
| sub = db_session.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)) | |
| sub.videos_used = sub.video_limit | |
| db_session.commit() | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a"}, headers=headers) | |
| assert r.status_code == 402 and r.json()["detail"]["code"] == "plan_limit" and not creates(b2v) | |
| def test_two_workspaces_same_url_get_separate_videos(client, b2v, owner, other): | |
| make_video(client, owner[0]) | |
| make_video(client, other[0]) | |
| calls = creates(b2v) | |
| assert calls[0]["headers"]["Idempotency-Key"].split(":")[0] != calls[1]["headers"]["Idempotency-Key"].split(":")[0] | |
| def test_not_configured(client, owner): | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a"}, headers=owner[0]) | |
| assert r.status_code == 503 | |
| # Isolation: another workspace's ids are refused before blog2video is called | |
| def theirs(db_session, other): | |
| _, ws = other | |
| db_session.add_all([B2VTemplate(b2v_template_id=77, workspace_id=ws.id, name="Theirs", ready=True), | |
| B2VStyle(b2v_style_id=9, workspace_id=ws.id, name="Theirs"), | |
| B2VCustomVoice(b2v_custom_voice_id=90, workspace_id=ws.id, voice_id="vc_theirs", name="T", | |
| source="clone")]) | |
| db_session.commit() | |
| def test_foreign_references_are_refused_before_calling_out(client, b2v, owner, theirs, db_session, field, value): | |
| headers, ws = owner | |
| r = client.post("/api/videos", json={"url": "https://ada.example.com/p/a", field: value}, headers=headers) | |
| assert r.status_code == 404 and not creates(b2v) and used(db_session, ws) == 0 | |
| def test_own_references_pass(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| db_session.add_all([B2VTemplate(b2v_template_id=58, workspace_id=ws.id, name="Mine", ready=True), | |
| B2VStyle(b2v_style_id=7, workspace_id=ws.id, name="Mine")]) | |
| db_session.commit() | |
| make_video(client, headers, template="custom_58", video_style="custom:7", custom_voice_id=SARAH) | |
| def test_template_still_generating_is_refused(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| db_session.add(B2VTemplate(b2v_template_id=58, workspace_id=ws.id, name="Mine", ready=False)) | |
| db_session.commit() | |
| r = client.post("/api/videos", json={"url": "https://a.test/x", "template": "custom_58"}, headers=headers) | |
| assert r.status_code == 400 and not creates(b2v) | |
| def test_other_workspace_video_is_404_everywhere(client, b2v, owner, other): | |
| art = make_video(client, owner[0]) | |
| b2v.calls.clear() | |
| for method, path in [("GET", ""), ("GET", "/status"), ("GET", "/script"), ("DELETE", ""), | |
| ("PUT", "/p/scenes/1"), ("POST", "/p/render")]: | |
| r = client.request(method, f"/api/videos/{art['id']}{path}", json={}, headers=other[0]) | |
| assert r.status_code == 404, (method, path) | |
| assert b2v.calls == [] | |
| def test_catalog_shows_only_this_workspace(client, b2v, owner, theirs, db_session): | |
| headers, ws = owner | |
| db_session.add_all([B2VStyle(b2v_style_id=7, workspace_id=ws.id, name="Mine"), | |
| B2VTemplate(b2v_template_id=58, workspace_id=ws.id, name="Mine", ready=True), | |
| B2VTemplate(b2v_template_id=59, workspace_id=ws.id, name="Not yet", ready=False)]) | |
| db_session.commit() | |
| cat = client.get("/api/videos/catalog", headers=headers).json() | |
| assert [s["id"] for s in cat["video_styles"]] == ["explainer", "custom:7"] | |
| assert [t["id"] for t in cat["my_templates"]] == ["custom_58"] | |
| def house(monkeypatch): | |
| """Our account has the two house styles (made on blog2video itself, not by any workspace).""" | |
| styles = DEFAULTS["GET /api/video-styles"]["styles"] + [ | |
| {"id": "custom:21", "custom_id": 21, "name": "Deep dive", "kind": "custom", "guidance": "Context first."}, | |
| {"id": "custom:20", "custom_id": 20, "name": "Overview", "kind": "custom", "guidance": "Neutral."}] | |
| monkeypatch.setitem(DEFAULTS, "GET /api/video-styles", {"styles": styles}) | |
| def test_catalog_offers_only_the_house_styles(client, b2v, owner, house, db_session): | |
| headers, ws = owner | |
| db_session.add(B2VStyle(b2v_style_id=7, workspace_id=ws.id, name="Mine")) | |
| db_session.commit() | |
| styles = client.get("/api/videos/catalog", headers=headers).json()["video_styles"] | |
| assert [(s["id"], s["name"], s["kind"]) for s in styles] == [("custom:20", "Overview", "house"), | |
| ("custom:21", "Deep dive", "house")] | |
| assert all(s["description"] and "length" not in s for s in styles) | |
| def test_house_style_sets_the_length(client, b2v, owner, house, free_plan, style, length): | |
| make_video(client, owner[0], video_style=style, video_length="short" if style == "custom:21" else "medium") | |
| sent = creates(b2v)[-1]["json"] | |
| assert sent["video_style"] == style and sent["video_length"] == length | |
| def test_other_account_styles_stay_refused(client, b2v, owner, house): | |
| r = client.post("/api/videos", json={"url": "https://a.test/x", "video_style": "custom:9"}, headers=owner[0]) | |
| assert r.status_code == 404 and not creates(b2v) | |
| def suggest(monkeypatch): | |
| """The AI's focus suggestions, canned: records what it was given.""" | |
| from app.llm import run | |
| seen: list[dict] = [] | |
| def predict(signature, *, db, workspace_id, job=None, lm=None, **inputs): | |
| seen.append({"signature": signature.__name__, **inputs}) | |
| topic = lambda title, desc="": {"title": title, "description": desc} # noqa: E731 | |
| return {"topics": [topic("Why prices go up.", "Why raises happen. Once a year,\n with notice."), | |
| topic("why prices go up"), topic("Grandfathering early readers", "Keep early prices."), | |
| topic("Annual raises", "One raise a year."), topic("A fourth one")]} | |
| monkeypatch.setattr(run, "predict", predict) | |
| return seen | |
| def test_focus_topics_for_posts_a_chat_and_one_post(client, owner, db_session, suggest): | |
| headers, ws = owner | |
| a = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year, and say why. " * 3) | |
| b = add_doc(db_session, ws, "https://ada.example.com/p/b", "Start cheaper than you think you should. " * 3) | |
| chat = add_chat(db_session, ws, [("user", "How often should I raise prices?"), | |
| ("assistant", "Once a year, and tell readers why it is happening [1].")]) | |
| for body in ({"document_ids": [str(a.id), str(b.id)]}, {"chat_ids": [str(chat.id)]}, {"document_id": str(a.id)}): | |
| r = client.post("/api/videos/focus-topics", json=body, headers=headers) | |
| assert r.status_code == 200, r.text | |
| assert r.json()["topics"] == [ | |
| {"title": "Why prices go up", "description": "Why raises happen. Once a year, with notice."}, | |
| {"title": "Grandfathering early readers", "description": "Keep early prices."}, | |
| {"title": "Annual raises", "description": "One raise a year."}] | |
| assert [s["signature"] for s in suggest] == ["VideoFocusAngles"] * 3 # no topics to show: the AI writes them | |
| # Each item is a title and the start of its text: a little, never the whole post or transcript. | |
| posts, chat_item, one = (s["items"] for s in suggest) | |
| assert len(posts) == 2 and all(p.startswith("On Pricing\n") for p in posts) | |
| assert any("Raise prices once a year" in p for p in posts) and any("Start cheaper" in p for p in posts) | |
| assert chat_item == ["Pricing chat\nHow often should I raise prices?\nOnce a year, and tell readers why it is " | |
| "happening."] | |
| assert one[0].startswith("On Pricing\nRaise prices") | |
| def test_focus_material_is_short_and_untitled_chats_still_get_topics(client, owner, db_session, suggest): | |
| headers, ws = owner | |
| long = add_doc(db_session, ws, "https://ada.example.com/p/a", "Pricing is a promise to your readers. " * 200) | |
| other = add_doc(db_session, ws, "https://ada.example.com/p/b", "Grandfather the people who came early. " * 200) | |
| untitled = add_chat(db_session, ws, [("user", "Should I raise prices?"), ("assistant", "Yes, once a year [1].")], | |
| title=None) | |
| r = client.post("/api/videos/focus-topics", json={"document_ids": [str(long.id), str(other.id)]}, headers=headers) | |
| assert len(r.json()["topics"]) == 3 | |
| assert all(len(i) < 450 and i.endswith("...") for i in suggest[-1]["items"]) # 400 characters of text per post | |
| titled = add_chat(db_session, ws, [("user", "What about early readers?"), ("assistant", "Keep their price.")]) | |
| r = client.post("/api/videos/focus-topics", json={"chat_ids": [str(untitled.id), str(titled.id)]}, headers=headers) | |
| assert len(r.json()["topics"]) == 3 and len(suggest[-1]["items"]) == 2 | |
| r = client.post("/api/videos/focus-topics", json={"chat_ids": [str(untitled.id)]}, headers=headers) | |
| assert len(r.json()["topics"]) == 3 | |
| assert suggest[-1]["items"] == ["Should I raise prices?\nYes, once a year."] | |
| def test_unusable_topics_get_one_fresh_retry(client, owner, db_session, monkeypatch): | |
| from app.llm import run | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year, and say why. " * 3) | |
| lms = [] | |
| answers = [ # what GLM did: the word "Description" as every title, then a proper answer | |
| {"topics": [{"topic": "Description", "summary": f"Angle {i}."} for i in range(3)]}, | |
| {"topics": [{"topic": "Annual raises", "summary": "One raise a year."}, | |
| {"topic": "Grandfathering early readers", "summary": "Keep early prices."}, | |
| {"topic": "Saying why prices go up", "summary": "Explain the raise."}]}, | |
| ] | |
| def predict(signature, *, db, workspace_id, job=None, lm=None, **inputs): | |
| lms.append(lm) | |
| return answers[len(lms) - 1] | |
| monkeypatch.setattr(run, "predict", predict) | |
| r = client.post("/api/videos/focus-topics", json={"document_id": str(doc.id)}, headers=headers) | |
| assert [t["title"] for t in r.json()["topics"]] == ["Annual raises", "Grandfathering early readers", | |
| "Saying why prices go up"] | |
| assert len(lms) == 2 and lms[0] is not lms[1] # the retry is a fresh call, not the cached bad answer | |
| def test_focus_topics_check_the_source(client, owner, other, db_session, suggest): | |
| headers, ws = owner | |
| theirs = add_chat(db_session, other[1], [("user", "Q?"), ("assistant", "A long enough answer, really.")]) | |
| assert client.post("/api/videos/focus-topics", json={"chat_ids": [str(theirs.id)]}, headers=headers).status_code == 404 | |
| locked = add_doc(db_session, ws, "https://ada.example.com/p/b") | |
| locked.path, locked.metadata_json = None, {"locked": True} | |
| db_session.commit() | |
| r = client.post("/api/videos/focus-topics", json={"document_id": str(locked.id)}, headers=headers) | |
| assert r.status_code == 400 and not suggest | |
| # Focus topics: the workspace's own, shown as they are | |
| def angles(monkeypatch): | |
| """The AI, canned: VideoFocusAngles (only when there are no topics to show) answers three fixed angles. | |
| ExtractTopics (tagging on demand) answers per post from `tag_answers` (title -> tags; None = the call failed).""" | |
| from app.llm import run | |
| calls: dict = {"focus": [], "tagged": [], "tag_answers": {}} | |
| def predict(signature, *, db, workspace_id, job=None, lm=None, **inputs): | |
| calls["focus"].append(inputs) | |
| return {"topics": [{"topic": f"AI angle {i}", "summary": "Written by the AI.", "tag": ""} for i in range(3)]} | |
| def predict_many(signature, inputs_list, *, db, workspace_id, lm=None, workers=5): | |
| calls["tagged"].append([i["title"] for i in inputs_list]) | |
| answers = [calls["tag_answers"].get(i["title"], []) for i in inputs_list] | |
| return [None if a is None else {"topics": [{"name": n, "weight": 0.8} for n in a]} for a in answers] | |
| monkeypatch.setattr(run, "predict", predict) | |
| monkeypatch.setattr(run, "predict_many", predict_many) | |
| monkeypatch.setattr(settings, "llm_api_key", "test-key") | |
| return calls | |
| def tagged_doc(db, ws, slug: str, tags: list[str], title: str = "On Pricing", | |
| weights: list[float] | None = None) -> Document: | |
| doc = add_doc(db, ws, f"https://ada.example.com/p/{slug}", f"All about {slug}. " * 20) | |
| weights = weights or [0.8] * len(tags) | |
| doc.title = title | |
| doc.metadata_json = {"topics": [{"name": t, "weight": w} for t, w in zip(tags, weights, strict=True)]} | |
| db.commit() | |
| return doc | |
| def merge(db, ws, name: str, summary: str, *links: tuple[Document, float]) -> Topic: | |
| """A topic of the topic table (what rebuild_topics makes), linked to posts with a weight.""" | |
| topic = Topic(workspace_id=ws.id, name=name, slug=name.lower().replace(" ", "-"), summary=summary, | |
| post_count=len(links)) | |
| db.add(topic) | |
| db.flush() | |
| db.add_all([DocumentTopic(document_id=d.id, topic_id=topic.id, weight=w) for d, w in links]) | |
| db.commit() | |
| return topic | |
| def focus(client, headers, **body) -> list[dict]: | |
| r = client.post("/api/videos/focus-topics", json=body, headers=headers) | |
| assert r.status_code == 200, r.text | |
| return r.json()["topics"] | |
| def test_a_tag_shared_by_the_selection_outranks_one_posts_strongest(owner, db_session): | |
| _, ws = owner | |
| docs = [add_doc(db_session, ws, f"https://ada.example.com/p/{i}", "Text.") for i in range(3)] | |
| merge(db_session, ws, "Pricing", "How to price.", *[(d, 0.5) for d in docs]) | |
| merge(db_session, ws, "Churn", "", (docs[0], 1.0)) | |
| pool = video_focus.build_pool(db_session, ws.id, {str(d.id): {d.id: 1.0} for d in docs}, docs) | |
| assert [c.name for c in pool] == ["Pricing", "Churn"] | |
| assert pool[0].summary == "How to price." and len(pool[0].sources) == 3 and pool[0].merged | |
| def test_a_posts_own_tags_join_the_merged_topics(owner, db_session): | |
| """The merge folds a post's specific tags into a few broad topics; the specific ones still count.""" | |
| _, ws = owner | |
| doc = tagged_doc(db_session, ws, "a", ["Content Repurposing", "Video Marketing", "Finance Writing"]) | |
| merge(db_session, ws, "Finance Substacks", "Growing a finance newsletter.", (doc, 0.9)) | |
| merge(db_session, ws, "Finance Writing", "", (doc, 0.4)) | |
| pool = video_focus.build_pool(db_session, ws.id, {str(doc.id): {doc.id: 1.0}}, [doc]) | |
| assert sorted((c.name, c.merged) for c in pool) == [ | |
| ("Content Repurposing", False), ("Finance Substacks", True), ("Finance Writing", True), # once, merged | |
| ("Video Marketing", False)] | |
| def test_existing_topics_are_shown_as_they_are_without_the_ai(client, owner, db_session, angles): | |
| headers, ws = owner | |
| doc = tagged_doc(db_session, ws, "a", ["Chinese LLMs", "Token Pricing", "Software Margins"], | |
| weights=[0.9, 0.7, 0.5]) | |
| merge(db_session, ws, "AI & China LLM Competition", "Why Chinese LLMs are cheaper.", (doc, 0.9)) | |
| cards = focus(client, headers, document_id=str(doc.id)) | |
| # Merged topics first (with their summary), then the post's own tags (a name only), best first. | |
| assert cards == [{"title": "AI & China LLM Competition", "description": "Why Chinese LLMs are cheaper."}, | |
| {"title": "Chinese LLMs", "description": ""}, | |
| {"title": "Token Pricing", "description": ""}] | |
| assert not angles["focus"] and not angles["tagged"] # no AI at all | |
| def test_a_fresh_tag_that_names_a_merged_topic_is_shown_with_its_summary(client, owner, db_session, angles): | |
| headers, ws = owner | |
| merge(db_session, ws, "Annual Raises", "Once a year, with notice.") | |
| doc = tagged_doc(db_session, ws, "a", ["annual raises"]) | |
| assert focus(client, headers, document_id=str(doc.id)) == [ | |
| {"title": "Annual Raises", "description": "Once a year, with notice."}] # one topic: one card, no AI | |
| assert not angles["focus"] | |
| def test_several_posts_show_their_shared_topics_first_spread_over_the_posts(client, owner, db_session, angles): | |
| headers, ws = owner | |
| a, b, c = (tagged_doc(db_session, ws, s, [f"{s} own"], title=f"Post {s}") for s in "abc") | |
| merge(db_session, ws, "Pricing", "Shared by all.", (a, 0.5), (b, 0.5), (c, 0.5)) | |
| merge(db_session, ws, "A Big", "Post a's main topic.", (a, 1.0)) | |
| merge(db_session, ws, "A Small", "Post a again.", (a, 0.9)) | |
| merge(db_session, ws, "B Side", "Post b's.", (b, 0.8)) | |
| cards = focus(client, headers, document_ids=[str(d.id) for d in (a, b, c)]) | |
| # Pricing (in all three) first; then post a's topics are already covered, so b's comes before a's second one. | |
| assert [x["title"] for x in cards] == ["Pricing", "A Big", "B Side"] | |
| assert not angles["focus"] | |
| def test_untagged_posts_are_tagged_on_demand_then_shown(client, owner, db_session, angles): | |
| headers, ws = owner | |
| docs = [add_doc(db_session, ws, f"https://ada.example.com/p/{i}", "Some text. " * 20) for i in range(7)] | |
| for i, d in enumerate(docs): | |
| d.title, d.published_at = f"Post {i}", datetime(2026, 1, 1 + i, tzinfo=UTC) | |
| db_session.commit() | |
| angles["tag_answers"] = {f"Post {i}": [f"Theme {i}", "Pricing"] for i in range(7)} | |
| angles["tag_answers"]["Post 6"] = None # that call failed | |
| body = {"document_ids": [str(d.id) for d in docs]} | |
| cards = focus(client, headers, **body) | |
| assert angles["tagged"] == [["Post 6", "Post 5", "Post 4", "Post 3", "Post 2"]] # the newest five | |
| assert cards[0] == {"title": "Pricing", "description": ""} and len(cards) == 3 # in 4 posts: first | |
| assert not angles["focus"] | |
| db_session.expire_all() | |
| assert [d.metadata_json.get("topics", [{}])[0].get("name") for d in docs] == [ | |
| None, None, "Theme 2", "Theme 3", "Theme 4", "Theme 5", None] | |
| jobs = db_session.scalars(select(Job).where(Job.workspace_id == ws.id, Job.kind == "topics")).all() | |
| assert len(jobs) == 1 and jobs[0].params == {} | |
| focus(client, headers, **body) # the job is still queued: not again | |
| assert len(db_session.scalars(select(Job).where(Job.workspace_id == ws.id, Job.kind == "topics")).all()) == 1 | |
| def test_with_no_topics_to_show_the_ai_writes_them(client, owner, db_session, angles, monkeypatch): | |
| headers, ws = owner | |
| monkeypatch.setattr(settings, "llm_api_key", "") # nothing is tagged | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year. " * 5) | |
| cards = focus(client, headers, document_id=str(doc.id)) | |
| assert [c["title"] for c in cards] == ["AI angle 0", "AI angle 1", "AI angle 2"] and not angles["tagged"] | |
| assert angles["focus"][0]["candidate_tags"] == [] and angles["focus"][0]["avoid_titles"] == [] | |
| def cite(db, chat: Chat, *docs: Document) -> None: | |
| answer = db.scalars(select(Message).where(Message.chat_id == chat.id, Message.role == "assistant")).first() | |
| db.add_all([Citation(message_id=answer.id, document_id=d.id, marker=i + 1, path="sources/x/p.md", line_start=1, | |
| line_end=2) for i, d in enumerate(docs)]) | |
| db.commit() | |
| def test_a_chat_shows_the_topics_of_the_posts_it_cites(client, owner, db_session, angles): | |
| headers, ws = owner | |
| a = tagged_doc(db_session, ws, "a", ["Annual Raises"]) | |
| b = tagged_doc(db_session, ws, "b", ["Grandfathering"]) | |
| tagged_doc(db_session, ws, "c", ["Never Cited"]) | |
| merge(db_session, ws, "Pricing", "How to price.", (a, 0.9), (b, 0.9)) | |
| chat = add_chat(db_session, ws, [("user", "How should I price?"), | |
| ("assistant", "Once a year [1], keep early [2].")]) | |
| cite(db_session, chat, a, b) | |
| cards = focus(client, headers, chat_ids=[str(chat.id)]) | |
| assert cards[0] == {"title": "Pricing", "description": "How to price."} | |
| assert {c["title"] for c in cards[1:]} == {"Annual Raises", "Grandfathering"} and not angles["focus"] | |
| def test_a_chats_cited_posts_get_tagged_and_locked_ones_are_skipped(client, owner, db_session, angles): | |
| headers, ws = owner | |
| fresh = add_doc(db_session, ws, "https://ada.example.com/p/fresh", "New post. " * 20) | |
| fresh.title = "Fresh" | |
| locked = add_doc(db_session, ws, "https://ada.example.com/p/locked") | |
| locked.title, locked.path, locked.metadata_json = "Locked", None, {"locked": True} | |
| db_session.commit() | |
| angles["tag_answers"] = {"Fresh": ["Pricing", "Churn"]} | |
| chat = add_chat(db_session, ws, [("user", "Q?"), ("assistant", "An answer [1] [2].")]) | |
| cite(db_session, chat, fresh, locked) | |
| cards = focus(client, headers, chat_ids=[str(chat.id)]) | |
| assert angles["tagged"] == [["Fresh"]] and {c["title"] for c in cards} == {"Pricing", "Churn"} | |
| def test_a_chat_without_citations_gets_ai_topics(client, owner, db_session, angles): | |
| headers, ws = owner | |
| chat = add_chat(db_session, ws, [("user", "Hi there"), ("assistant", "Hello! Ask me about your posts.")]) | |
| cards = focus(client, headers, chat_ids=[str(chat.id)]) | |
| assert len(cards) == 3 and angles["focus"] and not angles["tagged"] | |
| assert angles["focus"][0]["items"] == ["Pricing chat\nHi there\nHello! Ask me about your posts."] | |
| def test_focus_goes_ahead_of_the_material_worded_for_the_style(client, b2v, owner, house, db_session, | |
| style, length, words): | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year, and say why. " * 3) | |
| make_video(client, headers, url=None, document_id=str(doc.id), video_style=style, focus="Annual raises", | |
| focus_detail="One raise a year, announced early. Why readers accept it.") | |
| sent = creates(b2v)[-1]["json"] | |
| assert "url" not in sent and sent["video_length"] == length # a focus needs the text, not the link | |
| head, material = sent["content"].split("\n\n---\n\n", 1) | |
| assert head.index('"Annual raises"') < head.index("What to cover: One raise a year, announced early. Why " | |
| "readers accept it.") < head.index(words) | |
| assert material.startswith("Raise prices") | |
| # The topic alone (no description) still works, without the "What to cover" part. | |
| make_video(client, headers, url=None, document_id=str(doc.id), video_style=style, focus="Annual raises") | |
| head = creates(b2v)[-1]["json"]["content"].split("\n\n---\n\n", 1)[0] | |
| assert '"Annual raises"' in head and "What to cover" not in head | |
| def test_a_custom_topic_goes_ahead_of_the_material(client, b2v, owner, house, db_session, style, depth): | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year, and say why. " * 3) | |
| make_video(client, headers, url=None, document_id=str(doc.id), video_style=style, | |
| focus_prompt=' Why the "yearly" raise\nbeats small ones ') | |
| sent = creates(b2v)[-1]["json"] | |
| assert "url" not in sent # the instruction needs the text, not the link | |
| head, material = sent["content"].split("\n\n---\n\n", 1) | |
| assert head.startswith("The user wants this video to focus on: \"Why the 'yearly' raise beats small ones\".") | |
| assert "only as far as the material below supports it" in head and "make a general video" in head | |
| assert head.endswith(depth) and material.startswith("Raise prices") | |
| def test_a_blank_custom_topic_is_no_focus(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year, and say why. " * 3) | |
| make_video(client, headers, url=None, document_id=str(doc.id), focus_prompt=" ") | |
| assert creates(b2v)[-1]["json"]["url"] == "https://ada.example.com/p/a" # the link, as with no focus | |
| def test_a_suggested_and_a_custom_topic_together_are_refused(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| doc = add_doc(db_session, ws, "https://ada.example.com/p/a", "Raise prices once a year, and say why. " * 3) | |
| r = client.post("/api/videos", json={"document_id": str(doc.id), "focus": "Annual raises", | |
| "focus_prompt": "My own angle"}, headers=headers) | |
| assert r.status_code == 422 and "not both" in r.text and not creates(b2v) | |
| def test_chat_with_a_focus_keeps_the_chat_context_first(client, b2v, owner, house, db_session): | |
| from app.routers.videos import CHAT_CONTEXT | |
| headers, ws = owner | |
| chat = add_chat(db_session, ws, [("user", "How often should I raise prices?"), | |
| ("assistant", "Once a year, and tell readers why it is happening.")]) | |
| make_video(client, headers, url=None, chat_ids=[str(chat.id)], video_style="custom:20", focus="Annual raises") | |
| content = creates(b2v)[-1]["json"]["content"] | |
| assert content.startswith(CHAT_CONTEXT) and content.index('"Annual raises"') < content.index("User:") | |
| # The standard options only, and per-workspace counters | |
| def test_options_that_are_not_offered_are_refused_on_every_plan(client, b2v, owner, body): | |
| headers, ws = owner | |
| r = client.post("/api/videos", json={"url": "https://a.test/x", **body}, headers=headers) | |
| assert r.status_code in (400, 404, 422) and not creates(b2v) | |
| def test_free_plan_can_make_a_plain_video(client, b2v, owner, free_plan): | |
| make_video(client, owner[0], custom_voice_id=SARAH) | |
| def test_ai_edits_counted_and_given_back_on_failure(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| r = client.post(f"/api/videos/{art['id']}/p/scenes/1/regenerate", json={"description": "x"}, headers=headers) | |
| assert r.status_code == 200 and video_limits.used(db_session, ws.id, "ai_edits") == 1 | |
| b2v.queue("POST", "/api/projects/812/scenes/1/regenerate", (409, {"detail": "busy"})) | |
| r = client.post(f"/api/videos/{art['id']}/p/scenes/1/regenerate", json={}, headers=headers) | |
| assert r.status_code == 409 and video_limits.used(db_session, ws.id, "ai_edits") == 1 | |
| def test_ai_edits_stop_at_the_workspace_limit(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| db_session.add(UsageCounter(workspace_id=ws.id, period=video_limits.period_key("ai_edits"), metric="ai_edits", | |
| used=1000)) | |
| db_session.commit() | |
| b2v.calls.clear() | |
| r = client.post(f"/api/videos/{art['id']}/p/scenes/1/regenerate", json={"description": "shorter"}, headers=headers) | |
| assert r.status_code == 402 and r.json()["detail"]["kind"] == "video_limits.ai_edits" and b2v.calls == [] | |
| def test_template_switch_uses_a_video_and_checks_the_template(client, b2v, owner, theirs, db_session): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| r = client.post(f"/api/videos/{art['id']}/p/change-template-regenerate-layouts", json={"template": "custom_77"}, | |
| headers=headers) | |
| assert r.status_code == 404 and used(db_session, ws) == 1 | |
| r = client.post(f"/api/videos/{art['id']}/p/change-template-regenerate-layouts", json={"template": "geometric"}, | |
| headers=headers) | |
| assert r.status_code == 200 and used(db_session, ws) == 2 | |
| # Edit jobs keep blog2video's own status on the artifact (not "generating"), so the list and editor can tell them apart | |
| def artifact_status(db, art) -> str: | |
| db.expire_all() | |
| return db.get(Artifact, uuid.UUID(art["id"])).status | |
| def test_an_edit_job_keeps_blog2videos_status(client, b2v, owner, db_session, raw): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| b2v.queue("GET", "/api/v1/videos/812/status", (200, {"status": raw, "ready": False})) | |
| assert client.get(f"/api/videos/{art['id']}/status", headers=headers).json()["artifact_status"] == raw | |
| assert artifact_status(db_session, art) == raw | |
| def test_starting_a_template_switch_marks_the_video(client, b2v, owner, db_session): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| r = client.post(f"/api/videos/{art['id']}/p/change-template-regenerate-layouts", json={"template": "geometric"}, | |
| headers=headers) | |
| assert r.status_code == 200 and artifact_status(db_session, art) == "regenerating" | |
| def test_the_list_rechecks_a_video_whose_job_ended_unwatched(client, b2v, owner, db_session): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| b2v.queue("GET", "/api/v1/videos/812/status", (200, {"status": "regenerating", "ready": False}), | |
| (200, {"status": "generated", "ready": True})) | |
| client.get(f"/api/videos/{art['id']}/status", headers=headers) | |
| assert artifact_status(db_session, art) == "regenerating" | |
| listed = client.get("/api/videos", headers=headers).json() | |
| assert listed[0]["status"] == "ready" and listed[0]["b2v_status"] == "generated" | |
| def test_editor_proxy_only_allows_listed_paths(client, b2v, owner): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| b2v.calls.clear() | |
| for method, path in [("GET", "members"), ("POST", "review"), ("POST", "launch-studio"), ("GET", "../../auth/me")]: | |
| assert client.request(method, f"/api/videos/{art['id']}/p/{path}", headers=headers).status_code == 404 | |
| assert b2v.calls == [] | |
| def test_editor_passes_downloads_through(client, b2v, owner): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| r = client.get(f"/api/videos/{art['id']}/p/download", headers=headers) | |
| assert r.status_code == 200 and r.content == b"ID3audio" and r.headers["content-type"] == "audio/mpeg" | |
| def test_removed_editor_features_are_not_proxied(client, b2v, owner): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| b2v.calls.clear() | |
| for method, path in [("POST", "chat"), ("GET", "chat/history"), ("GET", "download-studio"), | |
| ("POST", "avatar-portrait"), ("POST", "avatar-batch/authorize")]: | |
| assert client.request(method, f"/api/videos/{art['id']}/p/{path}", json={}, headers=headers).status_code == 404 | |
| assert b2v.calls == [] | |
| # Script review | |
| def test_script_review_flow(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| scenes = client.get(f"/api/videos/{art['id']}/script", headers=headers).json()["scenes"] | |
| assert [s["id"] for s in scenes] == [1, 2] | |
| draft = {"title": "A", "display_text": "Text", "narration_text": "", "draft_scenes": [], "revision": 1} | |
| r = client.post(f"/api/videos/{art['id']}/script/scenes/1/ai-preview", json={**draft, "instruction": "punchier"}, | |
| headers=headers) | |
| assert r.status_code in (404, 405) # rewriting a scene with AI is not offered | |
| r = client.post(f"/api/videos/{art['id']}/script/approve", | |
| json={"scenes": [{"id": 1, "title": "A", "narration_text": "Hi"}]}, headers=headers) | |
| assert r.status_code == 200 and "/api/projects/812/script-review/approve" in b2v.paths("POST") | |
| # Status, refunds, sweep | |
| def test_failure_refunds_once_from_poller_and_sweep(client, b2v, owner, db_session, session_factory): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| failed = (200, {"status": "failed", "step": 1, "running": False, "error": "unreachable url"}) | |
| b2v.queue("GET", "/api/v1/videos/812/status", failed, failed) | |
| assert client.get(f"/api/videos/{art['id']}/status", headers=headers).json()["artifact_status"] == "failed" | |
| assert used(db_session, ws) == 0 | |
| client.get(f"/api/videos/{art['id']}/status", headers=headers) | |
| age(db_session, db_session.scalar(select(B2VVideo))) | |
| video_quota.sweep_failed_videos(session_factory) | |
| assert used(db_session, ws) == 0 | |
| def test_sweep_refunds_videos_nobody_polled(client, b2v, owner, db_session, session_factory): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| age(db_session, db_session.scalar(select(B2VVideo))) | |
| b2v.queue("GET", "/api/v1/videos/812/status", (404, {"detail": "Not found"})) | |
| assert video_quota.sweep_failed_videos(session_factory) == 1 | |
| db_session.expire_all() | |
| assert used(db_session, ws) == 0 and db_session.get(Artifact, uuid.UUID(art["id"])).status == "failed" | |
| def test_a_finished_video_is_not_refunded_by_a_later_error(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| b2v.queue("GET", "/api/v1/videos/812/status", (200, {"status": "generated", "ready": True}), | |
| (200, {"status": "error", "error": "a later edit failed"})) | |
| client.get(f"/api/videos/{art['id']}/status", headers=headers) | |
| client.get(f"/api/videos/{art['id']}/status", headers=headers) | |
| assert used(db_session, ws) == 1 and db_session.scalar(select(B2VVideo)).quota_state == "kept" | |
| def test_legacy_video_shows_stored_links_and_refuses_edits(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| a = Artifact(workspace_id=ws.id, type="video", status="ready", | |
| content_json={"provider": "blog2video", "b2v_video_id": 99, "video_url": "https://cdn.test/old.mp4"}) | |
| db_session.add(a) | |
| db_session.commit() | |
| got = client.get(f"/api/videos/{a.id}", headers=headers).json() | |
| assert got["legacy"] is True and got["video_url"] == "https://cdn.test/old.mp4" | |
| assert client.patch(f"/api/videos/{a.id}/p/update-project", json={}, headers=headers).status_code == 410 | |
| assert b2v.calls == [] | |
| def test_get_video_mints_the_preview_link_once(client, b2v, owner): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| first = client.get(f"/api/videos/{art['id']}", headers=headers).json() | |
| client.get(f"/api/videos/{art['id']}", headers=headers) | |
| assert first["preview_url"] == f"{API}/embed/t" and first["project"]["id"] == 812 | |
| assert len([p for p in b2v.paths("POST") if p.startswith("/api/embed/token/")]) == 1 | |
| def test_scene_stock_clips_are_read_from_assigned_video(client, b2v, owner): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| clip = {"id": 7, "asset_type": "video", "filename": "scene_1_99.mp4", "r2_url": "https://r2.example/clip.mp4", | |
| "excluded": False} | |
| still = {"id": 8, "asset_type": "image", "filename": "img_a.webp", "r2_url": "https://r2.example/a.webp", | |
| "excluded": False} | |
| code = '{"layout": "news_headline", "layoutProps": {"assignedVideo": "scene_1_99.mp4"}}' | |
| b2v.queue("GET", "/api/projects/812", (200, {"id": 812, "status": "generated", "assets": [clip, still], | |
| "scenes": [{"id": 1, "order": 1, "title": "A", "remotion_code": code}]})) | |
| project = client.get(f"/api/videos/{art['id']}", headers=headers).json()["project"] | |
| assert project["scenes"][0]["images"] == [{"filename": "scene_1_99.mp4", "asset_id": 7, "kind": "video", | |
| "url": "https://r2.example/clip.mp4"}] | |
| assert project["summary"]["images"] == 1 and project["summary"]["clips"] == 1 | |
| def test_delete_removes_it_from_blog2video(client, b2v, owner): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| assert client.delete(f"/api/videos/{art['id']}", headers=headers).status_code == 200 | |
| assert "/api/projects/812" in b2v.paths("DELETE") | |
| def test_capacity_alert(b2v, caplog): | |
| with caplog.at_level(logging.ERROR): | |
| video_quota.check_capacity() | |
| assert "ALERT" in caplog.text and "videos 145/150" in caplog.text | |
| # Voices | |
| def test_new_workspace_gets_starter_voices_and_library(client, b2v, owner): | |
| got = client.get("/api/video-voices", headers=owner[0]).json() | |
| assert [v["voice_id"] for v in got["saved"]] == [SARAH] # free voices only | |
| assert [v["voice_id"] for v in got["library"]] == [SARAH] # the paid voice is not offered | |
| assert "custom" not in got and all("premium" not in v for v in got["saved"] + got["library"]) | |
| def test_a_paid_voice_cannot_be_saved_or_used_on_any_plan(client, b2v, owner): | |
| assert client.post("/api/video-voices/saved", json={"voice_id": BELLA}, headers=owner[0]).status_code == 404 | |
| r = client.post("/api/videos", json={"url": "https://a.test/x", "custom_voice_id": BELLA}, headers=owner[0]) | |
| assert r.status_code == 404 and not creates(b2v) | |
| def test_saved_voices_never_touch_blog2video_saved_list(client, b2v, owner, db_session): | |
| headers, _ = owner | |
| client.post("/api/video-voices/saved", json={"voice_id": BELLA}, headers=headers) | |
| client.delete(f"/api/video-voices/saved/{BELLA}", headers=headers) | |
| assert not [p for p in b2v.paths() if "/voices/saved" in p] | |
| def notestack_voice(db, ws, voice_id="nv_narrator", name="Narrator") -> None: | |
| """A voice the workspace designed on the Voice page (same ElevenLabs account as blog2video).""" | |
| db.add(VoiceProfile(workspace_id=ws.id, host_voices={"custom": [{"voice_id": voice_id, "name": name}]})) | |
| db.commit() | |
| def test_any_number_of_voices_can_be_saved(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| client.get("/api/video-voices", headers=headers) # the starter voice | |
| for i in range(7): | |
| db_session.add(UserSavedVoice(workspace_id=ws.id, voice_id=f"v{i}", name=f"V{i}")) | |
| db_session.commit() | |
| assert client.post("/api/video-voices/saved", json={"voice_id": SARAH}, headers=headers).status_code in (200, 201) | |
| got = client.get("/api/video-voices", headers=headers).json() | |
| assert len(got["saved"]) == 8 and "max_saved" not in got | |
| def test_notestack_voices_are_saved_for_audio_but_not_used_in_videos(client, b2v, owner, other, db_session): | |
| headers, ws = owner | |
| notestack_voice(db_session, ws) | |
| got = client.get("/api/video-voices", headers=headers).json() | |
| assert got["notestack"] == [{"voice_id": "nv_narrator", "name": "Narrator", "kind": "designed", "saved": False}] | |
| r = client.post("/api/video-voices/saved", json={"voice_id": "nv_narrator"}, headers=headers) | |
| assert r.status_code == 201 and r.json()["source"] == "notestack" | |
| saved = client.get("/api/video-voices", headers=headers).json()["saved"] | |
| assert [v["source"] for v in saved if v["voice_id"] == "nv_narrator"] == ["notestack"] # kept, for audio overviews | |
| r = client.post("/api/videos", json={"url": "https://a.test/x", "custom_voice_id": "nv_narrator"}, headers=headers) | |
| assert r.status_code == 404 and not creates(b2v) # a video uses the free built-in voices | |
| def test_revoking_the_clone_drops_it_from_video_voices(client, b2v, owner, db_session, monkeypatch): | |
| headers, ws = owner | |
| monkeypatch.setattr(settings, "elevenlabs_api_key", "") # no call out to delete the voice | |
| user_id = db_session.scalar(select(User.id)) | |
| db_session.add(VoiceConsent(workspace_id=ws.id, user_id=user_id, sample_key="ws/x/sample.mp3", | |
| consent_text="ok", elevenlabs_voice_id="nv_clone")) | |
| db_session.commit() | |
| assert client.post("/api/video-voices/saved", json={"voice_id": "nv_clone"}, headers=headers).status_code == 201 | |
| client.delete("/api/voice/consent", headers=headers) | |
| assert "nv_clone" not in [v["voice_id"] for v in client.get("/api/video-voices", headers=headers).json()["saved"]] | |
| def test_designing_cloning_and_sampling_voices_for_videos_are_gone(client, b2v, owner): | |
| headers, _ = owner | |
| for method, path in [("POST", "design/prompt"), ("POST", "design/preset"), ("POST", "custom"), ("POST", "clone"), | |
| ("POST", "sample"), ("GET", "custom/31/preview"), ("DELETE", "custom/31")]: | |
| assert client.request(method, f"/api/video-voices/{path}", json={}, headers=headers).status_code in (404, 405), path | |
| # Styles | |
| def test_styles_are_per_workspace(client, b2v, owner, other, db_session): | |
| headers, ws = owner | |
| made = client.post("/api/video-styles", json={"name": "Punchy", "guidance": "Punchy."}, headers=headers).json() | |
| assert made["id"] == "custom:41" and db_session.get(B2VStyle, 41).workspace_id == ws.id | |
| assert client.patch("/api/video-styles/41", json={"name": "x", "guidance": "y", "version": 0}, | |
| headers=other[0]).status_code == 404 | |
| assert client.delete("/api/video-styles/41", headers=other[0]).status_code == 404 | |
| # Templates | |
| def test_custom_templates_are_switched_off(client, b2v, owner): | |
| headers, _ = owner | |
| assert client.post("/api/video-templates/extract/prompt", json={"prompt": "x"}, headers=headers).status_code == 404 | |
| assert client.post("/api/video-templates", json={"name": "x", "theme": {}}, headers=headers).status_code == 404 | |
| assert client.get("/api/video-templates", headers=headers).status_code == 404 | |
| assert b2v.calls == [] | |
| def test_template_create_generate_and_ready(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| theme = client.post("/api/video-templates/extract/prompt", json={"prompt": "A bold dark brand with neon"}, | |
| headers=headers).json() | |
| made = client.post("/api/video-templates", json={"name": "Brand", "theme": theme["theme"]}, headers=headers) | |
| assert made.status_code == 201 and made.json()["ref"] == "custom_58" | |
| client.post("/api/video-templates/58/generate", headers=headers) | |
| b2v.queue("GET", "/api/custom-templates/58/generation-status", (200, {"status": "complete"})) | |
| assert client.get("/api/video-templates/58/generation-status", headers=headers).json()["ready"] is True | |
| assert video_limits.used(db_session, ws.id, "templates") == 1 | |
| assert video_limits.used(db_session, ws.id, "template_ai_daily") == 2 | |
| def test_free_plan_can_not_start_a_template(client, b2v, owner, free_plan): | |
| r = client.post("/api/video-templates/extract/url", json={"url": "https://brand.test"}, headers=owner[0]) | |
| assert r.status_code == 402 and b2v.calls == [] | |
| def test_template_delete_forces_only_over_own_videos(client, b2v, owner, other, db_session): | |
| headers, ws = owner | |
| db_session.add(B2VTemplate(b2v_template_id=58, workspace_id=ws.id, name="Mine", ready=True)) | |
| db_session.add(B2VVideo(workspace_id=other[1].id, b2v_video_id=5, template_ref="custom_58")) | |
| db_session.commit() | |
| b2v.queue("DELETE", "/api/custom-templates/58", (409, {"detail": {"error": "template_in_use"}})) | |
| assert client.delete("/api/video-templates/58", headers=headers).status_code == 409 | |
| db_session.query(B2VVideo).delete() | |
| db_session.commit() | |
| b2v.queue("DELETE", "/api/custom-templates/58", (409, {"detail": {"error": "template_in_use"}})) | |
| assert client.delete("/api/video-templates/58", headers=headers).status_code == 200 | |
| assert b2v.calls[-1]["params"] == {"force": "true"} | |
| def test_template_editor_is_per_workspace_and_allowlisted(client, b2v, owner, other, db_session): | |
| headers, ws = owner | |
| db_session.add(B2VTemplate(b2v_template_id=58, workspace_id=ws.id, name="Mine", ready=True)) | |
| db_session.commit() | |
| assert client.get("/api/video-templates/58/p/versions", headers=headers).status_code == 200 | |
| assert client.get("/api/video-templates/58/p/versions", headers=other[0]).status_code == 404 | |
| assert client.get("/api/video-templates/58/p/internal/ids", headers=headers).status_code == 404 | |
| r = client.post("/api/video-templates/58/p/scenes/intro/ai-edit", json={"prompt": "bigger"}, headers=headers) | |
| assert r.status_code == 200 and video_limits.used(db_session, ws.id, "ai_edits") == 1 | |
| # Quota bookkeeping | |
| def test_sync_follows_effective_plan(client, owner, db_session, monkeypatch): | |
| _, ws = owner | |
| sub = video_quota.sync_video_quota(db_session, ws) | |
| assert (sub.video_plan, sub.video_limit) == ("studio", 20) # billing disabled | |
| monkeypatch.setattr(settings, "billing_enabled", True) | |
| sub.plan, sub.status = "writer", "active" | |
| db_session.commit() | |
| assert video_quota.sync_video_quota(db_session, ws).video_limit == 10 | |
| sub.status = "canceled" | |
| db_session.commit() | |
| sub = video_quota.sync_video_quota(db_session, ws) | |
| assert (sub.video_plan, sub.video_limit) == ("free", 2) | |
| def test_monthly_fallback_reset(client, owner, db_session, session_factory): | |
| _, ws = owner | |
| sub = video_quota.sync_video_quota(db_session, ws) # studio (billing disabled): monthly | |
| now = datetime.now(UTC) | |
| # No Stripe renewal, period started 40 days ago: resets. | |
| sub.videos_used, sub.videos_period_start = 1, now - timedelta(days=40) | |
| db_session.commit() | |
| assert video_quota.reset_due_video_periods(session_factory) == 1 | |
| db_session.expire_all() | |
| assert sub.videos_used == 0 | |
| # Monthly Stripe subscription renewing soon: left to invoice.paid. | |
| sub.videos_used, sub.videos_period_start = 3, now - timedelta(days=40) | |
| sub.provider_subscription_id, sub.status = "sub_1", "active" | |
| sub.current_period_end = now - timedelta(days=10) | |
| db_session.commit() | |
| assert video_quota.reset_due_video_periods(session_factory) == 0 | |
| # Annual subscription (renews in 11 months): still resets monthly. | |
| sub.current_period_end = now + timedelta(days=330) | |
| db_session.commit() | |
| assert video_quota.reset_due_video_periods(session_factory) == 1 | |
| # The video's reference lives on its row | |
| def test_row_holds_the_full_reference(client, b2v, owner, db_session): | |
| headers, ws = owner | |
| art = make_video(client, headers) | |
| row = db_session.scalar(select(B2VVideo)) | |
| assert row.user_id == ws.owner_id and row.artifact_id == uuid.UUID(art["id"]) | |
| assert (row.title, row.source_url, row.aspect_ratio) == ("https://ada.example.com/p/a", "https://ada.example.com/p/a", | |
| "landscape") | |
| sent = creates(b2v)[0]["json"]["metadata"] | |
| assert sent["user_id"] == str(ws.owner_id) and sent["workspace_id"] == str(ws.id) | |
| assert sent["video_ref"] == str(row.id) | |
| b2v.queue("GET", "/api/v1/videos/812/status", (200, {"status": "done", "ready": True, | |
| "video_url": "https://cdn.test/v.mp4"})) | |
| client.get(f"/api/videos/{art['id']}/status", headers=headers) | |
| client.get(f"/api/videos/{art['id']}", headers=headers) | |
| db_session.expire_all() | |
| assert row.video_url == "https://cdn.test/v.mp4" and row.preview_url == f"{API}/embed/t" | |
| def test_calls_use_the_row_not_the_artifact_copy(client, b2v, owner, db_session): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| a = db_session.get(Artifact, uuid.UUID(art["id"])) | |
| a.content_json = {**a.content_json, "b2v_video_id": 999} # a stale or tampered copy | |
| db_session.commit() | |
| client.patch(f"/api/videos/{art['id']}/p/update-project", json={"accent_color": "#000000"}, headers=headers) | |
| assert "/api/projects/812/update-project" in b2v.paths("PATCH") | |
| def test_library_delete_removes_it_from_blog2video_and_keeps_the_record(client, b2v, owner, db_session): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| assert client.delete(f"/api/artifacts/{art['id']}", headers=headers).status_code == 200 | |
| assert "/api/projects/812" in b2v.paths("DELETE") | |
| db_session.expire_all() | |
| row = db_session.scalar(select(B2VVideo)) | |
| assert row.status == "deleted" and row.artifact_id is None | |
| # Video limits: Free 1 in total, Writer 10 a month, Studio 20 a month | |
| def _on_plan(db, ws, plan: str) -> None: | |
| sub = db.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)) | |
| sub.plan, sub.status = plan, "active" | |
| db.commit() | |
| def test_plan_video_limits(client, owner, db_session, monkeypatch): | |
| _, ws = owner | |
| monkeypatch.setattr(settings, "billing_enabled", True) | |
| video_quota.sync_video_quota(db_session, ws) | |
| for plan, limit in (("free", 2), ("writer", 10), ("studio", 20)): | |
| _on_plan(db_session, ws, plan) | |
| assert video_quota.sync_video_quota(db_session, ws).video_limit == limit | |
| def test_free_is_two_videos_in_total(client, b2v, owner, free_plan, db_session, session_factory): | |
| headers, ws = owner | |
| make_video(client, headers) | |
| make_video(client, headers, url="https://a.test/2") | |
| r = client.post("/api/videos", json={"url": "https://a.test/3"}, headers=headers) | |
| assert r.status_code == 402 | |
| usage = client.get("/api/videos/quota", headers=headers).json() | |
| assert usage["resets_at"] is None | |
| sub = db_session.scalar(select(Subscription).where(Subscription.workspace_id == ws.id)) | |
| sub.videos_period_start = datetime.now(UTC) - timedelta(days=90) | |
| db_session.commit() | |
| assert video_quota.reset_due_video_periods(session_factory) == 0 # never refills | |
| # Deleting the video does not give it back. | |
| art_id = db_session.scalar(select(B2VVideo)).artifact_id | |
| client.delete(f"/api/videos/{art_id}", headers=headers) | |
| assert client.post("/api/videos", json={"url": "https://a.test/4"}, headers=headers).status_code == 402 | |
| def test_free_video_that_failed_is_given_back(client, b2v, owner, free_plan, db_session): | |
| headers, ws = owner | |
| b2v.queue("POST", "/api/v1/videos", (422, {"detail": "bad url"})) | |
| assert client.post("/api/videos", json={"url": "https://a.test/1"}, headers=headers).status_code == 400 | |
| make_video(client, headers) | |
| def test_dropping_to_free_counts_what_was_made(client, b2v, owner, db_session, monkeypatch): | |
| headers, ws = owner | |
| monkeypatch.setattr(settings, "billing_enabled", True) | |
| video_quota.sync_video_quota(db_session, ws) | |
| _on_plan(db_session, ws, "writer") | |
| for i in range(3): | |
| make_video(client, headers, url=f"https://a.test/{i}") | |
| _on_plan(db_session, ws, "free") | |
| usage = client.get("/api/videos/quota", headers=headers).json() | |
| assert (usage["used"], usage["limit"]) == (3, 2) | |
| assert client.post("/api/videos", json={"url": "https://a.test/x"}, headers=headers).status_code == 402 | |
| def test_list_restores_a_video_whose_library_item_went_missing(client, b2v, owner, db_session): | |
| headers, _ = owner | |
| art = make_video(client, headers) | |
| row = db_session.scalar(select(B2VVideo)) | |
| row.artifact_id, row.status = None, "generated" | |
| db_session.delete(db_session.get(Artifact, uuid.UUID(art["id"]))) | |
| db_session.commit() | |
| listed = client.get("/api/videos", headers=headers).json() | |
| assert len(listed) == 1 and listed[0]["status"] == "ready" | |
| b2v.calls.clear() | |
| client.get(f"/api/videos/{listed[0]['id']}/status", headers=headers) | |
| assert "/api/v1/videos/812/status" in b2v.paths("GET") # the same blog2video video | |
| assert len(client.get("/api/videos", headers=headers).json()) == 1 # restored once | |