ROMFASTSQL/proxmox/lxc171-claude-agent/maria-whatsapp-bridge/tests/test_escaladare.py

"""Escaladarea la suport: jurnalul si notificarea.

Nu atinge reteaua — `requests.post` e inlocuit si se verifica CE s-ar fi trimis.
"""

from __future__ import annotations

import json

import pytest

import config
import consumer


@pytest.fixture
def apeluri(monkeypatch):
    """Prinde apelurile HTTP catre punte, fara sa le execute."""
    prinse = []

    class Raspuns:
        status_code = 200

        def raise_for_status(self):
            pass

    def fals_post(url, json=None, timeout=None):
        prinse.append((url, json))
        return Raspuns()

    monkeypatch.setattr(consumer.requests, "post", fals_post)
    return prinse


@pytest.fixture
def rezultat_fara_acoperire():
    return {
        "chunks": [], "covered": False, "best_cosine": 0.55,
        "reason": "cosinus 0.550 < 0.58",
        "top": [{"source": "efactura_knowledge.md", "cosine": 0.55, "bm25": 0.0}],
    }


def _mesaj(**kw):
    baza = {"from": "40712345678@s.whatsapp.net", "pushName": "Ion",
            "id": "ABC123", "text": "cum schimb uleiul la masina"}
    baza.update(kw)
    return baza


def test_jurnalul_se_scrie_si_fara_canal_de_suport(apeluri, rezultat_fara_acoperire, tmp_path):
    # SUPPORT_JID nesetat: nu notificam pe nimeni, dar intrebarea nu are voie sa
    # dispara fara urma — asta e chiar problema pe care escaladarea o rezolva.
    consumer.escalate(_mesaj(), "cum schimb uleiul la masina", rezultat_fara_acoperire)

    fisiere = list((config.STATE_DIR / "escalations").glob("*.json"))
    assert len(fisiere) == 1
    inregistrare = json.loads(fisiere[0].read_text(encoding="utf-8"))
    assert inregistrare["notified"] is False
    assert inregistrare["reason"] == "cosinus 0.550 < 0.58"
    assert inregistrare["best_cosine"] == 0.55
    assert not apeluri


def test_referinta_e_stabila_pentru_acelasi_mesaj():
    # Se poate cita la telefon si nu se schimba daca escaladarea se reia.
    a = consumer.reference(1788212023.0, "ABC123")
    b = consumer.reference(1788212023.0, "ABC123")
    assert a == b
    assert a.startswith("M-")
    assert consumer.reference(1788212023.0, "ALT") != a


def test_referinta_ajunge_si_la_suport_si_in_jurnal(apeluri, rezultat_fara_acoperire, monkeypatch):
    monkeypatch.setitem(config._env, "SUPPORT_JID", "40799999999@s.whatsapp.net")
    inreg = consumer.escalate(_mesaj(), "intrebare", rezultat_fara_acoperire)
    _, corp = apeluri[0]
    assert inreg["ref"] in corp["text"]
    pe_disc = json.loads(next((config.STATE_DIR / "escalations").glob("*.json")).read_text())
    assert pe_disc["ref"] == inreg["ref"]


def test_notifica_suportul_ca_text(apeluri, rezultat_fara_acoperire, monkeypatch):
    monkeypatch.setitem(config._env, "SUPPORT_JID", "40799999999@s.whatsapp.net")
    consumer.escalate(_mesaj(), "cum schimb uleiul la masina", rezultat_fara_acoperire)

    url, corp = apeluri[0]
    assert url.endswith("/send")
    assert corp["to"] == "40799999999@s.whatsapp.net"
    assert "Ion" in corp["text"]
    assert "cum schimb uleiul" in corp["text"]

    inregistrare = json.loads(next((config.STATE_DIR / "escalations").glob("*.json")).read_text())
    assert inregistrare["notified"] is True


