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