Files
descarga_factura_transporte/tests/test_respuesta_proveedor.py
T
GeorgeandClaude Sonnet 5.5 f0ad6aa78e se valida si la factura ya existe en SAP y se avisa cuando SAP está fuera de línea
- sapb1_manager.buscar_factura_existente busca la factura (proveedor, tipo,
  serie, número) en facturas de compra no anuladas y en drafts; si existe no
  se crea otro draft y se responde "ya está registrada" sin pisar el registro
  de una aceptación anterior.
- SapNoDisponible (subclase de ConnectionError) distingue una caída (timeout,
  sin conexión, HTTP 5xx) de un rechazo normal de SAP (HTTP 4xx). Una caída
  de Service Layer con HANA sin resultados ya no se lee como "guía no existe".
- Con SAP o HANA caídos se avisa una sola vez al proveedor con la plantilla
  "En cola" y el correo se reintenta en los ciclos siguientes (hasta 24), sin
  repetir avisos; al agotarse se avisa al equipo.
- La lógica por factura pasa a _procesar_factura; lo ya resuelto dentro de un
  correo no se repite al reintentarlo.
- seguimiento: columna intentos (con migración), registrar_intento y el
  estado de notificación en_espera.
- database_utilities.DB_NAME pasa a ser absoluto (junto al código).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-22 12:08:38 -05:00

296 lines
13 KiB
Python

