"""Revisione settimanale (src/live/revisione.py): sola lettura, firmata, e il silenzio non e' un esito. Nessun test chiama il modello o Telegram: `interroga` e `invia` sono iniettati. Nessun test scrive nel repo: tutto in una radice temporanea con file sintetici. """ from __future__ import annotations import re import subprocess import sys from datetime import datetime, timedelta, timezone from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.live import revisione as R # noqa: E402 from src.live.venue_probe import in_release_window # noqa: E402 ADESSO = datetime(2026, 9, 14, 6, 15, tzinfo=timezone.utc) # un lunedi' def _radice(tmp_path: Path) -> Path: (tmp_path / "CLAUDE.md").write_text("# CLAUDE\n\n## 4. Gate\n| STATARB | 2026-09-27 | soglie | ritiro |\n") (tmp_path / "config").mkdir(); (tmp_path / "config" / "live.json").write_text('{"disaster_sl_pct": 0.30}') j = tmp_path / "docs" / "journal"; j.mkdir(parents=True) for i in range(1, 10): g = (ADESSO - timedelta(days=i)).date().isoformat() (j / f"{g}.md").write_text(f"# Giornale {g}\nEquity $4,477.46 · leva 0.26x\n") d = tmp_path / "docs" / "diary"; d.mkdir() (d / "2026-09-09b-recente.md").write_text("recente: f 0,712\n") (d / "2026-08-20-vecchio.md").write_text("vecchio\n") (d / "senza-data.md").write_text("senza data\n") live = tmp_path / "data" / "live"; live.mkdir(parents=True) (live / "scale_watch_state.json").write_text('{"scala": 1.0}') (live / "usde_watch.jsonl").write_text("\n".join(f'{{"i": {i}}}' for i in range(10)) + "\n") return tmp_path # ---------------------------------------------------------------- materiale def test_il_materiale_prende_la_finestra_dichiarata_e_dice_cosa_non_legge(tmp_path): voci = R.materiale(_radice(tmp_path), ADESSO) nomi = [n for n, _, _ in voci] giornali = [n for n in nomi if n.startswith("giornale ")] assert len(giornali) == R.FINESTRA_GIORNALE_G, "7 pagine, non 9: la finestra e' dichiarata" assert "giornale 2026-09-13" in nomi and "giornale 2026-09-07" in nomi and "giornale 2026-09-06" not in nomi assert "diario 2026-09-09b-recente.md" in nomi assert "diario 2026-08-20-vecchio.md" not in nomi and "diario senza-data.md" not in nomi testo = dict((n, t) for n, t, _ in voci) assert "0,712" in testo["diario 2026-09-09b-recente.md"] assert testo["stato usde_watch.jsonl (ultime 3)"].splitlines() == ['{"i": 7}', '{"i": 8}', '{"i": 9}'] assert testo["stato scale_history.jsonl (ultime 3)"] == "(assente)", "un file assente si DICE (P5)" assert testo["git log dal 2026-08-31 + stato del working tree"].startswith("non leggibile"), \ "in una cartella che non e' un repo il log non si inventa" assert any(n.startswith("monitor_health") for n in nomi) def test_i_sorveglianti_vengono_prima_delle_voci_grandi(tmp_path): """Il tetto totale taglia in coda: gli stati (piccoli) devono stare PRIMA di giornale e diari, altrimenti sono i primi a sparire — e' successo alla prima stesura (usde_convert a 2 caratteri).""" nomi = [n for n, _, _ in R.materiale(_radice(tmp_path), ADESSO)] i_stato = max(i for i, n in enumerate(nomi) if n.startswith("stato ")) i_giornale = min(i for i, n in enumerate(nomi) if n.startswith("giornale ")) assert i_stato < i_giornale def test_il_troncamento_e_dichiarato(tmp_path, monkeypatch): r = _radice(tmp_path) (r / "CLAUDE.md").write_text("x" * 1000) monkeypatch.setattr(R, "MAX_CHARS_VOCE", 100) n, t, tr = R.materiale(r, ADESSO)[0] assert tr and "[… troncato a 100 caratteri su 1,000]" in t assert "[TRONCATO]" in R.costruisci_prompt([(n, t, tr)], ADESSO) def test_il_prompt_porta_le_regole_e_la_data(tmp_path): p = R.costruisci_prompt(R.materiale(_radice(tmp_path), ADESSO), ADESSO) assert "DATA DELLA REVISIONE: 2026-09-14 06:15Z" in p for t in R.TITOLI: assert t in p for parola in ("SOLA LETTURA", "CITA LA FONTE", "DECISIONI VINCOLANTI", "docs/memory", "I GATE HANNO UNA DATA", "NIENTE PREVISIONI"): assert parola in p, parola # ---------------------------------------------------------------- guardia def _revisione(sintesi="Tutto letto. Equity $4,477.46.", extra=""): return ("## Sintesi\n" + sintesi + "\n## Cosa non torna\nniente\n## Cosa e' maturo da decidere\n" "STATARB il 2026-09-27\n## Proposte (nessuna azione eseguita)\nnessuna\n" "## Cosa ho letto e cosa mi mancava\nCLAUDE.md" + extra) def test_la_guardia_rifiuta_il_vuoto_e_i_titoli_mancanti(): assert R.valida("", "fonte")[0] == "rifiutata" stato, motivi = R.valida("## Sintesi\nsolo questa", "fonte") assert stato == "rifiutata" and "titoli mancanti" in motivi[0] def test_la_guardia_sui_numeri_ha_il_controllo_positivo(): fonte = "Equity $4,477.46 · leva 0.26x · 2026-09-27" assert R.valida(_revisione(), fonte) == ("ok", []) stato, motivi = R.valida(_revisione(sintesi="Equity 9.999,99 e leva 3,14x"), fonte) assert stato == "sospetta" and "non presenti nel materiale" in motivi[0] # ---------------------------------------------------------------- rapporto e Telegram def test_il_rapporto_e_firmato_e_dichiara_la_sola_lettura(): voci = [("CLAUDE.md", "x" * 10, False), ("giornale", "y", True)] r = R.rapporto(_revisione(), None, "ok", [], "claude-fable-5-1", ADESSO, voci) assert r.startswith("# Revisione settimanale — 2026-09-14") for frase in ("`claude-fable-5-1`", "SOLA LETTURA", "lettore fallibile (P13)", "## Firma e materiale", "| CLAUDE.md | 10 | |", "| giornale | 1 | si |", "l'unico file"): assert frase in r, frase def test_un_modello_muto_produce_un_rapporto_che_lo_dice(): r = R.rapporto(None, "timeout dopo 900s", "rifiutata", ["timeout dopo 900s"], "m", ADESSO, []) assert "## Revisione NON eseguita" in r and "timeout dopo 900s" in r and "P5" in r tg = R.per_telegram(None, "timeout dopo 900s", "rifiutata", ADESSO, Path("docs/revisioni/x.md")) assert tg.startswith("🚨") and "timeout" in tg def test_telegram_porta_la_sintesi_e_dichiara_il_taglio(): tg = R.per_telegram(_revisione(sintesi="A corta."), None, "ok", ADESSO, Path("p.md")) assert "A <b>corta</b>." in tg and "niente" not in tg, "solo la Sintesi, con l'HTML neutralizzato" lunga = R.per_telegram(_revisione(sintesi="z" * 5000), None, "ok", ADESSO, Path("p.md")) assert "[… sintesi tagliata" in lunga and len(lunga) < 4096 # ---------------------------------------------------------------- orchestrazione: cosa scrive, cosa no def _albero(root: Path) -> set[str]: return {str(p.relative_to(root)) for p in root.rglob("*") if p.is_file()} def test_secco_non_chiama_e_non_scrive(tmp_path, capsys): r = _radice(tmp_path); prima = _albero(r) def mai(*a, **k): raise AssertionError("il modello NON va chiamato in --secco") esito = R.scrivi(r, ADESSO, secco=True, interroga=mai, invia=mai) assert esito["secco"] and _albero(r) == prima assert "SECCO" in capsys.readouterr().out def test_un_giro_riuscito_scrive_UN_solo_file_e_manda_la_sintesi(tmp_path): r = _radice(tmp_path); prima = _albero(r); mandati = [] esito = R.scrivi(r, ADESSO, modello="claude-fable-5-1", quiet=True, interroga=lambda p, m, t: (_revisione(), None), invia=lambda h: mandati.append(h) or True) nuovi = _albero(r) - prima assert nuovi == {"docs/revisioni/2026-09-14.md"}, "l'unico file scritto e' il rapporto" assert esito["stato"] == "ok" and esito["telegram"] is True assert "## Sintesi" in (r / "docs" / "revisioni" / "2026-09-14.md").read_text() assert len(mandati) == 1 and "Revisione settimanale" in mandati[0] def test_un_errore_del_modello_scrive_il_rapporto_e_allerta(tmp_path): r = _radice(tmp_path); mandati = [] esito = R.scrivi(r, ADESSO, quiet=True, interroga=lambda p, m, t: ("", "CLI `claude` non trovata"), invia=lambda h: mandati.append(h) or True) assert esito["stato"] == "rifiutata" and "non trovata" in esito["errore"] assert "NON eseguita" in (r / "docs" / "revisioni" / "2026-09-14.md").read_text() assert mandati and mandati[0].startswith("🚨") def test_una_risposta_che_non_e_una_revisione_e_un_errore_dichiarato(tmp_path): r = _radice(tmp_path) esito = R.scrivi(r, ADESSO, quiet=True, telegram=False, interroga=lambda p, m, t: ("ciao", None)) assert esito["stato"] == "rifiutata" and "titoli mancanti" in esito["errore"] assert "NON eseguita" in (r / "docs" / "revisioni" / "2026-09-14.md").read_text() def test_no_telegram_non_manda(tmp_path): r = _radice(tmp_path) def mai(h): raise AssertionError("Telegram NON va chiamato con telegram=False") esito = R.scrivi(r, ADESSO, quiet=True, telegram=False, interroga=lambda p, m, t: (_revisione(), None), invia=mai) assert "telegram" not in esito def test_il_prompt_va_su_stdin_non_su_argv(monkeypatch): """Al primo giro reale (09/09) 195k caratteri su argv hanno dato `Argument list too long` PRIMA di chiamare il modello. DEVE FALLIRE SE: qualcuno rimette il prompt fra gli argomenti.""" visto = {} class _P: returncode, stdout, stderr = 0, _revisione(), "" def finto_run(args, **kw): visto["args"], visto["input"] = args, kw.get("input") return _P() monkeypatch.setattr(R.subprocess, "run", finto_run) prompt = "PROMPT " * 40_000 testo, errore = R.interroga(prompt, "claude-fable-5-1", 5) assert errore is None and "## Sintesi" in testo assert visto["input"] == prompt and all(len(a) < 200 for a in visto["args"]) assert "--model" in visto["args"] and "claude-fable-5-1" in visto["args"] def test_una_eccezione_nella_chiamata_diventa_un_errore_non_un_crash(tmp_path, monkeypatch): """Controllo positivo sul difetto del primo giro: `OSError` dalla CLI → rapporto NON eseguita + 🚨.""" def esplode(args, **kw): raise OSError(7, "Argument list too long") monkeypatch.setattr(R.subprocess, "run", esplode) assert R.interroga("x", "m", 5) == ("", "OSError: [Errno 7] Argument list too long") r = _radice(tmp_path); mandati = [] def esplode_in_scrivi(p, m, t): raise RuntimeError("boom") esito = R.scrivi(r, ADESSO, quiet=True, interroga=esplode_in_scrivi, invia=lambda h: mandati.append(h) or True) assert esito["stato"] == "rifiutata" and "RuntimeError: boom" in esito["errore"] assert "NON eseguita" in (r / "docs" / "revisioni" / "2026-09-14.md").read_text() assert mandati and mandati[0].startswith("🚨") def test_il_revisore_non_e_il_modello_che_scrive_il_codice(): from src.live import analista as A assert R.MODELLO == "claude-fable-5-1" and R.MODELLO != A.MODELLO_DEFAULT # ---------------------------------------------------------------- cadenza: tre dichiarazioni d'accordo def _riga_crontab() -> str | None: try: out = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=10) except (FileNotFoundError, subprocess.TimeoutExpired): return None if out.returncode != 0: return None righe = [r for r in out.stdout.splitlines() if "cron_review.sh" in r and not r.lstrip().startswith("#")] return righe[0] if righe else "" def test_la_cadenza_dichiarata_nel_cron_sh_e_quella_installata_e_fuori_dagli_slot(): sh = (ROOT / "scripts" / "cron_review.sh").read_text() m = re.search(r"lunedi' (\d\d):(\d\d) UTC", sh) assert m, "cron_review.sh deve dichiarare «lunedi' HH:MM UTC»" ora, minuto = int(m.group(1)), int(m.group(2)) assert minuto != 0, "mai al minuto tondo (rate-limit per-IP)" prossimo = ADESSO.replace(hour=ora, minute=minuto) # ADESSO e' un lunedi' assert prossimo.weekday() == 0 and not in_release_window(prossimo) assert "revisione.py --quiet" in sh riga = _riga_crontab() if riga is None: pytest.skip("crontab non leggibile: la dichiarazione installata non e' verificabile qui") assert riga, "crontab leggibile ma cron_review.sh NON installato" campi = riga.split() assert campi[:5] == [str(minuto), str(ora), "*", "*", "1"], f"installata {campi[:5]}, dichiarata {minuto} {ora} * * 1"