"""La cadenza del libro live e' ORARIA, e le tre dichiarazioni devono restare d'accordo. Debito §5.7 (chiuso il 2026-09-02): il docstring di `book_execute.py` prescriveva «ogni ~230 minuti» mentre il cron gira ogni ora. Non era cosmetico: `r0823_sl_anchor.py` misura che sulla riga 4h BTC RADDOPPIA gli scatti del disaster-SL rotolante (2 contro 1; a 24h ETH arriva a 5). Chi avesse "corretto" il cron verso il docstring avrebbe spostato il libro sulla riga peggiore. P1: il bersaglio si DERIVA da cio' che gira. Qui le fonti sono tre — il docstring (cio' che un umano legge), `cron_book.sh` (cio' che dichiara di essere installato) e la crontab (cio' che gira davvero) — e il test le confronta fra loro invece di ridichiarare «60 minuti». """ from __future__ import annotations import ast import re import shutil import subprocess from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] BOOK_EXECUTE = ROOT / "scripts" / "live" / "book_execute.py" CRON_BOOK = ROOT / "scripts" / "cron_book.sh" GRIGLIA_SKH_MIN = 230 # data1 di SKH01 (src/strategies/skyhook.py): la riga da NON usare RIGA_PEGGIORE_MIN = 240 # 4h: dove BTC raddoppia gli scatti (r0823_sl_anchor.py) def cadenza_minuti(expr: str) -> int: """Minuti fra due esecuzioni di un'espressione crontab a 5 campi, per i soli casi regolari.""" campi = expr.split() assert len(campi) == 5, expr m, h, dom, mon, dow = campi assert dom == mon == dow == "*", f"cadenza non regolare: {expr}" if h == "*" and m.isdigit(): return 60 if h == "*" and m.startswith("*/"): return int(m[2:]) if h.startswith("*/") and m.isdigit(): return 60 * int(h[2:]) raise ValueError(f"espressione non riconosciuta: {expr}") def _docstring() -> str: return ast.get_docstring(ast.parse(BOOK_EXECUTE.read_text())) or "" def _cron_dichiarato() -> str: """L'espressione crontab che `cron_book.sh` dichiara nel suo intestazione, fra backtick.""" m = re.search(r"`(\d+ \* \* \* \*)`", CRON_BOOK.read_text()) assert m, "cron_book.sh non dichiara piu' la sua riga di crontab" return m.group(1) def _cron_installato() -> str | None: if not shutil.which("crontab"): return None try: out = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=10) except Exception: return None if out.returncode != 0: return None for riga in out.stdout.splitlines(): if "cron_book.sh" in riga and not riga.lstrip().startswith("#"): return " ".join(riga.split()[:5]) return None # ---------------------------------------------------------------- regressione sul docstring # La guardia non cerca due letterali («ogni ~230»): estrae OGNI prescrizione di cadenza scritta # come «ogni/every N unita'» e pretende che valga 60 minuti. ⚠️ Limite dichiarato (P13): una # guardia sui numeri non copre il ragionamento — una prosa che prescrive una cadenza senza la # parola «ogni» passa. Il controllo positivo qui sotto dice cosa la guardia VEDE. _NUMERI = {"un": 1, "una": 1, "due": 2, "tre": 3, "quattro": 4, "sei": 6, "otto": 8, "dodici": 12} _RX_CADENZA = re.compile( r"\b(?:ogni|every)\s*~?\s*(\d+|un'?|una|due|tre|quattro|sei|otto|dodici)?\s*'?\s*" r"(minuti|minuto|minutes?|min|m|ore|ora|hours?|h)\b", re.I) def prescrizioni_minuti(testo: str) -> list[int]: """Ogni «ogni N unita'» del testo, in minuti. «ogni ora» -> 60, «ogni ~230 minuti» -> 230.""" out = [] for m in _RX_CADENZA.finditer(testo): n_raw, unita = m.group(1), m.group(2).lower() n = 1 if n_raw is None else (int(n_raw) if n_raw.isdigit() else _NUMERI[n_raw.rstrip("'")]) out.append(n * (60 if unita.startswith(("or", "h")) else 1)) return out def test_la_guardia_vede_le_prescrizioni_che_deve_vedere(): # controllo positivo (M15): i modi in cui il docstring potrebbe ri-prescrivere la riga 4h assert prescrizioni_minuti("va lanciato ogni ~230 minuti con la feed") == [230] assert prescrizioni_minuti("il giro va lanciato ogni quattro ore") == [240] assert prescrizioni_minuti("every 4 hours, then every 230 min") == [240, 230] assert prescrizioni_minuti("controllato ogni ora; ogni 60 minuti") == [60, 60] assert prescrizioni_minuti("fino a 230m di latenza") == [] # non e' una prescrizione def test_il_docstring_prescrive_solo_la_cadenza_oraria(): d = _docstring() p = prescrizioni_minuti(d) assert p, "il docstring deve contenere almeno una prescrizione di cadenza" assert all(x == 60 for x in p), f"prescrizioni non orarie nel docstring: {p}" assert re.search(r"CADENZA:\s*ORARIA", d), "il docstring deve dichiarare la cadenza ORARIA" def test_il_docstring_dichiara_la_stessa_riga_del_cron(): # la riga di crontab citata nel docstring e' quella di cron_book.sh, non un'altra d = _docstring() assert _cron_dichiarato() in d # ---------------------------------------------------------------- cron_book.sh def test_cron_book_e_orario_e_sta_lontano_dalla_riga_peggiore(): expr = _cron_dichiarato() c = cadenza_minuti(expr) assert c == 60 assert c < GRIGLIA_SKH_MIN < RIGA_PEGGIORE_MIN assert "ORARIA" in CRON_BOOK.read_text() def test_cron_book_evita_il_minuto_tondo(): # vincolo 1 del commento di cron_book.sh: fuori dai ~26s del :00 (rate-limit per-IP) minuto = int(_cron_dichiarato().split()[0]) assert minuto != 0 @pytest.mark.parametrize("expr,atteso", [("47 * * * *", 60), ("*/30 * * * *", 30), ("7 */4 * * *", 240), ("0 */24 * * *", 1440)]) def test_il_parser_della_cadenza(expr, atteso): assert cadenza_minuti(expr) == atteso def test_il_parser_rifiuta_cio_che_non_sa_leggere(): with pytest.raises((ValueError, AssertionError)): cadenza_minuti("47 9 * * 2") # ---------------------------------------------------------------- la crontab che GIRA def test_la_crontab_installata_e_quella_dichiarata(): """P16: il cron gira dalla macchina, non dal repo. Dove la crontab e' leggibile (la VPS), deve coincidere con cio' che cron_book.sh dichiara; altrove il test si dichiara saltato, non verde (P5: 'non vedo' non e' 'va tutto bene').""" vivo = _cron_installato() if vivo is None: pytest.skip("crontab non leggibile o senza cron_book.sh: non verificabile qui") assert vivo == _cron_dichiarato() assert cadenza_minuti(vivo) == 60