researchowl/tests/test_short_producer.py

"""El pipeline del Short: orden de los pasos y fallbacks.

Todo con dobles: ni Claude ni shortsmith ni SQLite. Lo que se comprueba aquí es
que el fundamento se mira ANTES de renderizar y que ningún camino de fallo se
come el spec.
"""
import json

import pytest

from src.config import settings
from src.generator.short import ShortProducer, ShortResult, ShortsDisabled
from src.generator.shortsmith import JobResult, ShortsmithError, ShortsmithUnavailable
from tests.test_spec_contract import TEMPLATES

SPEC = {
    "version": 1,
    "meta": {"id": "jal1628", "title": "JAL 1628"},
    "shots": [
        {"template": "radar_sweep", "duration": 15.0, "props": {"headline": "3 RADARS"}},
        {"template": "scale_bars", "duration": 15.0, "props": {
            "headline": "REPORTED SCALE",
            "bars": [{"label": "BOEING 747", "value": 232, "unit": "FT"}]}},
    ],
}

CHUNKS = [{
    "content": "Three radars tracked the object. A Boeing 747 is 232 ft long.",
    "url": "https://faa.example/jal1628",
    "title": "FAA file",
    "source_type": "web",
}]


class FakeDB:
    def __init__(self, article_url=None):
        self.article_url = article_url
        self.saved: list[tuple] = []

    async def get_session(self, session_id):
        return {"id": session_id, "topic": "JAL 1628 Alaska 1986"}

    async def get_article_url(self, session_id):
        return self.article_url

    async def save_output(self, session_id, output_type, content):
        self.saved.append((session_id, output_type, content))
        return len(self.saved)

    async def log_api_call(self, *a, **kw):
        return None


class FakeProcessor:
    def __init__(self, chunks=None):
        self.chunks = CHUNKS if chunks is None else chunks

    async def rag_chunks(self, session_id, query, top_k=20):
        return self.chunks


PRESETS = {
    "sonar": "low drone — the case-file mood",
    "pulse": "sub-bass heartbeat — debunks",
    "static": "shortwave static — document drops",
    "none": "digital silence",
}


class FakeClient:
    """shortsmith de mentira. `fail_at` decide dónde se rompe."""

    def __init__(self, fail_at=None, job_status="done", warnings=None):
        self.fail_at = fail_at
        self.job_status = job_status
        self.warnings = warnings or []
        self.rendered = None

    async def templates(self, refresh=False):
        if self.fail_at == "templates":
            raise ShortsmithUnavailable("no hay nadie al otro lado")
        return TEMPLATES

    async def audio_presets(self, refresh=False):
        if self.fail_at == "audio":
            raise ShortsmithUnavailable("sin /audio")
        return dict(PRESETS)

    async def render(self, spec):
        if self.fail_at == "render":
            raise ShortsmithUnavailable("conexión rechazada")
        self.rendered = spec
        return "job-1"

    async def poll(self, job_id, on_progress=None, **kw):
        if self.fail_at == "poll":
            raise ShortsmithError("job atascado")
        if on_progress:
            await on_progress(0.5, "running")
        return JobResult(job_id, self.job_status, 1.0, self.warnings,
                         "OOMKilled" if self.job_status == "error" else None)

    async def fetch_video(self, job_id):
        if self.fail_at == "fetch":
            raise ShortsmithError("404 del vídeo")
        return b"\x00\x00\x00 ftypisom" + b"\x00" * 2048


def llm_returning(*responses):
    queue = list(responses)

    async def call(system, prompt):
        call.prompts.append(prompt)
        return queue.pop(0) if len(queue) > 1 else queue[0]

    call.prompts = []
    return call


