Jair Cedillo
16aca537be
test_efc_outbox.py y test_gateway_rutas.py importaban crm.expedientes, el módulo del
expediente paralelo que 5c4df59 descartó al rebasar sobre crm.cases. Con el
ModuleNotFoundError, pytest ni siquiera las coleccionaba: 28 pruebas de la máquina de
reintentos y del tablero de ops llevaban sin correr, justo las del carril que se está
extendiendo.
Se traducen al modelo vigente preservando cada invariante:
- el expediente es crm.cases y se llega por la liga case_id de la solicitud, en vez de
find_by_service_request;
- el folio es Case.reference, no .folio;
- la idempotencia que se probaba vía ensure_expediente ahora se ejercita en
replicate_expediente_best_effort, que es donde vive la guarda _expediente_ya_encolado.
No se relaja ninguna aserción.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
136 lines
5.0 KiB
Python
136 lines
5.0 KiB
Python
"""Contrato del tablero de ops del carril CRM -> EFC.
|
|
|
|
Lo que se fija aquí es lo que el frontend espera recibir: el 404 del reintento sobre una fila que no
|
|
existe (y **no** un 200 silencioso), la forma exacta de la respuesta de éxito, y el aislamiento por
|
|
tenant/company.
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from api.v1.modules.crm.expediente_gateway import routes
|
|
from api.v1.modules.crm.expediente_gateway import service as gateway
|
|
from api.v1.modules.crm.expediente_gateway.models import (
|
|
FILE_KIND_DOCUMENTO,
|
|
SOURCE_CRM_DOCUMENTS,
|
|
STATUS_FAILED,
|
|
STATUS_PENDING,
|
|
STATUS_SENT,
|
|
EfcFileOutbox,
|
|
)
|
|
from api.v1.modules.crm.cases.models import Case
|
|
from api.v1.modules.crm.service_requests import service as sr_service
|
|
from api.v1.modules.crm.service_requests.dto import ServiceRequestCreate
|
|
from tests.conftest import COMPANY_ID, TENANT_ID
|
|
|
|
OTRO_TENANT = 99
|
|
OTRA_COMPANY = 77
|
|
USUARIO = {"tenant_id": TENANT_ID, "sub": "user-1"}
|
|
|
|
|
|
@pytest.fixture()
|
|
def entorno(db, monkeypatch):
|
|
from core.config import settings
|
|
|
|
monkeypatch.setattr(settings, "EFC_API_URL", "https://efc.example.test/", raising=False)
|
|
monkeypatch.setattr(gateway, "_dispatch_delivery", lambda *a, **k: None)
|
|
monkeypatch.setattr(gateway, "_dispatch_file_delivery", lambda *a, **k: None)
|
|
monkeypatch.setattr(gateway, "_tenant_slug", lambda tid: ("temex", "TEMEX"))
|
|
|
|
solicitud = sr_service.create_service_request(
|
|
db, ServiceRequestCreate(operation_type="importacion"), TENANT_ID, COMPANY_ID, "user-1"
|
|
)
|
|
# El expediente es ``crm.cases`` y se llega a él por la liga ``case_id`` de la solicitud.
|
|
expediente = db.query(Case).filter(Case.id == solicitud.case_id).one()
|
|
return {"db": db, "expediente": expediente}
|
|
|
|
|
|
def _fila_archivo(db, expediente, **kwargs) -> EfcFileOutbox:
|
|
row = EfcFileOutbox(
|
|
kind=FILE_KIND_DOCUMENTO,
|
|
s3_key="k",
|
|
file_name="guia.pdf",
|
|
content_type="application/pdf",
|
|
efc_tipo="MBL",
|
|
source_table=SOURCE_CRM_DOCUMENTS,
|
|
source_id=kwargs.pop("source_id", 1),
|
|
crm_document_ref="CRMDOC-1-1",
|
|
expediente_ref=expediente.id,
|
|
status=kwargs.pop("status", STATUS_PENDING),
|
|
tenant_id=kwargs.pop("tenant_id", TENANT_ID),
|
|
company_id=kwargs.pop("company_id", COMPANY_ID),
|
|
**kwargs,
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
return row
|
|
|
|
|
|
def test_retry_de_una_fila_inexistente_da_404_con_mensaje_especifico(entorno):
|
|
"""**Es contrato con el frontend.** Un 200 le haría pintar «reencolado» cuando no hay nada que
|
|
entregar, y el usuario esperaría un badge que nunca va a cambiar."""
|
|
with pytest.raises(HTTPException) as exc:
|
|
routes.retry_outbox(999999, COMPANY_ID, "file", USUARIO, entorno["db"])
|
|
|
|
assert exc.value.status_code == 404
|
|
assert exc.value.detail == "Fila de outbox no encontrada"
|
|
|
|
|
|
def test_retry_exitoso_devuelve_requeued_con_el_id(entorno):
|
|
row = _fila_archivo(entorno["db"], entorno["expediente"], status=STATUS_FAILED)
|
|
|
|
resp = routes.retry_outbox(row.id, COMPANY_ID, "file", USUARIO, entorno["db"])
|
|
|
|
assert resp == {"status": "requeued", "id": row.id}
|
|
assert row.status == STATUS_PENDING
|
|
|
|
|
|
def test_retry_de_una_fila_de_otra_company_da_404(entorno):
|
|
"""No se filtra la existencia: para ese usuario la fila simplemente no existe."""
|
|
row = _fila_archivo(entorno["db"], entorno["expediente"], company_id=OTRA_COMPANY)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
routes.retry_outbox(row.id, COMPANY_ID, "file", USUARIO, entorno["db"])
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_metrics_cuenta_por_status(entorno):
|
|
db, expediente = entorno["db"], entorno["expediente"]
|
|
_fila_archivo(db, expediente, source_id=1, status=STATUS_FAILED)
|
|
_fila_archivo(db, expediente, source_id=2, status=STATUS_SENT)
|
|
|
|
metricas = routes.metrics(COMPANY_ID, USUARIO, db)
|
|
|
|
assert set(metricas) == {"pending", "sent", "failed"}
|
|
assert metricas["failed"] == 1
|
|
assert metricas["sent"] == 1
|
|
|
|
|
|
def test_metrics_no_ve_otro_tenant(entorno):
|
|
db, expediente = entorno["db"], entorno["expediente"]
|
|
_fila_archivo(db, expediente, source_id=3, status=STATUS_FAILED, tenant_id=OTRO_TENANT)
|
|
|
|
assert routes.metrics(COMPANY_ID, USUARIO, db)["failed"] == 0
|
|
|
|
|
|
def test_el_listado_solo_devuelve_lo_del_tenant_y_la_company(entorno):
|
|
db, expediente = entorno["db"], entorno["expediente"]
|
|
_fila_archivo(db, expediente, source_id=1)
|
|
_fila_archivo(db, expediente, source_id=2, tenant_id=OTRO_TENANT)
|
|
_fila_archivo(db, expediente, source_id=3, company_id=OTRA_COMPANY)
|
|
|
|
filas = routes.list_outbox(COMPANY_ID, "file", None, 100, USUARIO, db)
|
|
|
|
assert len(filas) == 1
|
|
assert filas[0]["source_id"] == 1
|
|
|
|
|
|
def test_el_listado_filtra_por_status(entorno):
|
|
db, expediente = entorno["db"], entorno["expediente"]
|
|
_fila_archivo(db, expediente, source_id=1, status=STATUS_FAILED)
|
|
_fila_archivo(db, expediente, source_id=2, status=STATUS_SENT)
|
|
|
|
fallidas = routes.list_outbox(COMPANY_ID, "file", STATUS_FAILED, 100, USUARIO, db)
|
|
|
|
assert [f["source_id"] for f in fallidas] == [1]
|