import logging
import pytest
import observaciones as obs
import outlook_utilities
import plantillas
import respuesta_proveedor as rp
import seguimiento
FACTURA = {
"serie": "F002", "numero": "486", "tipo": "01", "ruc_emisor": "20537929520",
"razon_social_emisor": "TRANSPORTES SÁNCHEZ & HIJOS S.A.C.",
"lineas": [
{"total_linea": 624.38, "guias_remision": [{"serie": "T001", "numero": "1"}, {"serie": "T001", "numero": "2"}]},
{"total_linea": 100.0, "guias_remision": [{"serie": "T001", "numero": "3"}]},
],
}
RESULTADOS = [
{"ok": True, "guias_lt": [{"guia": {"serie": "T001", "numero": "1"}}, {"guia": {"serie": "T001", "numero": "2"}}]},
{"ok": False, "guias_lt": [{"guia": {"serie": "T001", "numero": "3"}}]},
]
EMAIL = {"id": "c1", "fecha": "2026-10-14 10:35:00", "subject": "Factura", "sender": "prov@empresa.com"}
@pytest.fixture
def registro():
return seguimiento.registrar_comprobante("c1", "20537929520", "01", "F002", "486")
# ============================================================================
# Configuración
# ============================================================================
@pytest.mark.parametrize("valor, esperado", [(None, "prueba"), ("", "prueba"), ("real", "real"), (" REAL ", "real"),
("prueba", "prueba"), ("produccion", "prueba")])
def test_modo(monkeypatch, valor, esperado):
if valor is not None:
monkeypatch.setenv("NOTIFICACION_MODO", valor)
assert rp.modo() == esperado
def test_equipo(monkeypatch):
assert rp.equipo() == rp.EQUIPO_POR_DEFECTO
monkeypatch.setenv("NOTIFICACION_EQUIPO", " a@x.com ; b@x.com,c@x.com ,, ")
assert rp.equipo() == ["a@x.com", "b@x.com", "c@x.com"]
def test_contacto_usa_el_buzon_si_no_se_define(monkeypatch):
assert rp.contacto_correo() == "buzon@aberio.com"
monkeypatch.setenv("NOTIFICACION_CONTACTO", "ayuda@aberio.com")
assert rp.contacto_correo() == "ayuda@aberio.com"
# ============================================================================
# Protecciones
# ============================================================================
BUZON = "buzon@aberio.com"
def _correo(**kv):
return {**EMAIL, **kv}
@pytest.mark.parametrize("sender", [None, "", "Desconocido"])
def test_sin_remitente(sender):
assert "desconocido" in rp.motivo_omision(_correo(sender=sender), BUZON)
@pytest.mark.parametrize("sender", ["test@local", "sin-arroba", "a@b", "a b@c.com"])
def test_direcciones_no_validas(sender):
assert "no válida" in rp.motivo_omision(_correo(sender=sender), BUZON)
def test_el_propio_buzon_aunque_cambie_la_capitalizacion():
assert "propio buzón" in rp.motivo_omision(_correo(sender="Buzon@ABERIO.com"), BUZON)
@pytest.mark.parametrize("sender", ["no-reply@x.com", "noreply@x.com", "No_Reply@x.com", "donotreply@x.com",
"do-not-reply@x.com", "mailer-daemon@x.com", "postmaster@x.com", "bounces@x.com",
"no-responder@x.com"])
def test_direcciones_automaticas(sender):
assert "automática" in rp.motivo_omision(_correo(sender=sender), BUZON)
def test_una_direccion_normal_con_palabras_parecidas_si_se_responde():
assert rp.motivo_omision(_correo(sender="reply.compras@empresa.com"), BUZON) is None
assert rp.motivo_omision(_correo(sender="ventas@empresa.com"), BUZON) is None
def test_mensaje_automatico():
assert "automático" in rp.motivo_omision(_correo(es_automatico=True), BUZON)
def test_respuestas_a_notificaciones_solo_se_omiten_si_no_hay_comprobante():
respuesta = _correo(subject="Re: Aceptado · Factura F002-1 · ABR-261019-00042")
assert rp.motivo_omision(respuesta, BUZON, sin_comprobante=True) is not None
assert rp.motivo_omision(respuesta, BUZON, sin_comprobante=False) is None # con comprobante se procesa normal
assert rp.motivo_omision(_correo(subject="RV: documentos"), BUZON, sin_comprobante=True) is not None
assert rp.motivo_omision(_correo(subject="Factura ABR-261019-00042"), BUZON, sin_comprobante=True) is not None
assert rp.motivo_omision(_correo(subject="Factura de octubre"), BUZON, sin_comprobante=True) is None
# ============================================================================
# Contextos
# ============================================================================
def test_guias_el_importe_de_la_linea_va_una_vez():
filas = rp.guias_para_plantilla(FACTURA, RESULTADOS)
assert filas == [
{"guia": "T001-1", "importe_factura": "S/ 624.38", "conforme": True},
{"guia": "T001-2", "importe_factura": "incluida en la línea", "conforme": True},
{"guia": "T001-3", "importe_factura": "S/ 100.00", "conforme": False},
]
def test_guias_sin_resultado_de_validacion_usa_las_del_xml_como_no_conformes():
filas = rp.guias_para_plantilla({"lineas": [{"total_linea": 5, "guias_remision": [{"serie": "T001", "numero": "9"}]}]}, [])
assert filas == [{"guia": "T001-9", "importe_factura": "S/ 5.00", "conforme": False}]
def test_linea_sin_guias_no_genera_filas():
assert rp.guias_para_plantilla({"lineas": [{"total_linea": 5, "guias_remision": []}]}, [{"ok": False, "guias_lt": []}]) == []
@pytest.mark.parametrize("vence, esperado", [("2026-11-13", "13/11/2026"), ("2026-10-14", None), ("2026-10-01", None)])
def test_fecha_estimada_de_pago_solo_si_hay_plazo(registro, vence, esperado):
draft = {"DocDate": "2026-10-14", "DocDueDate": vence}
contexto = rp.contexto_aceptado(EMAIL, registro, FACTURA, RESULTADOS, draft)
assert contexto["fecha_estimada_pago"] == esperado
def test_contexto_aceptado(registro):
contexto = rp.contexto_aceptado(EMAIL, registro, FACTURA, RESULTADOS, {})
assert contexto["asunto"] == f"Aceptado · Factura F002-486 · {registro['codigo_seguimiento']}"
assert contexto["recepcion_fecha"] == "14/10/2026" and contexto["recepcion_hora"] == "10:35"
assert contexto["comprobante_tipo"] == "Factura electrónica" and contexto["proveedor_ruc"] == "20537929520"
assert contexto["fecha_estimada_pago"] is None
def test_contexto_rechazo_numera_las_observaciones(registro):
lista = [obs.guia_no_encontrada("T001-3"), obs.precio_no_coincide(2, 100, 120, ["T001-3"], True)]
contexto = rp.contexto_rechazo_factura(EMAIL, registro, FACTURA, RESULTADOS, lista)
assert contexto["observaciones_texto"] == "2 observaciones"
assert [o["numero"] for o in contexto["observaciones"]] == [1, 2]
assert contexto["observaciones"][0]["filas"] == [{"etiqueta": "Guía", "valor": "T001-3"}]
assert contexto["tiene_guias"] is True and contexto["preheader"].startswith("Tiene 2 observaciones.")
def test_contexto_rechazo_de_correo_sin_comprobante(registro):
contexto = rp.contexto_rechazo_correo(EMAIL, registro, [obs.sin_pdf_xml(1)])
assert contexto["asunto"] == f"Rechazado · Comprobante no recibido · {registro['codigo_seguimiento']}"
assert contexto["comprobante_numero"] is None and contexto["proveedor_ruc"] is None
assert contexto["proveedor_razon_social"] == "prov@empresa.com"
assert contexto["guias"] == [] and contexto["tiene_guias"] is False
def test_los_tres_contextos_renderizan_sin_variables_faltantes(registro):
lista = [obs.sin_pdf_xml(1)]
for plantilla, contexto in [
("aceptado", rp.contexto_aceptado(EMAIL, registro, FACTURA, RESULTADOS, {"DocDate": "2026-10-14", "DocDueDate": "2026-11-13"})),
("rechazado", rp.contexto_rechazo_factura(EMAIL, registro, FACTURA, RESULTADOS, lista)),
("rechazado", rp.contexto_rechazo_correo(EMAIL, registro, lista)),
]:
html = plantillas.renderizar_plantilla(plantilla, contexto)
assert "{{" not in html and registro["codigo_seguimiento"] in html
def test_razon_social_con_ampersand_sale_escapada(registro):
contexto = rp.contexto_aceptado(EMAIL, registro, FACTURA, RESULTADOS, {})
assert "SÁNCHEZ &amp; HIJOS" in plantillas.renderizar_plantilla("aceptado", contexto)
# ============================================================================
# responder()
# ============================================================================
class _Enviados(list):
"""Correos que recibió Graph; ``ok`` controla si Graph los acepta."""
ok = True
@pytest.fixture
def graph(monkeypatch):
enviados = _Enviados()
def enviar(token, desde, para, asunto, html, adjuntos=None, cc_recipients=None, bcc_recipients=None):
enviados.append({"para": para, "bcc": bcc_recipients, "asunto": asunto})
return enviados.ok
monkeypatch.setattr(outlook_utilities, "send_email", enviar)
monkeypatch.setenv("NOTIFICACION_EQUIPO", "eq@aberio.com")
return enviados
def _contexto(registro):
return rp.contexto_aceptado(EMAIL, registro, FACTURA, RESULTADOS, {})
def test_responder_en_modo_real(graph, registro, monkeypatch):
monkeypatch.setenv("NOTIFICACION_MODO", "real")
assert rp.responder("t", BUZON, EMAIL, "aceptado", _contexto(registro), registro) == "enviada"
assert graph == [{"para": ["prov@empresa.com"], "bcc": ["eq@aberio.com"], "asunto": _contexto(registro)["asunto"]}]
fila = seguimiento.obtener(registro["codigo_seguimiento"])
assert (fila["notificacion_estado"], fila["notificacion_destinatario"]) == ("enviada", "prov@empresa.com")
def test_responder_en_modo_prueba(graph, registro):
assert rp.responder("t", BUZON, EMAIL, "aceptado", _contexto(registro), registro) == "enviada"
(envio,) = graph
assert envio["para"] == ["eq@aberio.com"] and envio["bcc"] is None
assert envio["asunto"].startswith("[PRUEBA → prov@empresa.com] Aceptado")
def test_responder_omite_y_anota_el_motivo(graph, registro):
estado = rp.responder("t", BUZON, _correo(sender="no-reply@x.com"), "aceptado", _contexto(registro), registro)
assert estado == "omitida" and graph == []
fila = seguimiento.obtener(registro["codigo_seguimiento"])
assert fila["notificacion_estado"] == "omitida" and "automática" in fila["notificacion_error"]
def test_responder_si_graph_falla(graph, registro):
graph.ok = False
assert rp.responder("t", BUZON, EMAIL, "aceptado", _contexto(registro), registro) == "fallida"
fila = seguimiento.obtener(registro["codigo_seguimiento"])
assert fila["notificacion_estado"] == "fallida" and "Graph" in fila["notificacion_error"]
def test_responder_si_la_plantilla_no_se_puede_armar_no_envia_nada(graph, registro):
contexto = _contexto(registro)
del contexto["codigo_seguimiento"]
assert rp.responder("t", BUZON, EMAIL, "aceptado", contexto, registro) == "fallida"
assert graph == []
assert "codigo_seguimiento" in seguimiento.obtener(registro["codigo_seguimiento"])["notificacion_error"]
def test_el_limite_solo_aplica_a_correos_sin_comprobante(graph, monkeypatch):
import database_utilities
monkeypatch.setenv("NOTIFICACION_MODO", "real")
database_utilities.save_emails([{"id": f"e{n}", "fecha": "2026-10-14 10:00:00", "sender": "prov@empresa.com",
"subject": "s", "has_attach": True} for n in range(5)])
estados = []
for n in range(5):
reg = seguimiento.registrar_correo(f"e{n}")
email = _correo(id=f"e{n}")
estados.append(rp.responder("t", BUZON, email, "rechazado", rp.contexto_rechazo_correo(email, reg, [obs.sin_pdf_xml(1)]),
reg, sin_comprobante=True))
assert estados == ["enviada"] * rp.MAX_RESPUESTAS_CORREO + ["omitida"] * (5 - rp.MAX_RESPUESTAS_CORREO)
def test_encabezado_tecnico(registro):
html = rp.encabezado_tecnico(EMAIL, registro)
assert "prov@empresa.com" in html and registro["codigo_seguimiento"] in html and "prueba" in html
assert "sin código" in rp.encabezado_tecnico(EMAIL, None)
def test_enviar_a_equipo(graph):
assert rp.enviar_a_equipo("t", BUZON, "Asunto", "<p/>", ["a.xml"]) is True
assert graph[0]["para"] == ["eq@aberio.com"]
def test_contexto_fuera_de_linea_renderiza(registro):
contexto = rp.contexto_fuera_de_linea(EMAIL, registro, FACTURA)
assert contexto["asunto"] == f"En cola · Factura F002-486 · {registro['codigo_seguimiento']}"
html = plantillas.renderizar_plantilla("fuera-de-linea", contexto)
assert "{{" not in html and "SÁNCHEZ &amp; HIJOS" in html
def test_responder_con_estado_de_exito_en_espera(graph, registro, monkeypatch):
monkeypatch.setenv("NOTIFICACION_MODO", "real")
estado = rp.responder("t", BUZON, EMAIL, "fuera-de-linea", rp.contexto_fuera_de_linea(EMAIL, registro, FACTURA),
registro, estado_exito="en_espera")
assert estado == "en_espera" and graph[0]["para"] == ["prov@empresa.com"]
assert seguimiento.obtener(registro["codigo_seguimiento"])["notificacion_estado"] == "en_espera"
def test_si_el_aviso_falla_queda_fallida_aunque_el_exito_sea_en_espera(graph, registro):
graph.ok = False
assert rp.responder("t", BUZON, EMAIL, "fuera-de-linea", rp.contexto_fuera_de_linea(EMAIL, registro, FACTURA),
registro, estado_exito="en_espera") == "fallida"