Adriano
1a7f670966
src/live/giri.registra_giro: una riga jsonl {ts ms, campi} dopo una misura completata,
mai un'eccezione. edge_watch, fee_watch, scale_watch, vrp_f_watch, venue_news (cron_daily)
e venue_watch (cron_book, orario) scrivono data/<dir>/runs.jsonl; niente riga senza misura.
monitor_health: sei MonitorSpec (30h i giornalieri, 3h venue_watch), open_labeled=False.
Stessa riparazione di cc01 (#10, #12), generalizzata. Primo giro notturno: NUOVO, non allerta.
Test: incrocio P1 percorso/RUNS, misura -> una riga, non-misura -> nessuna. Suite: 1098 passed.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
"""registra_giro (src/live/giri.py, #16): il registro che i sorveglianti scrivono a ogni giro.
|
|
|
|
Il contratto e' quello di cc01_regime_watch, generalizzato: appende `{"ts": <ms>, **campi}` come
|
|
riga jsonl e non solleva MAI — un registro che fallisce non deve buttare via una misura gia' fatta.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from src.live.giri import registra_giro # noqa: E402
|
|
|
|
|
|
def test_scrive_una_riga_con_ts_in_millisecondi(tmp_path):
|
|
p = tmp_path / "sub" / "runs.jsonl"
|
|
prima = time.time() * 1000
|
|
assert registra_giro(p, stato="OK", n=3) is True
|
|
riga = json.loads(p.read_text().strip())
|
|
assert abs(riga["ts"] - prima) < 5_000
|
|
assert isinstance(riga["ts"], int)
|
|
assert riga["stato"] == "OK" and riga["n"] == 3
|
|
|
|
|
|
def test_crea_la_cartella_se_manca(tmp_path):
|
|
p = tmp_path / "non-esiste-ancora" / "runs.jsonl"
|
|
assert not p.parent.exists()
|
|
registra_giro(p, x=1)
|
|
assert p.exists()
|
|
|
|
|
|
def test_appende_senza_troncare(tmp_path):
|
|
p = tmp_path / "runs.jsonl"
|
|
registra_giro(p, giro=1)
|
|
registra_giro(p, giro=2)
|
|
righe = [json.loads(r) for r in p.read_text().splitlines()]
|
|
assert [r["giro"] for r in righe] == [1, 2]
|
|
|
|
|
|
def test_default_str_copre_i_valori_non_serializzabili(tmp_path):
|
|
"""json.dumps(default=str): un valore che non e' nativamente JSON (es. un Path, o un
|
|
dataclass senza __dict__ diretto) non deve far fallire l'append — diventa la sua str()."""
|
|
p = tmp_path / "runs.jsonl"
|
|
assert registra_giro(p, percorso=Path("/tmp/x")) is True
|
|
riga = json.loads(p.read_text().strip())
|
|
assert riga["percorso"] == str(Path("/tmp/x"))
|
|
|
|
|
|
def test_oserror_ritorna_false_senza_sollevare(tmp_path, capsys):
|
|
"""Un percorso impossibile (qui: dentro un FILE, non una cartella) e' un OSError alla
|
|
mkdir/scrittura. Il chiamante non deve vedere un'eccezione: la misura e' gia' fatta."""
|
|
ostacolo = tmp_path / "non-e-una-cartella"
|
|
ostacolo.write_text("sono un file")
|
|
p = ostacolo / "runs.jsonl"
|
|
assert registra_giro(p, stato="OK") is False
|
|
err = capsys.readouterr().err
|
|
assert "registra_giro" in err and str(p) in err
|
|
|
|
|
|
def test_un_valore_non_serializzabile_diventa_stringa_non_solleva(tmp_path):
|
|
p = tmp_path / "runs.jsonl"
|
|
oggetto = object()
|
|
assert registra_giro(p, oggetto=oggetto) is True
|
|
riga = json.loads(p.read_text().strip())
|
|
assert riga["oggetto"] == str(oggetto)
|