Blame for backend/tests/test_videos.py

"""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()


@pytest.fixture()
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']}"}


@pytest.fixture()
def owner(client, db_session):
    headers = register(client, "ada@example.com")
    return headers, db_session.scalar(select(Workspace))


@pytest.fixture()
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


@pytest.fixture()
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


@pytest.mark.parametrize("status", [401, 402, 403])
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


@pytest.mark.parametrize("method,path", [
    ("POST", "/api/projects"), ("POST", "/api/projects/bulk"), ("GET", "/api/projects"),
    ("GET", "/api/projects/template-availability"), ("GET", "/api/voices/saved"), ("POST", "/api/voices/saved"),
    ("DELETE", "/api/voices/saved/abc"), ("PATCH", "/api/video-styles/builtin/explainer"),
    ("PUT", "/api/video-styles/pin"), ("PUT", "/api/video-styles/selection"), ("PATCH", "/api/video-styles/your-style"),
    ("DELETE", "/api/auth/me/script-preferences"),
])
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


@pytest.fixture()
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()


@pytest.mark.parametrize("field,value", [("template", "custom_77"), ("video_style", "custom:9"),
                                         ("custom_voice_id", "vc_theirs"), ("video_style", "your_style"),
                                         ("template", "custom_12345"), ("template", "not-a-template"),
                                         ("template", "crafted_other")])
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"]


@pytest.fixture()
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)


@pytest.mark.parametrize("style,length", [("custom:20", "short"), ("custom:21", "medium")])
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)


@pytest.fixture()
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


@pytest.fixture()
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."]


@pytest.mark.parametrize("style,length,words", [("custom:20", "short", "overview of that topic"),
                                                ("custom:21", "medium", "deep dive into that topic")])
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


@pytest.mark.parametrize("style,depth", [("custom:20", "Make it a brief overview."),
                                         ("custom:21", "Make it a deep dive.")])
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


@pytest.mark.parametrize("body", [{"video_length": "detailed"}, {"video_length": "more_detailed"},
                                  {"custom_voice_id": BELLA}, {"template": "crafted_neon"}])
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


@pytest.mark.parametrize("raw", ["regenerating", "voice_regenerating", "language_regenerating", "script_regenerating"])
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 == []


@pytest.mark.skip(reason="custom templates are switched off (video_templates is not mounted)")
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


@pytest.mark.skip(reason="custom templates are switched off (video_templates is not mounted)")
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 == []


@pytest.mark.skip(reason="custom templates are switched off (video_templates is not mounted)")
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"}


@pytest.mark.skip(reason="custom templates are switched off (video_templates is not mounted)")
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