descarga_factura_transporte/tests/test_sapb1_manager.py

import pytest
import requests

import hana_manager
import sapb1_manager as s


class _Resp:
    def __init__(self, status=200, body=None, text=""):
        self.status_code, self._body, self.text = status, body if body is not None else {}, text
        self.content = b"x"
        self.cookies = {}

    def json(self):
        return self._body


@pytest.fixture(autouse=True)
def sesion_abierta(monkeypatch):
    monkeypatch.setitem(s._sap_session, "headers", {"Cookie": "B1SESSION=x"})


# ============================================================================
# Clasificación de errores: caída vs rechazo de SAP
# ============================================================================
@pytest.mark.parametrize("error", [requests.exceptions.ConnectTimeout("t"), requests.exceptions.ReadTimeout("t"),
                                   requests.exceptions.ConnectionError("c")])
def test_timeout_o_sin_conexion_es_sap_no_disponible(monkeypatch, error):
    def falla(**k):
        raise error
    monkeypatch.setattr(s.requests, "request", lambda **k: falla(**k))
    with pytest.raises(s.SapNoDisponible):
        s.forward_to_sapb1("GET", "Drafts")


@pytest.mark.parametrize("status", [500, 502, 503])
def test_http_5xx_es_sap_no_disponible(monkeypatch, status):
    monkeypatch.setattr(s.requests, "request", lambda **k: _Resp(status, {"error": {"message": {"value": "caído"}}}))
    with pytest.raises(s.SapNoDisponible, match=f"HTTP {status}"):
        s.forward_to_sapb1("GET", "Drafts")


def test_http_4xx_es_un_rechazo_no_una_caida(monkeypatch):
    monkeypatch.setattr(s.requests, "request", lambda **k: _Resp(400, {"error": {"message": {"value": "ItemCode inexistente"}}}))
    with pytest.raises(ConnectionError, match="ItemCode inexistente") as ex:
        s.forward_to_sapb1("POST", "Drafts", json_data={})
    assert not isinstance(ex.value, s.SapNoDisponible)


def test_sap_no_disponible_sigue_siendo_connection_error():
    assert issubclass(s.SapNoDisponible, ConnectionError)


def test_login_sin_conexion_es_caida(monkeypatch):
    def falla(*a, **k):
        raise requests.exceptions.ConnectTimeout("t")
    monkeypatch.setattr(s.requests, "post", falla)
    with pytest.raises(s.SapNoDisponible):
        s.login()


def test_login_5xx_es_caida_pero_credenciales_malas_no(monkeypatch):
    monkeypatch.setattr(s.requests, "post", lambda *a, **k: _Resp(503, text="down"))
    with pytest.raises(s.SapNoDisponible):
        s.login()

    monkeypatch.setattr(s.requests, "post", lambda *a, **k: _Resp(401, text="Invalid credentials"))
    with pytest.raises(ConnectionError) as ex:
        s.login()
    assert not isinstance(ex.value, s.SapNoDisponible)


# ============================================================================
# Guías: una caída no se confunde con "no existe"
# ============================================================================
def _forward(respuestas):
    """forward_to_sapb1 simulado: ``respuestas`` es una función (metodo, entidad) -> dict | excepción."""
    def forward(method, endpoint_path, **k):
        r = respuestas(method, endpoint_path)
        if isinstance(r, Exception):
            raise r
        return _Resp(200, r)
    return forward


def test_sl_caido_y_hana_sin_resultados_es_caida_no_guia_inexistente(monkeypatch):
    monkeypatch.setattr(s, "forward_to_sapb1", _forward(lambda m, e: s.SapNoDisponible("sin conexión")))
    monkeypatch.setattr(hana_manager, "get_delivery", lambda *a: [])

    with pytest.raises(s.SapNoDisponible):
        s.get_delivery_notes("09", "T001", "1")


def test_sl_caido_pero_hana_encuentra_la_guia_sigue_adelante(monkeypatch):
    monkeypatch.setattr(s, "forward_to_sapb1", _forward(lambda m, e: s.SapNoDisponible("sin conexión")))
    monkeypatch.setattr(hana_manager, "get_delivery", lambda *a: [{"DocEntry": 7, "Tabla": "ODLN"}])

    entradas = s._get_delivery_entries("09", "T001", "1")

    assert entradas == [{"DocEntry": 7, "Tabla": "ODLN", "Entidad": "DeliveryNotes"}]