def producer(tmp_path, monkeypatch, *, client=None, llm=None, db=None, processor=None):
    monkeypatch.setattr(settings, "shorts_dir", str(tmp_path / "shorts"))
    monkeypatch.setattr(settings, "shortsmith_enabled", True)
    return ShortProducer(
        db or FakeDB(),
        processor or FakeProcessor(),
        client=client or FakeClient(),
        llm_call=llm or llm_returning(json.dumps(SPEC)),
    )


# --- camino feliz -----------------------------------------------------------

@pytest.mark.asyncio
async def test_happy_path_writes_the_mp4_to_disk(tmp_path, monkeypatch):
    db = FakeDB(article_url="https://www.theexclusionzone.com/jal-1628/")
    p = producer(tmp_path, monkeypatch, db=db)

    result = await p.produce(153)

    assert result.has_video
    assert result.video_path.endswith("153.mp4")
    assert open(result.video_path, "rb").read()[:12].endswith(b"ftypisom")
    assert result.title == "JAL 1628"
    assert result.duration_s == 30.0
    assert result.article_url.endswith("/jal-1628/")
    assert result.failure is None


@pytest.mark.asyncio
async def test_the_spec_is_saved_before_the_render(tmp_path, monkeypatch):
    """Si el render se cae, la parte cara ya está en la DB y /short_spec la
    devuelve."""
    db = FakeDB()
    p = producer(tmp_path, monkeypatch, db=db, client=FakeClient(fail_at="render"))

    result = await p.produce(153)

    assert db.saved and db.saved[0][1] == "short_en"
    assert json.loads(db.saved[0][2])["meta"]["id"] == "jal1628"
    assert not result.has_video


@pytest.mark.asyncio
async def test_grounding_runs_before_rendering(tmp_path, monkeypatch):
    """El informe existe aunque el render no llegue a empezar: ese es el orden
    del §12 y es lo que hace que la revisión humana llegue igual."""
    p = producer(tmp_path, monkeypatch, client=FakeClient(fail_at="render"))

    result = await p.produce(153)

    assert result.grounding is not None
    assert result.grounding.total > 0


@pytest.mark.asyncio
async def test_ungrounded_claims_do_not_block_the_render(tmp_path, monkeypatch):
    """Un dato sin encontrar puede ser una fabricación o un artefacto de
    formato. Lo decide una persona: el vídeo se entrega con el aviso al lado."""
    invented = json.loads(json.dumps(SPEC))
    invented["shots"][0]["props"]["headline"] = "41,000 FT"
    p = producer(tmp_path, monkeypatch, llm=llm_returning(json.dumps(invented)))

    result = await p.produce(153)

    assert result.has_video
    assert [c.text for c in result.grounding.ungrounded] == ["41,000 FT"]


@pytest.mark.asyncio
async def test_the_article_url_reaches_the_prompt(tmp_path, monkeypatch):
    llm = llm_returning(json.dumps(SPEC))
    p = producer(tmp_path, monkeypatch,
                 db=FakeDB(article_url="https://www.theexclusionzone.com/jal-1628/"),
                 llm=llm)

    await p.produce(153)

    assert "https://www.theexclusionzone.com/jal-1628/" in llm.prompts[0]


# --- fallbacks --------------------------------------------------------------

@pytest.mark.asyncio
@pytest.mark.parametrize("fail_at", ["render", "poll", "fetch"])
async def test_every_render_failure_still_returns_the_spec(tmp_path, monkeypatch, fail_at):
    p = producer(tmp_path, monkeypatch, client=FakeClient(fail_at=fail_at))

    result = await p.produce(153)

    assert not result.has_video
    assert result.failure
    assert json.loads(result.spec_json)["meta"]["id"] == "jal1628"


@pytest.mark.asyncio
async def test_a_job_that_errors_is_reported_with_its_reason(tmp_path, monkeypatch):
    p = producer(tmp_path, monkeypatch, client=FakeClient(job_status="error"))

    result = await p.produce(153)

    assert not result.has_video
    assert "OOMKilled" in result.failure


