ADUANASOFT/CRM_AGENTES_CARGA
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f9258bad053bcbd0824c29e0979c65dd2a23aecd
CRM_AGENTES_CARGA/backend/tests/test_fin_stamping.py
Jair Cedillo 6e208876f7 feat(fin): timbrado de CFDI 4.0 de ingreso con Comercio Digital 
Cierra el ciclo de la factura: construcción del comprobante, sellado con el CSD de la
empresa emisora y transmisión al PAC.

- cfdi_builder: XML 4.0 de ingreso en el orden de atributos del XSD, del que depende la
  cadena original y con ella el sello. Todo el dinero con Decimal.
- sealer: cadena original vía el XSLT oficial del SAT y firma con la llave del CSD.
- pac_comercio_digital: cliente de timbrarV5. Conserva el código y el saldo de folios que
  el legado leía en una variable que descartaba (CFDI.cs:19324-19336).
- csd_service y core/crypto: CSD por empresa, con la contraseña cifrada en la base. Antes
  el certificado había que dejarlo a mano en el almacenamiento y su contraseña era una
  variable de entorno global, lo que no funciona con varias empresas emisoras.
- Cada intento —también los rechazados— guarda el XML que se transmitió y el que contestó
  el PAC: sin ese par no hay forma de reconstruir un rechazo cuando termina la petición.

La declaración XML se escribe a mano con comillas dobles. lxml la emite con comillas
simples, que es XML válido, pero Comercio Digital compara la cadena literal version="1.0"
y responde 642 "la versión del XML no es 1.0".

El modo (pruebas o producción) sale de invoices.stamping_mode y no se puede pasar por la
API: es lo único que separa un timbre de prueba de un CFDI con validez fiscal.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-11 09:07:05 -05:00

548 lines
20 KiB
Python
Raw Blame History

