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


def test_registrar_notificacion_guarda_el_destinatario():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    seguimiento.registrar_notificacion(r["codigo_seguimiento"], "enviada", destinatario="a@x.com, b@x.com")
    assert seguimiento.obtener(r["codigo_seguimiento"])["notificacion_destinatario"] == "a@x.com, b@x.com"


def _correo_con_respuesta(email_id, sender, estado="enviada", comprobante=False):
    database_utilities.save_emails([{"id": email_id, "fecha": "2026-10-14 10:00:00", "sender": sender,
                                     "subject": "s", "has_attach": True}])
    r = (seguimiento.registrar_comprobante(email_id, *COMPROBANTE[:2], "F002", email_id) if comprobante
         else seguimiento.registrar_correo(email_id))
    seguimiento.registrar_notificacion(r["codigo_seguimiento"], estado)
    return r


def test_contar_respuestas_solo_cuenta_enviadas_a_correos_sin_comprobante_de_ese_remitente():
    _correo_con_respuesta("e1", "Prov@Empresa.com")                      # cuenta (sin distinguir mayúsculas)
    _correo_con_respuesta("e2", "prov@empresa.com")                      # cuenta
    _correo_con_respuesta("e3", "prov@empresa.com", estado="omitida")    # no: no se envió
    _correo_con_respuesta("e4", "prov@empresa.com", estado="fallida")    # no: no se envió
    _correo_con_respuesta("e5", "otro@empresa.com")                      # no: otro remitente
    _correo_con_respuesta("e6", "prov@empresa.com", comprobante=True)    # no: tiene comprobante

    assert seguimiento.contar_respuestas_correo("prov@empresa.com") == 2
    assert seguimiento.contar_respuestas_correo("nadie@empresa.com") == 0


def test_contar_respuestas_ignora_las_anteriores_a_la_ventana():
    r = _correo_con_respuesta("e1", "prov@empresa.com")
    conn = sqlite3.connect(database_utilities.DB_NAME)
    conn.execute("UPDATE comprobantes SET notificacion_fecha = '2020-01-01 00:00:00' WHERE codigo_seguimiento = ?",
                 (r["codigo_seguimiento"],))
    conn.commit()
    conn.close()

    assert seguimiento.contar_respuestas_correo("prov@empresa.com", horas=24) == 0
    assert seguimiento.contar_respuestas_correo("prov@empresa.com", horas=24 * 365 * 10) == 1


def test_init_db_agrega_la_columna_a_una_tabla_creada_antes():
    conn = sqlite3.connect(database_utilities.DB_NAME)
    conn.executescript("""
        DROP TABLE comprobantes;
        CREATE TABLE comprobantes (
            id INTEGER PRIMARY KEY AUTOINCREMENT, codigo_seguimiento TEXT UNIQUE, email_id TEXT NOT NULL,
            ruc_emisor TEXT, tipo TEXT, serie TEXT, numero TEXT, estado TEXT NOT NULL DEFAULT 'pendiente',
            observaciones TEXT, doc_entry INTEGER, veces_recibido INTEGER NOT NULL DEFAULT 1,
            notificacion_estado TEXT NOT NULL DEFAULT 'pendiente', notificacion_error TEXT, notificacion_fecha TEXT,
            creado_en TEXT NOT NULL, actualizado_en TEXT NOT NULL);
        INSERT INTO comprobantes (codigo_seguimiento, email_id, creado_en, actualizado_en)
            VALUES ('ABR-261001-00001', 'viejo', '2026-10-01 10:00:00', '2026-10-01 10:00:00');
    """)
    conn.commit()
    conn.close()

    database_utilities.init_db()

    assert seguimiento.obtener("ABR-261001-00001")["email_id"] == "viejo"       # el dato se conserva
    seguimiento.registrar_notificacion("ABR-261001-00001", "enviada", destinatario="a@x.com")
    assert seguimiento.obtener("ABR-261001-00001")["notificacion_destinatario"] == "a@x.com"


def test_registrar_intento_cuenta_de_uno_en_uno():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    assert r["intentos"] == 0

    assert seguimiento.registrar_intento(r["codigo_seguimiento"]) == 1
    assert seguimiento.registrar_intento(r["codigo_seguimiento"]) == 2
    assert seguimiento.obtener(r["codigo_seguimiento"])["intentos"] == 2


def test_registrar_intento_de_un_codigo_inexistente():
    with pytest.raises(LookupError):
        seguimiento.registrar_intento("ABR-000000-99999")


def test_en_espera_es_un_estado_de_notificacion_valido():
    r = seguimiento.registrar_comprobante("correo-1", *COMPROBANTE)
    seguimiento.registrar_notificacion(r["codigo_seguimiento"], "en_espera", destinatario="a@x.com")
    assert seguimiento.obtener(r["codigo_seguimiento"])["notificacion_estado"] == "en_espera"


def test_init_db_agrega_la_columna_intentos_a_una_tabla_anterior():
    conn = sqlite3.connect(database_utilities.DB_NAME)
    conn.executescript("""
        DROP TABLE comprobantes;
        CREATE TABLE comprobantes (
            id INTEGER PRIMARY KEY AUTOINCREMENT, codigo_seguimiento TEXT UNIQUE, email_id TEXT NOT NULL,
            ruc_emisor TEXT, tipo TEXT, serie TEXT, numero TEXT, estado TEXT NOT NULL DEFAULT 'pendiente',
            observaciones TEXT, doc_entry INTEGER, veces_recibido INTEGER NOT NULL DEFAULT 1,
            notificacion_estado TEXT NOT NULL DEFAULT 'pendiente', notificacion_error TEXT, notificacion_fecha TEXT,
            notificacion_destinatario TEXT, creado_en TEXT NOT NULL, actualizado_en TEXT NOT NULL);
        INSERT INTO comprobantes (codigo_seguimiento, email_id, creado_en, actualizado_en)
            VALUES ('ABR-261001-00001', 'viejo', '2026-10-01 10:00:00', '2026-10-01 10:00:00');
    """)
    conn.commit()
    conn.close()

    database_utilities.init_db()

    assert seguimiento.obtener("ABR-261001-00001")["intentos"] == 0           # el valor por defecto se aplica a lo existente
    assert seguimiento.registrar_intento("ABR-261001-00001") == 1