@pytest.mark.asyncio
async def test_an_unwritable_spec_still_returns_the_last_attempt(tmp_path, monkeypatch):
    """Tres intentos fallidos no son motivo para tirar la generación."""
    broken = json.loads(json.dumps(SPEC))
    broken["meta"]["id"] = "MAYÚSCULAS Y ESPACIOS"
    p = producer(tmp_path, monkeypatch, llm=llm_returning(json.dumps(broken)))

    result = await p.produce(153)

    assert not result.has_video
    assert result.attempts == 3
    assert result.spec["meta"]["id"] == "MAYÚSCULAS Y ESPACIOS"
    assert "no pasó la validación" in result.failure
    assert result.grounding is None      # no hay spec válido que comprobar


@pytest.mark.asyncio
async def test_a_response_that_is_not_json_at_all_comes_back_raw(tmp_path, monkeypatch):
    p = producer(tmp_path, monkeypatch,
                 llm=llm_returning("Lo siento, no puedo ayudarte con eso."))

    result = await p.produce(153)

    assert result.spec is None
    assert "Lo siento" in result.raw_response


@pytest.mark.asyncio
async def test_render_warnings_travel_with_the_result(tmp_path, monkeypatch):
    warnings = [{"template": "data_card", "text": "UNA FILA MUY LARGA",
                 "requested": 44, "size": 38}]
    p = producer(tmp_path, monkeypatch, client=FakeClient(warnings=warnings))

    result = await p.produce(153)

    assert result.has_video
    assert result.render_warnings[0]["template"] == "data_card"


# --- interruptores y precondiciones -----------------------------------------

@pytest.mark.asyncio
async def test_the_kill_switch_says_so_instead_of_crashing(tmp_path, monkeypatch):
    p = producer(tmp_path, monkeypatch)
    monkeypatch.setattr(settings, "shortsmith_enabled", False)

    with pytest.raises(ShortsDisabled):
        await p.produce(153)


@pytest.mark.asyncio
async def test_an_unreachable_renderer_is_named_clearly(tmp_path, monkeypatch):
    """Sin contrato no hay prompt que escribir: aquí no hay fallback posible y
    el mensaje lo dice."""
    p = producer(tmp_path, monkeypatch, client=FakeClient(fail_at="templates"))

    with pytest.raises(ShortsmithUnavailable):
        await p.produce(153)


@pytest.mark.asyncio
async def test_a_session_without_chunks_says_what_to_run(tmp_path, monkeypatch):
    p = producer(tmp_path, monkeypatch, processor=FakeProcessor(chunks=[]))

    with pytest.raises(ValueError, match="/process"):
        await p.produce(153)


def test_domain_is_drawn_without_protocol_or_www(monkeypatch):
    monkeypatch.setattr(settings, "ghost_url_en", "https://www.theexclusionzone.com")
    assert ShortProducer(FakeDB(), FakeProcessor(), client=FakeClient())._domain() \
        == "THEEXCLUSIONZONE.COM"


def test_short_result_without_a_spec_has_an_empty_json():
    assert ShortResult(topic="x").spec_json == ""


# --- limpieza ---------------------------------------------------------------

@pytest.mark.asyncio
async def test_purging_a_session_takes_its_video_with_it(tmp_path, monkeypatch):
    """Un MP4 huérfano en el PVC es negligible contra 5 Gi y es arqueología
    dentro de un año."""
    import time

    import aiosqlite

    from src.db import database
    from src.db.database import ResearchDB

    shorts = tmp_path / "shorts"
    shorts.mkdir()
    (shorts / "1.mp4").write_bytes(b"viejo")
    (shorts / "2.mp4").write_bytes(b"reciente")
    monkeypatch.setattr(settings, "shorts_dir", str(shorts))

    conn = await aiosqlite.connect(tmp_path / "t.db")
    conn.row_factory = aiosqlite.Row
    await conn.executescript(database.SCHEMA)
    old, now = time.time() - 90 * 86400, time.time()
    await conn.execute(
        "INSERT INTO research_sessions (id, topic, status, telegram_chat_id,"
        " created_at, updated_at) VALUES (1,'viejo','saturated',1,?,?)", (old, old))
    await conn.execute(
        "INSERT INTO research_sessions (id, topic, status, telegram_chat_id,"
        " created_at, updated_at) VALUES (2,'nuevo','saturated',1,?,?)", (now, now))
    await conn.commit()

    counts = await ResearchDB(conn).purge_old_sessions(30)
    await conn.close()

    assert counts["shorts"] == 1
    assert not (shorts / "1.mp4").exists()
    assert (shorts / "2.mp4").exists(), "la sesión reciente conserva su vídeo"


