"""Escritura del shot spec: prompt, bucle de reintento y fallback. El LLM entra como un callable, así que aquí se prueba el bucle, no a Haiku: respuesta buena, respuesta malformada, typo en una prop, y las tres seguidas. """ import json from pathlib import Path import pytest from src.generator.shortspec import ( MAX_ATTEMPTS, ShortSpecWriter, SpecWriteFailed, extract_json, ) from tests.test_spec_contract import TEMPLATES EXAMPLE = Path(__file__).resolve().parents[1] / "src/generator/examples/jal1628.json" GOOD = { "version": 1, "meta": {"id": "caso", "title": "Un caso"}, "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"}]}}, ], } class FakeLLM: """Devuelve respuestas de una cola y guarda los prompts que recibió.""" def __init__(self, *responses): self.responses = list(responses) self.prompts: list[str] = [] self.systems: list[str] = [] async def __call__(self, system, prompt): self.systems.append(system) self.prompts.append(prompt) return self.responses.pop(0) if len(self.responses) > 1 else self.responses[0] def writer(*responses, **kw): llm = FakeLLM(*responses) return ShortSpecWriter(llm, TEMPLATES, **kw), llm # --- parseo ----------------------------------------------------------------- def test_extract_json_survives_markdown_fences(): assert extract_json('```json\n{"a": 1}\n```') == {"a": 1} assert extract_json('Here you go:\n{"a": 1}\nHope that helps') == {"a": 1} assert extract_json('{"a": 1}') == {"a": 1} def test_extract_json_complains_when_there_is_no_object(): with pytest.raises(ValueError): extract_json("I'm afraid I can't do that") def test_straight_quotes_inside_a_string_are_repaired(): """Cómo falla esto en la vida real (sesión de Bélgica, 2026-08-01): el modelo escribe la cita con comillas rectas, que cierran la cadena JSON antes de tiempo. Se arregla aquí porque además es lo que se quiere dibujar.""" broken = '{"quote_a": ""CREDIBLE PEOPLE. THEY TOLD WHAT THEY SAW."", "n": 1}' assert extract_json(broken) == { "quote_a": "“CREDIBLE PEOPLE. THEY TOLD WHAT THEY SAW.”", "n": 1} def test_the_repair_leaves_correct_json_alone(): good = {"a": 'texto con “tipográficas” dentro', "b": [1, 2], "c": {"d": "e"}} assert extract_json(json.dumps(good, ensure_ascii=False)) == good def test_the_repair_does_not_eat_escaped_quotes(): assert extract_json(r'{"a": "dijo \"hola\" y se fue"}') == {"a": 'dijo "hola" y se fue'} def test_an_unrepairable_response_reports_the_offending_fragment(): """El modelo no ve su salida numerada: "line 189 column 22" no le sirve; el trozo sí.""" with pytest.raises(ValueError) as exc: extract_json('{"a": 1, "b": [1, 2,,,], "c": 3}') assert "aquí:" in str(exc.value) # --- el prompt -------------------------------------------------------------- def test_prompt_carries_the_fetched_contract_not_a_copy(): w, _ = writer("{}") prompt = w.build_prompt("Caso X", "material", None, "THEEXCLUSIONZONE.COM") assert "radar_sweep:" in prompt and "contact_bearing_deg" in prompt assert "1-3 elementos" in prompt # los límites de longitud, del esquema assert "ink, amber, amber_dark" in prompt # la paleta, también del esquema def test_prompt_states_the_editorial_constraints(): w, _ = writer("{}") prompt = w.build_prompt("Caso X", "material", "https://x.test/post", "X.TEST") assert "20-45 seconds" in prompt assert "never hex" in prompt assert "https://x.test/post" in prompt assert "X.TEST" in prompt assert "material" in prompt def test_prompt_includes_the_worked_example_in_full(): w, _ = writer("{}") prompt = w.build_prompt("Caso X", "material", None, "X.TEST") example = json.loads(EXAMPLE.read_text()) assert example["meta"]["title"] in prompt assert "counter_close" in prompt def test_prompt_says_out_loud_that_there_is_no_article_yet(): w, _ = writer("{}") assert "No article URL yet" in w.build_prompt("X", "m", None, "X.TEST") # --- bucle ------------------------------------------------------------------ @pytest.mark.asyncio async def test_a_good_response_validates_on_the_first_attempt(): w, llm = writer(json.dumps(GOOD)) result = await w.write("Caso X", "material") assert result.attempts == 1 assert result.spec["meta"]["id"] == "caso" assert len(llm.prompts) == 1 @pytest.mark.asyncio async def test_malformed_json_triggers_a_retry(): w, llm = writer("no soy JSON", json.dumps(GOOD)) result = await w.write("Caso X", "material") assert result.attempts == 2 assert "no es un objeto JSON válido" in result.history[0][0] assert "previous attempt was rejected" in llm.prompts[1] @pytest.mark.asyncio async def test_the_exact_error_paths_are_fed_back_verbatim(): """La ruta que devuelve la validación es lo más útil que se le puede dar al modelo: se le pasa tal cual, sin parafrasear.""" bad = json.loads(json.dumps(GOOD)) bad["shots"][0]["props"]["sweeeps"] = 2 w, llm = writer(json.dumps(bad), json.dumps(GOOD)) result = await w.write("Caso X", "material") assert result.attempts == 2 assert "shots.0.radar_sweep.props.sweeeps" in llm.prompts[1] assert "sweeps" in llm.prompts[1] # y cuáles sí valen @pytest.mark.asyncio async def test_three_failures_raise_but_keep_the_last_attempt(): """La parte cara es la generación, no el render: el último intento viaja en la excepción para poder editarlo a mano y reenviarlo.""" bad = json.loads(json.dumps(GOOD)) bad["meta"]["id"] = "Caso Con Espacios" w, _ = writer(json.dumps(bad)) with pytest.raises(SpecWriteFailed) as exc: await w.write("Caso X", "material") assert exc.value.attempts == MAX_ATTEMPTS assert exc.value.last_spec["meta"]["id"] == "Caso Con Espacios" assert any("meta.id" in e for e in exc.value.errors) @pytest.mark.asyncio async def test_an_off_target_duration_is_commented_once_then_accepted(): """70 s cumple el contrato pero no el encargo: se comenta y, si el modelo insiste, se renderiza igual antes que tirar la generación.""" long_spec = json.loads(json.dumps(GOOD)) long_spec["shots"][0]["duration"] = 55.0 # 70 s en total w, llm = writer(json.dumps(long_spec)) result = await w.write("Caso X", "material") assert result.attempts == MAX_ATTEMPTS assert result.notes and "recorta" in result.notes[0] assert "off-brief" in llm.prompts[1] @pytest.mark.asyncio async def test_a_valid_attempt_is_not_thrown_away_by_a_worse_one(): long_spec = json.loads(json.dumps(GOOD)) long_spec["shots"][0]["duration"] = 55.0 w, _ = writer(json.dumps(long_spec), "esto ya no es JSON", "tampoco") result = await w.write("Caso X", "material") assert result.spec["shots"][0]["duration"] == 55.0 assert result.notes @pytest.mark.asyncio async def test_the_contract_is_refetched_after_a_validation_failure(): """Si el renderizador se actualizó a mitad de la run, la plantilla nueva entra en el segundo intento.""" new_template = {"type": "object", "additionalProperties": False, "required": ["title"], "properties": {"title": {"type": "string", "minLength": 1}}} refreshed = {**TEMPLATES, "holo_scan": new_template} async def refresh(): return refreshed with_new = json.loads(json.dumps(GOOD)) with_new["shots"][1] = {"template": "holo_scan", "duration": 15.0, "props": {"title": "X"}} w, llm = writer(json.dumps(with_new), json.dumps(with_new), refresh_templates=refresh) result = await w.write("Caso X", "material") assert result.attempts == 2 assert "holo_scan" in llm.prompts[1] assert result.spec["shots"][1]["template"] == "holo_scan" @pytest.mark.asyncio async def test_progress_is_reported_only_when_it_retries(): seen = [] async def on_progress(text): seen.append(text) w, _ = writer("no JSON", json.dumps(GOOD)) await w.write("Caso X", "material", on_progress=on_progress) assert len(seen) == 1 and "attempt 2/3" in seen[0]