PythagorasGoal/tests/test_book_cadenza.py

"""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 gli
scatti del disaster-SL rotolante dipendono dalla cadenza — 1h: BTC 0 / ETH 1 in 7-8 anni; 4h:
BTC 2 (contro 1 del pavimento); 24h: ETH 5 con 2 in 30 giorni. Chi avesse "corretto" il CRON
verso il docstring avrebbe spostato il libro dalla riga 1h a quella 4h.

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
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 venue_probe as VP            # noqa: E402  lo slot di release, dichiarato li'
from src.strategies.skyhook import LTF_MIN         # noqa: E402  la griglia di SKH01, NON ridichiarata

BOOK_EXECUTE = ROOT / "scripts" / "live" / "book_execute.py"
CRON_BOOK = ROOT / "scripts" / "cron_book.sh"
RIGA_4H_MIN = 240    # la riga in cui il rotolante di BTC passa da 0 a 2 scatti (r0823_sl_anchor.py);
                     # la PEGGIORE misurata e' 24h (ETH 5, 2 in 30 giorni), non questa

_RX_CRON5 = re.compile(r"`([\d*/,-]+(?: [\d*/,-]+){4})`")


def cadenza_minuti(expr: str) -> int:
    """Minuti fra due esecuzioni di un'espressione crontab a 5 campi, SOLO per cadenze regolari
    (intervalli tutti uguali): `47 * * * *` -> 60, `*/30 * * * *` -> 30, `7 */4 * * *` -> 240.
    `*/45` (intervalli 45/15), `0 */5` (ultimo intervallo 4h) e `*/0` vengono RIFIUTATI."""
    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("*/"):
        passo = int(m[2:])
        if passo <= 0 or 60 % passo:
            raise ValueError(f"passo di minuti non regolare: {expr}")
        return passo
    if h.startswith("*/") and m.isdigit():
        passo = int(h[2:])
        if passo <= 0 or 24 % passo:
            raise ValueError(f"passo di ore non regolare: {expr}")
        return 60 * passo
    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:
    """La prima espressione crontab a 5 campi che `cron_book.sh` dichiara fra backtick — di
    qualunque forma: e' `cadenza_minuti` a giudicarla, non questa regex (P4: un'espressione
    presente ma non oraria deve fallire come 'non oraria', non come 'assente')."""
    m = _RX_CRON5.search(CRON_BOOK.read_text())
    assert m, "cron_book.sh non dichiara piu' la sua riga di crontab fra backtick"
    return m.group(1)


def _crontab() -> str | None:
    """Testo della crontab dell'utente, None se non leggibile (P5: distinto da 'vuota')."""
    if not shutil.which("crontab"):
        return None
    try:
        out = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=10)
    except Exception:
        return None
    return out.stdout if out.returncode == 0 else None


def _righe_attive(testo: str, marcatore: str) -> list[str]:
    """Le espressioni (5 campi) di TUTTE le righe attive che citano `marcatore`: due righe sono
    due esecuzioni, e la seconda non deve nascondersi dietro la prima."""
    return [" ".join(r.split()[:5]) for r in testo.splitlines()
            if marcatore in r and not r.lstrip().startswith("#")]


# ---------------------------------------------------------------- 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_sotto_la_griglia_skh_e_sotto_la_riga_4h():
    c = cadenza_minuti(_cron_dichiarato())
    assert c == 60
    assert c < LTF_MIN, "la cadenza deve stare SOTTO la griglia di SKH01 (latenza <= 1 barra)"
    assert c < RIGA_4H_MIN, "la riga 4h e' quella in cui il rotolante di BTC fa 2 scatti"
    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


def test_cron_book_evita_lo_slot_di_release_deribit():
    # vincolo 2 del commento di cron_book.sh — il motivo vero del :07 -> :47 (4 giri morti in
    # 63 giorni): il giro del martedi' alle 09:MM deve cadere FUORI dallo slot dichiarato in
    # venue_probe (derivato, non riscritto). Il :07 lo violava; se qualcuno tornasse a :15
    # con docstring e crontab coerenti, gli altri test passerebbero e questo no.
    minuto = int(_cron_dichiarato().split()[0])
    base = datetime(2026, 9, 1, tzinfo=timezone.utc)                 # un lunedi'
    martedi = base + timedelta(days=(VP.RELEASE_WEEKDAY - base.weekday()) % 7)
    giro = martedi.replace(hour=VP.RELEASE_START_MIN // 60, minute=minuto)
    assert not VP.in_release_window(giro), f"il giro :{minuto:02d} cade nello slot di release"
    assert VP.in_release_window(giro.replace(minute=7)), "controllo positivo: il :07 ci cadeva"


@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


@pytest.mark.parametrize("expr", ["47 9 * * 2", "*/45 * * * *", "0 */5 * * *", "*/0 * * * *",
                                  "47 * 1 * *"])
def test_il_parser_rifiuta_cio_che_non_e_regolare(expr):
    with pytest.raises((ValueError, AssertionError)):
        cadenza_minuti(expr)


# ---------------------------------------------------------------- la crontab che GIRA

def test_la_crontab_installata_e_quella_dichiarata_e_una_sola():
    """P16: il cron gira dalla macchina, non dal repo. TRE stati, non due (P5):
    - crontab NON leggibile -> SALTATO (non verificabile qui);
    - leggibile, e questo progetto non ci compare affatto -> SALTATO (macchina di sviluppo);
    - leggibile con altre righe del progetto ma SENZA cron_book.sh attivo -> ROSSO: il libro
      non e' schedulato dove il progetto e' installato (una riga commentata per manutenzione e
      dimenticata e' esattamente questo caso).
    E la riga deve essere UNA: due righe attive sono due esecuzioni l'ora."""
    testo = _crontab()
    if testo is None:
        pytest.skip("crontab non leggibile: non verificabile qui")
    if str(ROOT) not in testo and ROOT.name not in testo:
        pytest.skip("la crontab non cita questo progetto: non e' la macchina che lo esegue")
    righe = _righe_attive(testo, "cron_book.sh")
    assert righe, "questa macchina esegue il progetto ma cron_book.sh NON e' schedulato"
    assert len(righe) == 1, f"cron_book.sh schedulato {len(righe)} volte: {righe}"
    assert righe[0] == _cron_dichiarato()
    assert cadenza_minuti(righe[0]) == 60


def test_due_righe_attive_non_si_nascondono_dietro_la_prima():
    finta = ("47 * * * * /x/PythagorasGoal/scripts/cron_book.sh\n"
             "# 7 * * * * /x/PythagorasGoal/scripts/cron_book.sh\n"
             "17 * * * * /x/PythagorasGoal/scripts/cron_book.sh\n")
    assert _righe_attive(finta, "cron_book.sh") == ["47 * * * *", "17 * * * *"]
    assert _righe_attive("# 47 * * * * /x/cron_book.sh\n", "cron_book.sh") == []