descarga_factura_transporte/tests/test_seguimiento.py

import re
import sqlite3

import pytest

import database_utilities
import seguimiento

COMPROBANTE = ("20537929520", "01", "F002", "486")


def test_registrar_crea_codigo_y_estado_inicial():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)

    assert re.fullmatch(r"ABR-\d{6}-\d{5}", r["codigo_seguimiento"])
    assert r["estado"] == "pendiente"
    assert r["notificacion_estado"] == "pendiente"
    assert r["veces_recibido"] == 1
    assert r["reenvio"] is False
    assert (r["ruc_emisor"], r["tipo"], r["serie"], r["numero"]) == COMPROBANTE
    assert r["observaciones"] is None and r["doc_entry"] is None


def test_reprocesar_el_mismo_correo_conserva_todo():
    primero = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    segundo = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)

    assert segundo["codigo_seguimiento"] == primero["codigo_seguimiento"]
    assert segundo["veces_recibido"] == 1
    assert segundo["reenvio"] is False


def test_mismo_comprobante_en_otro_correo_conserva_el_codigo():
    primero = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    segundo = seguimiento.registrar_comprobante("correo-2", *COMPROBANTE)

    assert segundo["codigo_seguimiento"] == primero["codigo_seguimiento"]
    assert segundo["veces_recibido"] == 2
    assert segundo["email_id"] == "correo-2"      # queda el correo más reciente
    assert segundo["reenvio"] is True


def test_el_reenvio_conserva_el_resultado_anterior():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    seguimiento.actualizar_resultado(r["codigo_seguimiento"], "aceptado", doc_entry=2021)

    reenviado = seguimiento.registrar_comprobante("correo-2", *COMPROBANTE)

    assert reenviado["estado"] == "aceptado" and reenviado["doc_entry"] == 2021


def test_comprobantes_distintos_tienen_codigos_distintos():
    a = seguimiento.registrar_comprobante("c1", "20537929520", "01", "F002", "486")
    b = seguimiento.registrar_comprobante("c1", "20537929520", "01", "F002", "487")        # otro número
    c = seguimiento.registrar_comprobante("c1", "20111111111", "01", "F002", "486")        # otro proveedor
    d = seguimiento.registrar_comprobante("c1", "20537929520", "07", "F002", "486")        # otro tipo

    codigos = {x["codigo_seguimiento"] for x in (a, b, c, d)}
    assert len(codigos) == 4


def test_registrar_correo_es_idempotente_por_correo():
    a = seguimiento.registrar_correo("correo-1")
    de_nuevo = seguimiento.registrar_correo("correo-1")
    otro = seguimiento.registrar_correo("correo-2")

    assert de_nuevo["codigo_seguimiento"] == a["codigo_seguimiento"]
    assert otro["codigo_seguimiento"] != a["codigo_seguimiento"]
    assert a["serie"] is None and a["ruc_emisor"] is None


def test_correo_sin_comprobante_y_comprobante_del_mismo_correo_coexisten():
    sin_comprobante = seguimiento.registrar_correo("correo-1")
    con_comprobante = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)

    assert sin_comprobante["codigo_seguimiento"] != con_comprobante["codigo_seguimiento"]
    # y cada uno se sigue recuperando por separado
    assert seguimiento.registrar_correo("correo-1")["codigo_seguimiento"] == sin_comprobante["codigo_seguimiento"]
    assert seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)["codigo_seguimiento"] == con_comprobante["codigo_seguimiento"]


def test_obtener():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    assert seguimiento.obtener(r["codigo_seguimiento"])["numero"] == "486"
    assert seguimiento.obtener("ABR-000000-99999") is None


def test_actualizar_resultado_guarda_estado_observaciones_y_doc_entry():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    codigo = r["codigo_seguimiento"]

    seguimiento.actualizar_resultado(codigo, "rechazado", observaciones=[{"codigo": "precio", "detalle": "ñandú"}])
    rechazado = seguimiento.obtener(codigo)
    assert rechazado["estado"] == "rechazado"
    assert rechazado["observaciones"] == [{"codigo": "precio", "detalle": "ñandú"}]   # JSON de ida y vuelta

    seguimiento.actualizar_resultado(codigo, "aceptado", doc_entry=2021)
    aceptado = seguimiento.obtener(codigo)
    assert (aceptado["estado"], aceptado["doc_entry"], aceptado["observaciones"]) == ("aceptado", 2021, None)


def test_actualizar_resultado_valida_estado_y_codigo():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    with pytest.raises(ValueError):
        seguimiento.actualizar_resultado(r["codigo_seguimiento"], "inventado")
    with pytest.raises(LookupError):
        seguimiento.actualizar_resultado("ABR-000000-99999", "aceptado")


def test_registrar_notificacion():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    codigo = r["codigo_seguimiento"]
    assert r["notificacion_fecha"] is None

    seguimiento.registrar_notificacion(codigo, "fallida", error="Graph 503")
    fallida = seguimiento.obtener(codigo)
    assert (fallida["notificacion_estado"], fallida["notificacion_error"]) == ("fallida", "Graph 503")
    assert fallida["notificacion_fecha"] is not None

    seguimiento.registrar_notificacion(codigo, "enviada")
    assert seguimiento.obtener(codigo)["notificacion_error"] is None

    with pytest.raises(ValueError):
        seguimiento.registrar_notificacion(codigo, "inventado")
    with pytest.raises(LookupError):
        seguimiento.registrar_notificacion("ABR-000000-99999", "enviada")


def test_init_db_es_idempotente_y_conserva_los_datos():
    seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    database_utilities.save_emails([{"id": "correo-1", "fecha": "2026-10-19 10:00:00", "sender": "a@b.c",
                                     "subject": "s", "has_attach": True}])

    database_utilities.init_db()          # segunda vez: no debe borrar nada

    assert database_utilities.get_existing_ids(["correo-1"]) == ["correo-1"]
    assert seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)["veces_recibido"] == 1


def test_la_tabla_emails_existente_no_se_modifica():
    """Una BD anterior (solo con ``emails``) debe actualizarse al esquema nuevo sin perder correos."""
    conn = sqlite3.connect(database_utilities.DB_NAME)
    conn.execute("DROP TABLE comprobantes")
    conn.commit()
    conn.close()

    database_utilities.init_db()

    assert seguimiento.registrar_correo("correo-1")["codigo_seguimiento"].startswith("ABR-")