"""Teste pentru alerts.py (T12). Fara retea, fara email real.""" import os import pathlib import sys import time import pytest sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) import alerts # noqa: E402 @pytest.fixture() def sandbox(tmp_path, monkeypatch): """Muta starea si logurile in tmp, si inlocuieste trimiterea reala de email.""" monkeypatch.setattr(alerts, "STATE_DIR", tmp_path) monkeypatch.setattr(alerts, "LOG_DIR", tmp_path / "logs") monkeypatch.setattr(alerts, "DEDUP_FILE", tmp_path / "alerts-dedup.json") monkeypatch.setattr(alerts, "LOG_FILE", tmp_path / "logs" / "alerts.log") sent = [] real_send = alerts._send_mail def fake_send(subject_line, body): sent.append((subject_line, body)) return True, "trimis (fals)" monkeypatch.setattr(alerts, "_send_mail", fake_send) return {"dir": tmp_path, "sent": sent, "real_send": real_send} def _log_text(sandbox): path = sandbox["dir"] / "logs" / "alerts.log" return path.read_text() if path.exists() else "" # --- format ---------------------------------------------------------------- def test_subiect_are_prefixul_de_nivel(sandbox): alerts.alert("CRITICAL", "state.json corupt", "detalii") assert sandbox["sent"][0][0] == "[CRITICAL] state.json corupt" def test_nivel_necunoscut_devine_warn(sandbox): alerts.alert("chestii", "ceva", "corp") assert sandbox["sent"][0][0].startswith("[WARN] ") def test_nivel_case_insensitive(sandbox): alerts.alert("info", "ceva", "corp") assert sandbox["sent"][0][0].startswith("[INFO] ") def test_corpul_contine_contextul_de_host_si_sursa(sandbox): alerts.alert("WARN", "subiect", "linia mea") body = sandbox["sent"][0][1] assert "linia mea" in body assert "Host:" in body and "Sursa:" in body def test_corp_urias_e_trunchiat(sandbox): alerts.alert("WARN", "s", "x" * (alerts.MAX_BODY_CHARS + 5000)) assert "corp trunchiat" in sandbox["sent"][0][1] def test_subiect_multiline_e_pliat_pe_o_linie(sandbox): alerts.alert("WARN", "prima\nlinia a doua", "corp") assert "\n" not in sandbox["sent"][0][0] # --- dedup ----------------------------------------------------------------- def test_dedup_blocheaza_a_doua_alerta_cu_aceeasi_cheie(sandbox): alerts.alert("WARN", "proces mort", "a", dedup_key="proc-died:42") alerts.alert("WARN", "proces mort", "b", dedup_key="proc-died:42") assert len(sandbox["sent"]) == 1 assert "SKIP(dedup" in _log_text(sandbox) def test_dedup_e_pe_cheie_nu_pe_subiect(sandbox): alerts.alert("WARN", "proces mort", "a", dedup_key="proc-died:1") alerts.alert("WARN", "proces mort", "b", dedup_key="proc-died:2") assert len(sandbox["sent"]) == 2 def test_fara_dedup_key_se_trimite_de_fiecare_data(sandbox): for _ in range(3): alerts.alert("INFO", "acelasi subiect", "corp") assert len(sandbox["sent"]) == 3 def test_dedup_expira_dupa_o_ora(sandbox, monkeypatch): alerts.alert("WARN", "cost", "a", dedup_key="cost-cap") assert len(sandbox["sent"]) == 1 real_time = time.time monkeypatch.setattr(alerts.time, "time", lambda: real_time() + alerts.DEDUP_WINDOW_S + 10) alerts.alert("WARN", "cost", "b", dedup_key="cost-cap") assert len(sandbox["sent"]) == 2 def test_dedup_persistat_pe_disc(sandbox): alerts.alert("WARN", "x", "y", dedup_key="k") assert (sandbox["dir"] / "alerts-dedup.json").exists() def test_dedup_curata_intrarile_expirate(sandbox): import json vechi = {"expirat": time.time() - 2 * alerts.DEDUP_WINDOW_S} (sandbox["dir"] / "alerts-dedup.json").write_text(json.dumps(vechi)) alerts.alert("WARN", "x", "y", dedup_key="nou") data = json.loads((sandbox["dir"] / "alerts-dedup.json").read_text()) assert "expirat" not in data and "nou" in data def test_dedup_corupt_nu_opreste_alerta(sandbox): (sandbox["dir"] / "alerts-dedup.json").write_text("{{{ nu e json") alerts.alert("CRITICAL", "important", "corp", dedup_key="k") assert len(sandbox["sent"]) == 1 # --- degradare si robustete ------------------------------------------------ def test_lipsa_binarului_mail_nu_arunca_si_ramane_in_log(sandbox, monkeypatch): # aici vrem trimiterea REALA, ca sa vedem degradarea cand `mail` lipseste monkeypatch.setattr(alerts, "_send_mail", sandbox["real_send"]) monkeypatch.setattr(alerts, "_mail_binary", lambda: None) alerts.alert("CRITICAL", "fara mail", "corpul contine ceva important") text = _log_text(sandbox) assert "NESENT" in text assert "binarul `mail` lipseste" in text assert "corpul contine ceva important" in text def test_esecul_mail_nu_arunca(sandbox, monkeypatch): monkeypatch.setattr(alerts, "_send_mail", lambda s, b: (False, "cod 1")) alerts.alert("WARN", "esec", "corp") # nu trebuie sa arunce assert "NESENT" in _log_text(sandbox) def test_exceptie_interna_nu_scapa_din_alert(sandbox, monkeypatch): def explodeaza(*args, **kwargs): raise RuntimeError("boom") monkeypatch.setattr(alerts, "_send_mail", explodeaza) alerts.alert("CRITICAL", "boom", "corp") # contractul: NU arunca niciodata def test_alert_nu_arunca_nici_cand_logul_e_inaccesibil(sandbox, monkeypatch): # LOG_DIR indica un fisier, deci mkdir si open esueaza amandoua fisier = sandbox["dir"] / "blocaj" fisier.write_text("x") monkeypatch.setattr(alerts, "LOG_DIR", fisier / "logs") monkeypatch.setattr(alerts, "LOG_FILE", fisier / "logs" / "alerts.log") monkeypatch.setattr(alerts, "_send_mail", lambda s, b: (False, "nimic")) alerts.alert("WARN", "x", "y") def test_argumente_aiurea_nu_arunca(sandbox): alerts.alert(None, None, None) # type: ignore[arg-type] alerts.alert("", "", "", dedup_key="") alerts.alert(123, 456, 789) # type: ignore[arg-type] # --- destinatar ------------------------------------------------------------ def test_destinatar_implicit_root(monkeypatch): monkeypatch.delenv("ALERT_RECIPIENT", raising=False) monkeypatch.setattr(alerts, "_config", None) assert alerts._recipient() == "root" def test_destinatar_din_mediu(monkeypatch): monkeypatch.setattr(alerts, "_config", None) monkeypatch.setenv("ALERT_RECIPIENT", "ops@romfast.ro") assert alerts._recipient() == "ops@romfast.ro" def test_destinatar_din_config_daca_exista(monkeypatch): class FakeConfig: @staticmethod def get(key, default=None): return "din-config@romfast.ro" if key == "ALERT_RECIPIENT" else default monkeypatch.setattr(alerts, "_config", FakeConfig) assert alerts._recipient() == "din-config@romfast.ro" def test_config_care_arunca_nu_rupe_destinatarul(monkeypatch): class BadConfig: @staticmethod def get(key, default=None): raise RuntimeError("config stricat") monkeypatch.setattr(alerts, "_config", BadConfig) monkeypatch.delenv("ALERT_RECIPIENT", raising=False) assert alerts._recipient() == "root" # --- ajutoarele pentru conditiile Lane A ---------------------------------- def test_ajutoarele_produc_alerte_cu_dedup(sandbox): alerts.alert_process_died("123", 999, "SIGKILL") alerts.alert_crash_loop("123", 5, 300) alerts.alert_cost_cap(5.12, 5.00) alerts.alert_state_corrupt("/tmp/state.json") alerts.alert_orphans([{"pid": 1, "cmdline": "claude", "age_s": 10, "rss_mb": 406}]) assert len(sandbox["sent"]) == 5 niveluri = [s[0].split("]")[0] + "]" for s in sandbox["sent"]] assert "[CRITICAL]" in niveluri and "[WARN]" in niveluri # a doua oara acelasi lucru nu se retrimite alerts.alert_cost_cap(5.20, 5.00) assert len(sandbox["sent"]) == 5 def test_alert_orphans_cu_lista_goala_nu_arunca(sandbox): alerts.alert_orphans([]) assert len(sandbox["sent"]) == 1 # --- integrare reala cu subprocess (fara retea) --------------------------- def test_send_mail_foloseste_argumentele_corecte(tmp_path, monkeypatch): """Verifica linia de comanda construita, cu un `mail` fals scris pe disc.""" fake = tmp_path / "mail" marker = tmp_path / "primit.txt" fake.write_text( "#!/bin/sh\n" f'{{ echo "ARGS: $@"; cat; }} > "{marker}"\n' ) fake.chmod(0o755) monkeypatch.setattr(alerts, "_mail_binary", lambda: str(fake)) monkeypatch.setattr(alerts, "_config", None) monkeypatch.setenv("ALERT_RECIPIENT", "test@local") ok, detail = alerts._send_mail("[WARN] subiect de test", "corp de test") assert ok, detail text = marker.read_text() assert "ARGS: -s [WARN] subiect de test test@local" in text assert "corp de test" in text def test_send_mail_cod_de_eroare_e_raportat(tmp_path, monkeypatch): fake = tmp_path / "mail" fake.write_text("#!/bin/sh\necho 'MTA indisponibil'\nexit 75\n") fake.chmod(0o755) monkeypatch.setattr(alerts, "_mail_binary", lambda: str(fake)) ok, detail = alerts._send_mail("[WARN] x", "y") assert not ok and "75" in detail and "MTA indisponibil" in detail def test_send_mail_timeout_e_abandonat(tmp_path, monkeypatch): fake = tmp_path / "mail" fake.write_text("#!/bin/sh\nsleep 30\n") fake.chmod(0o755) monkeypatch.setattr(alerts, "_mail_binary", lambda: str(fake)) monkeypatch.setattr(alerts, "MAIL_TIMEOUT_S", 0.5) ok, detail = alerts._send_mail("[WARN] x", "y") assert not ok and "depasit" in detail def test_send_mail_binar_inexistent_nu_arunca(monkeypatch): monkeypatch.setattr(alerts, "_mail_binary", lambda: "/nu/exista/mail") ok, detail = alerts._send_mail("[WARN] x", "y") assert not ok and "nu a putut fi lansat" in detail def test_fisierele_de_stare_sunt_in_state_dir(sandbox): alerts.alert("INFO", "x", "y", dedup_key="k") nume = {p.name for p in sandbox["dir"].iterdir()} assert "alerts-dedup.json" in nume assert not any(n.endswith(".tmp") for n in nume), "fisier temporar ramas in urma" assert os.path.exists(sandbox["dir"] / "logs" / "alerts.log")