"""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 = _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"" + b"a" * 300 + b"" 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=""): 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"", 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="")) # El cuerpo se conserva: es lo que se guarda como XML de respuesta del intento. assert res.xml == "" 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"", "") assert [cuerpo for _, cuerpo in subidas] == [b"", b""] 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"", "") assert [cuerpo for _, cuerpo in subidas] == [b""] 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"", "") # no propaga assert stamp.request_xml_file_key is None assert stamp.response_xml_file_key is None