"""Pruebas del timbrado de CFDI 4.0 de ingreso. Ninguna sale a la red.
El CSD es autofirmado y se genera en el fixture: para comprobar que *sellamos bien* basta con
que el sello verifique contra la llave pública de su propio certificado, y así la suite no
depende de descargar nada ni de que exista un CSD en disco. El CSD real de pruebas del SAT
hace falta para timbrar contra el PAC de verdad, que es la prueba de integración aparte.
"""
import base64
import datetime
from decimal import Decimal
import httpx
import pytest
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.x509.oid import NameOID
from lxml import etree
from api.v1.modules.fin.stamping import cfdi_builder as B
from api.v1.modules.fin.stamping import pac_comercio_digital as pac
from api.v1.modules.fin.stamping import sealer, service
from core import s3_keys
CFDI_NS = "http://www.sat.gob.mx/cfd/4"
CSD_PASSWORD = "12345678a"
# Número de serie con el formato del SAT: 20 dígitos cuyos bytes son sus caracteres ASCII.
CERT_NUMBER = "00001000000700000001"
# ---------------------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------------------
@pytest.fixture(scope="module")
def csd():
"""CSD autofirmado con el formato del SAT: devuelve (cer_der, key_der_cifrada)."""
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
nombre = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "EMISOR DE PRUEBA")])
cert = (
x509.CertificateBuilder()
.subject_name(nombre)
.issuer_name(nombre)
.public_key(key.public_key())
.serial_number(int.from_bytes(CERT_NUMBER.encode("ascii"), "big"))
.not_valid_before(datetime.datetime(2026, 1, 1))
.not_valid_after(datetime.datetime(2035, 1, 1))
.sign(key, hashes.SHA256())
)
return (
cert.public_bytes(serialization.Encoding.DER),
key.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.PKCS8,
serialization.BestAvailableEncryption(CSD_PASSWORD.encode()),
),
)
def _data(**overrides) -> B.CfdiData:
"""CFDI de ingreso completo y válido; los kwargs sustituyen campos para los casos negativos."""
base = dict(
folio="1001",
serie="A",
date="2026-08-07T10:00:00",
payment_form="03",
payment_method="PUE",
currency="MXN",
exchange_rate=None,
expedition_zip="64000",
payment_conditions=None,
issuer_rfc="EKU9003173C9",
issuer_name="EMISOR DE PRUEBA",
issuer_tax_regime="601",
receiver_rfc="XAXX010101000",
receiver_name="PUBLICO EN GENERAL",
receiver_zip="64000",
receiver_tax_regime="616",
receiver_cfdi_use="S01",
concepts=[
B.ConceptLine(
product_service_code="78101800",
unit_code="E48",
description="Flete maritimo",
quantity=Decimal("1"),
unit_price=Decimal("1000.00"),
tax_object="02",
taxes=[B.TaxLine(code="002", rate=Decimal("0.160000"), amount=Decimal("160.00"))],
)
],
)
base.update(overrides)
return B.CfdiData(**base)
def _xml(csd, data=None) -> bytes:
numero, cert_b64 = sealer.read_certificate(csd[0])
return B.build_xml(data or _data(), cert_number=numero, cert_b64=cert_b64)
# ---------------------------------------------------------------------------------------
# Construcción del XML
# ---------------------------------------------------------------------------------------
def test_comprobante_tipo_ingreso_con_atributos_obligatorios(csd):
root = etree.fromstring(_xml(csd))
assert root.tag == f"{{{CFDI_NS}}}Comprobante"
assert root.get("Version") == "4.0"
assert root.get("TipoDeComprobante") == "I"
assert root.get("Exportacion") == "01"
assert root.get("SubTotal") == "1000.00"
assert root.get("Total") == "1160.00"
def test_declaracion_xml_con_comillas_dobles(csd):
"""Comercio Digital compara la cadena literal version="1.0".
lxml emitiría comillas simples —válido según la especificación, pero su validador lo
rechaza con el código 642 "la versión del XML no es 1.0"—, así que la declaración se
antepone a mano y tiene que sobrevivir también al sellado, que es lo que se transmite.
"""
esperado = b'<?xml version="1.0" encoding="UTF-8"?>'
xml = _xml(csd)
assert xml.startswith(esperado)
assert B.apply_seal(xml, "SELLO-DE-PRUEBA").startswith(esperado)
def test_fecha_sin_desplazamiento_horario(csd):
"""En CFDI 4.0 la fecha va sin offset; el '-06:00' era de 3.3 (CFDI.cs:12654)."""
fecha = etree.fromstring(_xml(csd)).get("Fecha")
assert fecha == "2026-08-07T10:00:00"
assert "+" not in fecha and not fecha.endswith("Z")
def test_orden_de_atributos_del_comprobante(csd):
"""El orden importa: la cadena original —y con ella el sello— se calcula recorriéndolo."""
root = etree.fromstring(_xml(csd))
orden = [k for k in root.attrib if not k.startswith("{")]
esperado = [
"Version",
"Serie",
"Folio",
"Fecha",
"Sello",
"FormaPago",
"NoCertificado",
"Certificado",
"SubTotal",
"Moneda",
"Total",
"TipoDeComprobante",
"Exportacion",
"MetodoPago",
"LugarExpedicion",
]
assert orden == esperado
def test_atributos_opcionales_se_omiten(csd):
root = etree.fromstring(_xml(csd, _data(serie=None, folio=None, payment_conditions=None)))
assert "Serie" not in root.attrib
assert "CondicionesDePago" not in root.attrib
# Moneda MXN: sin TipoCambio
assert "TipoCambio" not in root.attrib
def test_moneda_extranjera_exige_tipo_de_cambio():
with pytest.raises(B.CfdiBuildError) as exc:
B.build_xml(_data(currency="USD", exchange_rate=None))
assert any("tipo de cambio" in m for m in exc.value.missing)
def test_impuestos_del_concepto_y_totales(csd):
root = etree.fromstring(_xml(csd))
traslado = root.find(
f"{{{CFDI_NS}}}Conceptos/{{{CFDI_NS}}}Concepto/{{{CFDI_NS}}}Impuestos"
f"/{{{CFDI_NS}}}Traslados/{{{CFDI_NS}}}Traslado"
)
assert traslado.get("Base") == "1000.00"
assert traslado.get("Impuesto") == "002"
assert traslado.get("TipoFactor") == "Tasa"
assert traslado.get("TasaOCuota") == "0.160000" # el SAT exige 6 decimales
assert traslado.get("Importe") == "160.00"
totales = root.find(f"{{{CFDI_NS}}}Impuestos")
assert totales.get("TotalImpuestosTrasladados") == "160.00"
def test_retenciones_restan_del_total():
data = _data()
data.concepts[0].taxes.append(
B.TaxLine(
code="001", rate=Decimal("0.100000"), amount=Decimal("100.00"), is_withholding=True
)
)
# 1000 + 160 - 100
assert data.total == Decimal("1060.00")
def test_partida_objeto_de_impuesto_sin_impuestos_es_error():
"""ObjetoImp '02' obliga al desglose. No se inventa una tasa por defecto."""
data = _data()
data.concepts[0].taxes = []
with pytest.raises(B.CfdiBuildError) as exc:
B.build_xml(data)
assert any("objeto de impuesto" in m for m in exc.value.missing)
def test_validacion_reporta_todos_los_faltantes_juntos():
"""Quien captura la factura necesita la lista completa, no descubrirlos de uno en uno."""
data = _data(issuer_rfc="", receiver_rfc="", payment_form="", expedition_zip="")
with pytest.raises(B.CfdiBuildError) as exc:
B.build_xml(data)
assert len(exc.value.missing) >= 4
def test_importes_con_decimal_no_arrastran_error_de_punto_flotante():
data = _data()
data.concepts[0].quantity = Decimal("3")
data.concepts[0].unit_price = Decimal("0.10")
assert data.concepts[0].amount == Decimal("0.30")
# ---------------------------------------------------------------------------------------
# Cadena original y sello
# ---------------------------------------------------------------------------------------
def test_cadena_original_delimitada(csd):
cadena = sealer.build_original_string(_xml(csd))
assert cadena.startswith("||") and cadena.endswith("||")
assert "|4.0|A|1001|" in cadena
def test_cadena_original_no_descarga_nada():
"""Los includes del XSLT tienen que ser locales: libxslt sale a la red si son URLs.
Sin esto, la cadena original —el dato que se firma— vendría de una descarga no verificada
en tiempo de ejecución, y el timbrado dependería de que sat.gob.mx responda.
"""
from pathlib import Path
xslt = Path(sealer._XSLT_CADENA)
contenido = xslt.read_text(encoding="utf-8")
assert 'href="http' not in contenido, "el XSLT conserva includes remotos"
def test_numero_de_certificado_entra_en_la_cadena(csd):
"""Si NoCertificado se rellenara después de firmar, el sello no verificaría."""
numero, _ = sealer.read_certificate(csd[0])
assert numero in sealer.build_original_string(_xml(csd))
def test_certificado_da_numero_de_20_digitos_y_base64(csd):
numero, cert_b64 = sealer.read_certificate(csd[0])
assert numero == CERT_NUMBER
assert len(numero) == 20 and numero.isdigit()
assert base64.b64decode(cert_b64) == csd[0]
def test_sello_verifica_contra_la_llave_publica_del_certificado(csd):
"""La prueba fuerte del sellado: si esto pasa, sellamos como espera el SAT."""
cer_der, key_der = csd
xml = _xml(csd)
cadena = sealer.build_original_string(xml)
sello = sealer.sign(cadena, sealer.load_private_key(key_der, CSD_PASSWORD))
x509.load_der_x509_certificate(cer_der).public_key().verify(
base64.b64decode(sello), cadena.encode("utf-8"), padding.PKCS1v15(), hashes.SHA256()
)
def test_sello_no_verifica_si_la_cadena_cambia(csd):
"""Contraparte de la anterior: un verde que no se ve fallar no vale."""
cer_der, key_der = csd
cadena = sealer.build_original_string(_xml(csd))
sello = sealer.sign(cadena, sealer.load_private_key(key_der, CSD_PASSWORD))
with pytest.raises(Exception):
x509.load_der_x509_certificate(cer_der).public_key().verify(
base64.b64decode(sello),
(cadena + " ").encode("utf-8"),
padding.PKCS1v15(),
hashes.SHA256(),
)
def test_sellar_no_altera_la_cadena_original(csd):
"""El Sello no entra en la cadena: insertarlo no puede cambiarla."""
xml = _xml(csd)
antes = sealer.build_original_string(xml)
firmado = B.apply_seal(xml, "SELLO-DE-PRUEBA")
assert sealer.build_original_string(firmado) == antes
assert etree.fromstring(firmado).get("Sello") == "SELLO-DE-PRUEBA"
def test_contrasena_incorrecta_da_error_claro(csd):
with pytest.raises(sealer.SealingError) as exc:
sealer.load_private_key(csd[1], "incorrecta")
assert "contraseña" in str(exc.value).lower()
def test_sin_contrasena_no_se_intenta_firmar(csd):
with pytest.raises(sealer.SealingError):
sealer.load_private_key(csd[1], "")
def test_sellar_sin_numero_de_certificado_es_error():
xml = B.build_xml(_data(), cert_number="", cert_b64="")
with pytest.raises(B.CfdiBuildError):
B.apply_seal(xml, "SELLO")
# ---------------------------------------------------------------------------------------
# Cliente del PAC — resolución de host
# ---------------------------------------------------------------------------------------
HOST_TEST = "pruebas.comercio-digital.mx"
HOST_PROD = "ws.comercio-digital.mx"
def test_host_se_deriva_del_modo():
assert pac.resolve_host("pruebas", HOST_TEST, HOST_PROD) == HOST_TEST
assert pac.resolve_host("produccion", HOST_TEST, HOST_PROD) == HOST_PROD
@pytest.mark.parametrize("modo", ["", "prod", "PRUEBAS", "producción", None])
def test_modo_invalido_no_cae_a_ningun_host(modo):
"""El legado, con host vacío, caía silenciosamente a pruebas (CFDI.cs:19288)."""
with pytest.raises(pac.PacConfigError):
pac.resolve_host(modo, HOST_TEST, HOST_PROD)
def test_el_host_no_se_puede_inyectar_por_http():
"""`stamp` no acepta host ni URL: sólo el modo. Es la regla 1 de §4.5 del plan."""
import inspect
params = set(inspect.signature(pac.stamp).parameters)
assert "url" not in params
assert params & {"host_test", "host_prod"} == {"host_test", "host_prod"}
# host_test/host_prod son configuración del servidor, no entrada de la petición: el
# endpoint los toma de settings y nunca del cuerpo (ver routes.stamp_invoice).
# ---------------------------------------------------------------------------------------
# Cliente del PAC — casos de error, contra un doble
# ---------------------------------------------------------------------------------------
XML_VALIDO = b"<x>" + b"a" * 300 + b"</x>"
def _stamp(monkeypatch, *, respuesta=None, excepcion=None, **kwargs):
"""Ejecuta pac.stamp con httpx.post sustituido por un doble."""
def falso_post(url, content=None, headers=None, timeout=None):
falso_post.llamadas.append({"url": url, "headers": headers, "content": content})
if excepcion:
raise excepcion
return respuesta
falso_post.llamadas = []
monkeypatch.setattr(httpx, "post", falso_post)
opciones = dict(
mode="pruebas",
user="SCT050708AD1",
password="secreto",
host_test=HOST_TEST,
host_prod=HOST_PROD,
)
opciones.update(kwargs)
return pac.stamp(XML_VALIDO, **opciones), falso_post.llamadas
def _respuesta(status_code=200, headers=None, text="<cfdi/>"):
return httpx.Response(
status_code=status_code,
headers=headers or {},
text=text,
request=httpx.Request("POST", "https://x/timbre4/timbrarV5"),
)
def test_error_701_usuario_invalido(monkeypatch):
res, llamadas = _stamp(monkeypatch, user="corto")
assert res.ok is False and res.code == 701
assert llamadas == [], "no debe salir a la red con el usuario inválido"
def test_error_702_password_vacio(monkeypatch):
res, llamadas = _stamp(monkeypatch, password="")
assert res.ok is False and res.code == 702
assert llamadas == []
def test_error_711_xml_demasiado_corto(monkeypatch):
monkeypatch.setattr(httpx, "post", lambda *a, **k: pytest.fail("no debe llamar al PAC"))
res = pac.stamp(
b"<x/>",
mode="pruebas",
user="SCT050708AD1",
password="x",
host_test=HOST_TEST,
host_prod=HOST_PROD,
)
assert res.ok is False and res.code == 711
def test_error_833_fallo_de_red(monkeypatch):
res, _ = _stamp(monkeypatch, excepcion=httpx.ConnectError("sin ruta al host"))
assert res.ok is False and res.code == 833
def test_error_998_http_distinto_de_200(monkeypatch):
res, _ = _stamp(monkeypatch, respuesta=_respuesta(status_code=500, text="<fault/>"))
# El cuerpo se conserva: es lo que se guarda como XML de respuesta del intento.
assert res.xml == "<fault/>"
assert res.ok is False and res.code == 998
def test_timbrado_correcto_lee_codigo_y_saldo(monkeypatch):
"""Los dos valores que el legado perdía siempre (CFDI.cs:19324-19336)."""
res, llamadas = _stamp(
monkeypatch,
respuesta=_respuesta(
headers={"uuid": "ABC-123", "codigo": "0", "saldo": "4821", "errmsg": ""}
),
)
assert res.ok is True
assert res.uuid == "ABC-123"
assert res.code == 0, "el código del PAC no se está leyendo"
assert res.balance == 4821, "el saldo de folios no se está leyendo"
assert llamadas[0]["url"] == f"https://{HOST_TEST}/timbre4/timbrarV5"
def test_errmsg_con_contenido_es_error(monkeypatch):
res, _ = _stamp(
monkeypatch,
respuesta=_respuesta(headers={"errmsg": "307 CFDI previamente timbrado", "codigo": "307"}),
)
assert res.ok is False
assert res.code == 307
assert "307" in res.error_message
def test_respuesta_sin_uuid_no_es_exito(monkeypatch):
"""200 sin errmsg y sin UUID no es un comprobante timbrado."""
res, _ = _stamp(monkeypatch, respuesta=_respuesta(headers={"errmsg": "", "codigo": "0"}))
assert res.ok is False
assert "UUID" in res.error_message
def test_cabecera_no_numerica_no_rompe_el_timbrado(monkeypatch):
res, _ = _stamp(
monkeypatch,
respuesta=_respuesta(headers={"uuid": "U-1", "codigo": "n/d", "saldo": ""}),
)
assert res.ok is True
assert res.code is None and res.balance is None
def test_cabeceras_y_cuerpo_de_la_peticion(monkeypatch):
res, llamadas = _stamp(
monkeypatch,
respuesta=_respuesta(headers={"uuid": "U-1"}),
email="Avisos@Ejemplo.MX",
)
enviado = llamadas[0]
assert enviado["headers"]["tipo"] == "XML"
assert enviado["headers"]["Content-Type"] == "text/plain"
assert enviado["headers"]["email"] == "avisos@ejemplo.mx" # el PAC lo exige en minúsculas
assert enviado["content"] == XML_VALIDO, "el XML va en crudo, ni base64 ni SOAP"
def test_modo_produccion_apunta_al_host_de_produccion(monkeypatch):
"""Se ejercita SOLO contra el doble: ninguna prueba transmite a producción."""
_, llamadas = _stamp(
monkeypatch, mode="produccion", respuesta=_respuesta(headers={"uuid": "U-1"})
)
assert llamadas[0]["url"] == f"https://{HOST_PROD}/timbre4/timbrarV5"
# ---------------------------------------------------------------------------------------
# Rastro del intento: XML enviado y recibido
# ---------------------------------------------------------------------------------------
class _StampFalso:
"""Lo mínimo que _store_attempt_xml necesita de un InvoiceStamp, sin tocar la BD."""
def __init__(self):
self.tenant_id, self.company_id, self.invoice_id, self.id = 7, 3, 41, 9
self.request_xml_file_key = None
self.response_xml_file_key = None
@pytest.fixture
def subidas(monkeypatch):
"""Captura lo que se sube al almacenamiento en vez de escribir en MinIO."""
from core import storage_s3
hechas: list[tuple[str, bytes]] = []
def falso_put(key, body, content_type=None):
hechas.append((key, body))
monkeypatch.setattr(storage_s3, "put_object_bytes", falso_put)
return hechas
def test_clave_del_xml_del_intento_lleva_el_id_del_intento():
"""Por id de intento y no por UUID: un intento rechazado no tiene UUID."""
key = s3_keys.invoice_stamp_attempt_xml_key(7, 3, 41, 9, "request")
assert key.endswith("fin-invoices/41/stamps/attempts/9-request.xml")
def test_clave_del_xml_del_intento_rechaza_tipos_desconocidos():
with pytest.raises(ValueError):
s3_keys.invoice_stamp_attempt_xml_key(7, 3, 41, 9, "borrador")
def test_se_guardan_los_dos_xml_del_intento(subidas):
stamp = _StampFalso()
service._store_attempt_xml(stamp, b"<enviado/>", "<recibido/>")
assert [cuerpo for _, cuerpo in subidas] == [b"<enviado/>", b"<recibido/>"]
assert stamp.request_xml_file_key.endswith("9-request.xml")
assert stamp.response_xml_file_key.endswith("9-response.xml")
def test_sin_respuesta_del_pac_se_guarda_al_menos_lo_enviado(subidas):
"""Un fallo de red corta antes de que el PAC conteste: el envío sigue siendo el dato útil."""
stamp = _StampFalso()
service._store_attempt_xml(stamp, b"<enviado/>", "")
assert [cuerpo for _, cuerpo in subidas] == [b"<enviado/>"]
assert stamp.request_xml_file_key is not None
assert stamp.response_xml_file_key is None
def test_fallo_del_almacenamiento_no_tumba_el_timbrado(monkeypatch):
"""El rastro es para diagnóstico: perderlo no puede invalidar un timbre que el SAT ya dio
por bueno, ni tapar el error del PAC con uno de almacenamiento."""
from core import storage_s3
def revienta(*_a, **_k):
raise RuntimeError("MinIO no responde")
monkeypatch.setattr(storage_s3, "put_object_bytes", revienta)
stamp = _StampFalso()
service._store_attempt_xml(stamp, b"<enviado/>", "<recibido/>") # no propaga
assert stamp.request_xml_file_key is None
assert stamp.response_xml_file_key is None
Reference in New Issue View Git Blame Copy Permalink