@pytest.mark.asyncio
async def test_the_youtube_url_never_passes_for_an_article_url(tmp_path):
    """Subir un Short escribe su URL de YouTube en `published_url`. Si
    `get_article_url` no filtrara las filas short_en, el siguiente Short de esa
    sesión enlazaría al Short anterior: un bucle silencioso, porque la URL es
    válida y nadie la mira dos veces."""
    import time

    import aiosqlite

    from src.db import database
    from src.db.database import OutputType, ResearchDB

    conn = await aiosqlite.connect(tmp_path / "urls.db")
    conn.row_factory = aiosqlite.Row
    await conn.executescript(database.SCHEMA)
    now = time.time()
    await conn.execute(
        "INSERT INTO research_sessions (id, topic, status, telegram_chat_id,"
        " created_at, updated_at) VALUES (1,'x','saturated',1,?,?)", (now, now))
    await conn.execute(
        "INSERT INTO outputs (session_id, output_type, content, created_at,"
        " published_url) VALUES (1,?,'...',?,?)",
        (OutputType.BLOG.value, now, "https://theexclusionzone.com/x/"))
    # El Short, subido DESPUÉS: es la fila más reciente con URL.
    await conn.execute(
        "INSERT INTO outputs (session_id, output_type, content, created_at,"
        " published_url) VALUES (1,?,'{}',?,?)",
        (OutputType.SHORT_EN.value, now + 60, "https://youtube.com/shorts/abc"))
    await conn.commit()

    url = await ResearchDB(conn).get_article_url(1)
    await conn.close()

    assert url == "https://theexclusionzone.com/x/"


# --- re-render de un spec editado (la vuelta de /short_spec) ----------------

async def _llm_prohibido(system, prompt):
    raise AssertionError("el re-render no debe llamar al LLM: este camino es gratis")


@pytest.mark.asyncio
async def test_rerender_writes_the_mp4_without_touching_the_llm(tmp_path, monkeypatch):
    from src.db.database import OutputType

    db = FakeDB(article_url="https://www.theexclusionzone.com/jal-1628/")
    p = producer(tmp_path, monkeypatch, db=db, llm=_llm_prohibido)

    result = await p.rerender(153, json.loads(json.dumps(SPEC)))

    assert result.has_video
    assert result.cost_usd == 0.0
    assert result.article_url == "https://www.theexclusionzone.com/jal-1628/"
    # El spec editado queda guardado ANTES del render: /upload_short saca los
    # metadatos del último spec y tienen que describir este vídeo.
    assert db.saved and db.saved[-1][1] == OutputType.SHORT_EN


@pytest.mark.asyncio
async def test_rerender_rechecks_the_grounding_of_the_edited_strings(tmp_path, monkeypatch):
    """La edición a mano puede meter una cifra nueva: el informe se rehace."""
    edited = json.loads(json.dumps(SPEC))
    edited["shots"][0]["props"]["headline"] = "41,000 FT"
    p = producer(tmp_path, monkeypatch, llm=_llm_prohibido)

    result = await p.rerender(153, edited)

    assert result.has_video
    assert [c.text for c in result.grounding.ungrounded] == ["41,000 FT"]


