Files
researchowl/tests/test_shortsmith_live.py
ChemaVXandClaude Opus 5 20c8d03aa7
Build & Deploy ResearchOwl / build-and-push (push) Successful in 9s
feat(short): generación y render de Shorts vía shortsmith
Añade /generate short_en y /short_spec. El pipeline genera un shot spec
con Haiku, verifica cada cifra, fecha y cita contra los chunks de la
sesión, lo renderiza en shortsmith y entrega el MP4 por Telegram junto
a un informe de claims.

- ShortsmithClient con sondeo y fallback al spec JSON si el render falla
- Contrato de plantillas obtenido de GET /templates, no codificado
- Comprobación de fundamento determinista, sin LLM
- outputs.published_url para enlazar el artículo de Ghost
- Normalización de comillas rectas a tipográficas (ver KNOWN-ISSUES.md)

Lo que no aparece en los chunks se contrasta contra el ejemplo del
prompt: si casa ahí es fuga, no invención, y se informa como tal. El
purgado de sesiones se lleva también su MP4.

La subida a YouTube queda fuera a propósito: fase 3.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-01 21:55:42 +00:00

69 lines
2.4 KiB
Python

"""Prueba de fontanería contra un shortsmith VIVO.
Se salta salvo que se le dé una URL alcanzable desde donde corren los tests:
SHORTSMITH_LIVE_URL=http://10.43.86.57:8080 pytest tests/test_shortsmith_live.py -v
(dentro del cluster es `http://shortsmith-svc.shortsmith.svc.cluster.local:8080`;
desde el nodo, la ClusterIP de `kubectl get svc -n shortsmith`).
Renderiza el ejemplo de referencia entero — ~30 s de CPU en el pod — y comprueba
que vuelve un MP4. Es el paso 1 del §12 del spec de fase 2: probar el transporte
antes de generar nada.
"""
import json
import os
from pathlib import Path
import pytest
from src.generator.shortsmith import ShortsmithClient
LIVE_URL = os.environ.get("SHORTSMITH_LIVE_URL")
pytestmark = pytest.mark.skipif(
not LIVE_URL, reason="define SHORTSMITH_LIVE_URL para probar contra el servicio vivo")
EXAMPLE = Path(__file__).resolve().parents[1] / "src/generator/examples/jal1628.json"
@pytest.mark.asyncio
async def test_healthz_and_templates():
client = ShortsmithClient(LIVE_URL)
health = await client.health()
assert health["status"] == "ok"
assert health["templates"] >= 1
templates = await client.templates(refresh=True)
# No se comprueban nombres concretos a propósito: el contrato es de
# shortsmith y añadir plantillas allí no debe romper aquí.
assert templates, "GET /templates devolvió vacío"
for name, schema in templates.items():
assert schema.get("type") == "object", f"{name} no publica un esquema de objeto"
assert "properties" in schema
@pytest.mark.asyncio
async def test_render_the_reference_example_end_to_end(tmp_path):
client = ShortsmithClient(LIVE_URL)
spec = json.loads(EXAMPLE.read_text())
job_id = await client.render(spec)
seen = []
async def on_progress(fraction, status):
seen.append(fraction)
result = await client.poll(job_id, on_progress=on_progress)
assert result.ok, f"el job terminó en {result.status}: {result.error}"
assert seen and max(seen) == 1.0
video = await client.fetch_video(job_id)
# ftyp en los primeros bytes: es un MP4 de verdad, no una página de error.
assert b"ftyp" in video[:32]
assert len(video) > 100_000, f"solo {len(video)} bytes — sospechosamente corto"
out = tmp_path / "jal1628.mp4"
out.write_bytes(video)
print(f"\nrenderizado {len(video)/1e6:.2f} MB en {out}")