def test_captura_pleaca_la_suport_ca_imagine(apeluri, rezultat_fara_acoperire, monkeypatch, tmp_path):
    monkeypatch.setitem(config._env, "SUPPORT_JID", "40799999999@s.whatsapp.net")
    captura = tmp_path / "captura.png"
    captura.write_bytes(b"png")
    msg = _mesaj(text="", media={"path": str(captura)}, ocr_text="Eroare necunoscuta XYZ")

    consumer.escalate(msg, "intrebare", rezultat_fara_acoperire)

    url, corp = apeluri[0]
    assert url.endswith("/send-image")
    assert corp["path"] == str(captura)
    assert "Eroare necunoscuta XYZ" in corp["caption"]

    inregistrare = json.loads(next((config.STATE_DIR / "escalations").glob("*.json")).read_text())
    assert inregistrare["had_image"] is True
    assert inregistrare["ocr_text"] == "Eroare necunoscuta XYZ"


def test_notificarea_esuata_nu_pierde_jurnalul(rezultat_fara_acoperire, monkeypatch):
    monkeypatch.setitem(config._env, "SUPPORT_JID", "40799999999@s.whatsapp.net")

    def cade(*a, **kw):
        raise RuntimeError("puntea nu raspunde")

    monkeypatch.setattr(consumer.requests, "post", cade)
    consumer.escalate(_mesaj(), "intrebare", rezultat_fara_acoperire)

    inregistrare = json.loads(next((config.STATE_DIR / "escalations").glob("*.json")).read_text())
    assert inregistrare["notified"] is False
    assert "puntea nu raspunde" in inregistrare["notify_error"]


def test_puntea_respinge_mesajul_nu_promitem_ca_a_plecat(rezultat_fara_acoperire, monkeypatch):
    """503 „not connected to WhatsApp" nu ridica exceptie in requests — dar nu e trimitere.

    Fara verificarea codului, escaladarea se scria „notificata" si omul primea
    „te contacteaza cineva" pentru un mesaj care nu plecase nicaieri.
    """
    monkeypatch.setitem(config._env, "SUPPORT_JID", "40799999999@s.whatsapp.net")

    class Respins:
        status_code = 503
        text = '{"ok":false,"error":"not connected to WhatsApp"}'

    monkeypatch.setattr(consumer.requests, "post", lambda *a, **kw: Respins())
    consumer.escalate(_mesaj(), "intrebare", rezultat_fara_acoperire)

    inregistrare = json.loads(next((config.STATE_DIR / "escalations").glob("*.json")).read_text())
    assert inregistrare["notified"] is False
    assert "503" in inregistrare["notify_error"]


def test_raspunsul_omului_intra_in_fisierul_escaladarii(tmp_path, monkeypatch):
    """Perechea eroare + rezolvare, in acelasi fisier.

    Inainte, replica omului traia doar in firul din `conversations/`, care se
    sterge la TTL: jurnalul pastra intrebarile fara raspuns si niciun raspuns.
    """
    import json
    import time
    import config
    import consumer

    d = config.STATE_DIR / "escalations"
    d.mkdir(parents=True, exist_ok=True)
    (d / f"{int(time.time())}-M-TEST-OM.json").write_text(
        json.dumps({"ref": "M-TEST-OM", "text": "CUI cumparator incorect"}), encoding="utf-8")

    consumer.raspuns_de_om("M-TEST-OM", {"id": "1", "pushName": "Suport"},
                           "Se corecteaza CUI-ul partenerului si se retrimite factura.")
    consumer.raspuns_de_om("M-TEST-OM", {"id": "2", "pushName": "Suport"}, "   ")

    record = json.loads(next(d.glob("*-M-TEST-OM.json")).read_text(encoding="utf-8"))
    completari = record["completari"]
    assert len(completari) == 1, "mesajul gol nu are ce cauta in jurnal"
    assert completari[0]["fel"] == "raspuns_om"
    assert completari[0]["cine"] == "Suport"
    assert "retrimite" in completari[0]["text"]


def test_raspunsul_omului_pe_o_escaladare_care_nu_exista_nu_crapa():
    import consumer
    consumer.raspuns_de_om("M-NU-EXISTA", {"id": "1"}, "orice")