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