def test_guia_que_de_verdad_no_existe_devuelve_lista_vacia(monkeypatch):
    monkeypatch.setattr(s, "forward_to_sapb1", _forward(lambda m, e: {"value": []}))
    monkeypatch.setattr(hana_manager, "get_delivery", lambda *a: [])

    assert s.get_delivery_notes("09", "T001", "1") == []


def test_un_rechazo_normal_de_sl_con_hana_vacio_no_es_caida(monkeypatch):
    monkeypatch.setattr(s, "forward_to_sapb1", _forward(lambda m, e: ConnectionError("HTTP 400 - mal filtro")))
    monkeypatch.setattr(hana_manager, "get_delivery", lambda *a: [])

    assert s.get_delivery_notes("09", "T001", "1") == []


def test_una_caida_al_leer_el_documento_se_propaga(monkeypatch):
    def respuestas(metodo, entidad):
        if entidad == "DeliveryNotes":
            return {"value": [{"DocEntry": 7}]}
        return s.SapNoDisponible("se cayó a mitad")                  # GET DeliveryNotes(7)

    monkeypatch.setattr(s, "forward_to_sapb1", _forward(respuestas))

    with pytest.raises(s.SapNoDisponible):
        s.get_delivery_notes("09", "T001", "1")


# ============================================================================
# buscar_factura_existente
# ============================================================================
class _Grabadora:
    def __init__(self, existentes):
        self.llamadas, self.existentes = [], existentes

    def __call__(self, method, endpoint_path, **k):
        self.llamadas.append((endpoint_path, k["params"]))
        valor = self.existentes.get(endpoint_path)
        if isinstance(valor, Exception):
            raise valor
        return _Resp(200, {"value": [valor] if valor else []})


def test_encuentra_la_factura_de_compra(monkeypatch):
    graba = _Grabadora({"PurchaseInvoices": {"DocEntry": 10, "DocNum": 5}})
    monkeypatch.setattr(s, "forward_to_sapb1", graba)

    assert s.buscar_factura_existente("20537929520", "01", "F002", "486") == \
        {"entidad": "PurchaseInvoices", "DocEntry": 10, "DocNum": 5}

    entidad, params = graba.llamadas[0]
    assert len(graba.llamadas) == 1                                      # no hace falta mirar los drafts
    for trozo in ("CardCode eq 'PL20537929520'", "U_BPP_MDTD eq '01'", "U_BPP_MDSD eq 'F002'",
                  "U_BPP_MDCD eq '486'", "Cancelled eq 'tNO'"):
        assert trozo in params["$filter"]


def test_si_no_es_factura_busca_en_los_drafts(monkeypatch):
    graba = _Grabadora({"Drafts": {"DocEntry": 3, "DocNum": 3}})
    monkeypatch.setattr(s, "forward_to_sapb1", graba)

    assert s.buscar_factura_existente("20537929520", "01", "F002", "486")["entidad"] == "Drafts"
    assert [l[0] for l in graba.llamadas] == ["PurchaseInvoices", "Drafts"]
    assert "DocObjectCode eq 'oPurchaseInvoices'" in graba.llamadas[1][1]["$filter"]


def test_si_no_existe_devuelve_none(monkeypatch):
    monkeypatch.setattr(s, "forward_to_sapb1", _Grabadora({}))
    assert s.buscar_factura_existente("20537929520", "01", "F002", "486") is None


def test_las_comillas_se_escapan_en_el_filtro(monkeypatch):
    graba = _Grabadora({})
    monkeypatch.setattr(s, "forward_to_sapb1", graba)

    s.buscar_factura_existente("20537929520", "01", "F0'02", "4' or '1'='1")

    filtro = graba.llamadas[0][1]["$filter"]
    assert "U_BPP_MDSD eq 'F0''02'" in filtro and "U_BPP_MDCD eq '4'' or ''1''=''1'" in filtro


def test_una_caida_al_buscar_se_propaga(monkeypatch):
    monkeypatch.setattr(s, "forward_to_sapb1", _Grabadora({"PurchaseInvoices": s.SapNoDisponible("caído")}))
    with pytest.raises(s.SapNoDisponible):
        s.buscar_factura_existente("20537929520", "01", "F002", "486")