@pytest.mark.asyncio
async def test_an_edited_spec_that_breaks_the_contract_never_renders(tmp_path, monkeypatch):
    """Los errores vuelven con su ruta verbatim, igual que al modelo."""
    broken = json.loads(json.dumps(SPEC))
    broken["shots"][0]["template"] = "no_existe"
    client = FakeClient()
    p = producer(tmp_path, monkeypatch, client=client, llm=_llm_prohibido)

    result = await p.rerender(153, broken)

    assert not result.has_video
    assert "shots.0.template" in result.failure
    assert client.rendered is None
    # El spec editado se conserva para poder corregirlo y reenviarlo.
    assert json.loads(result.spec_json)["shots"][0]["template"] == "no_existe"


@pytest.mark.asyncio
async def test_rerender_on_a_purged_session_says_it_could_not_check(tmp_path, monkeypatch):
    """Sin chunks no hay contra qué mirar: se renderiza igual, avisando."""
    p = producer(tmp_path, monkeypatch, processor=FakeProcessor(chunks=[]),
                 llm=_llm_prohibido)

    result = await p.rerender(153, json.loads(json.dumps(SPEC)))

    assert result.has_video
    assert result.grounding is None
    assert any("NO se ha comprobado" in n for n in result.notes)


@pytest.mark.asyncio
async def test_rerender_failures_keep_the_spec_like_produce_does(tmp_path, monkeypatch):
    p = producer(tmp_path, monkeypatch, client=FakeClient(fail_at="render"),
                 llm=_llm_prohibido)

    result = await p.rerender(153, json.loads(json.dumps(SPEC)))

    assert not result.has_video
    assert "shortsmith no responde" in result.failure
    assert json.loads(result.spec_json)["meta"]["id"] == "jal1628"


@pytest.mark.asyncio
async def test_rerender_respects_the_kill_switch(tmp_path, monkeypatch):
    from src.generator.short import ShortsDisabled

    p = producer(tmp_path, monkeypatch, llm=_llm_prohibido)
    monkeypatch.setattr(settings, "shortsmith_enabled", False)

    with pytest.raises(ShortsDisabled):
        await p.rerender(153, json.loads(json.dumps(SPEC)))


# --- la paleta de audio (GET /audio) ----------------------------------------


@pytest.mark.asyncio
async def test_an_edited_preset_from_the_live_palette_renders(tmp_path, monkeypatch):
    """El retoque para el que existe el bucle de edición: cambiar la banda
    sonora a "pulse" sin pagar otra generación."""
    edited = json.loads(json.dumps(SPEC))
    edited["audio"] = {"preset": "pulse"}
    client = FakeClient()
    p = producer(tmp_path, monkeypatch, client=client, llm=_llm_prohibido)

    result = await p.rerender(153, edited)

    assert result.has_video
    assert client.rendered["audio"]["preset"] == "pulse"


@pytest.mark.asyncio
async def test_a_preset_the_renderer_does_not_know_is_rejected_with_the_palette(
        tmp_path, monkeypatch):
    edited = json.loads(json.dumps(SPEC))
    edited["audio"] = {"preset": "vaporwave"}
    client = FakeClient()
    p = producer(tmp_path, monkeypatch, client=client, llm=_llm_prohibido)

    result = await p.rerender(153, edited)

    assert not result.has_video
    assert "audio.preset" in result.failure and "pulse" in result.failure
    assert client.rendered is None


@pytest.mark.asyncio
async def test_a_dead_audio_endpoint_never_blocks_a_sonar_render(tmp_path, monkeypatch):
    """La paleta mejora el prompt, no lo define: sin /audio se cae a la base y
    un spec con sonar renderiza igual."""
    p = producer(tmp_path, monkeypatch, client=FakeClient(fail_at="audio"),
                 llm=_llm_prohibido)

    result = await p.rerender(153, json.loads(json.dumps(SPEC)))

    assert result.has_video