ROMFASTSQL/proxmox/lxc171-claude-agent/discord-bridge/tests/test_alerts.py

"""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")