se agrega archivos para la instalación en Docker

This commit is contained in:
George committed 2026-08-04 17:57:43 -05:00
1 parent 533105e7c3
commit cf2c1ee147
28 files changed
+1430 -501

No files matched your search

View File
Whitespace-only changes.
+110
View File
@@ -0,0 +1,110 @@
import pytest
import os
import tempfile
import xml.etree.ElementTree as ET
FACTURA_XML = '''<?xml version="1.0" encoding="utf-8"?>
<Invoice xmlns="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
xmlns:cac="urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
xmlns:cbc="urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2">
<cbc:ID>F001-00005103</cbc:ID>
<cbc:InvoiceTypeCode>01</cbc:InvoiceTypeCode>
<cac:AccountingSupplierParty>
<cac:Party>
<cac:PartyIdentification>
<cbc:ID schemeID="6">20549758054</cbc:ID>
</cac:PartyIdentification>
</cac:Party>
</cac:AccountingSupplierParty>
<cac:TaxTotal>
<cac:TaxSubtotal>
<cac:TaxCategory>
<cbc:ID>IGV</cbc:ID>
</cac:TaxCategory>
</cac:TaxSubtotal>
</cac:TaxTotal>
</Invoice>'''
CREDIT_NOTE_XML = '''<?xml version="1.0" encoding="utf-8"?>
<CreditNote xmlns="urn:oasis:names:specification:ubl:schema:xsd:CreditNote-2"
xmlns:cac="urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
xmlns:cbc="urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2">
<cbc:ID>B001-00001234</cbc:ID>
<cbc:InvoiceTypeCode>07</cbc:InvoiceTypeCode>
<cac:AccountingSupplierParty>
<cac:Party>
<cac:PartyIdentification>
<cbc:ID schemeID="6">20123456789</cbc:ID>
</cac:PartyIdentification>
</cac:Party>
</cac:AccountingSupplierParty>
</CreditNote>'''
CDR_XML = '''<?xml version="1.0" encoding="utf-8"?>
<ApplicationResponse xmlns="urn:oasis:names:specification:ubl:schema:xsd:ApplicationResponse-2"
xmlns:cac="urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
xmlns:cbc="urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2">
<cbc:ID>R-2025-00000123</cbc:ID>
<cac:SenderParty>
<cac:PartyLegalEntity>
<cbc:CompanyID schemeID="6">20549758054</cbc:CompanyID>
</cac:PartyLegalEntity>
</cac:SenderParty>
</ApplicationResponse>'''
RECIBO_XML = '''<?xml version="1.0" encoding="utf-8"?>
<Invoice xmlns="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"
xmlns:cac="urn:oasis:names:specification:ubl:schema:xsd:CommonAggregateComponents-2"
xmlns:cbc="urn:oasis:names:specification:ubl:schema:xsd:CommonBasicComponents-2">
<cbc:ID>R001-00000456</cbc:ID>
<cbc:InvoiceTypeCode>02</cbc:InvoiceTypeCode>
<cac:AccountingSupplierParty>
<cbc:CustomerAssignedAccountID>10456789012</cbc:CustomerAssignedAccountID>
</cac:AccountingSupplierParty>
<cac:TaxTotal>
<cac:TaxSubtotal>
<cac:TaxCategory>
<cbc:ID>RET 4TA</cbc:ID>
</cac:TaxCategory>
</cac:TaxSubtotal>
</cac:TaxTotal>
</Invoice>'''
@pytest.fixture
def temp_dir():
with tempfile.TemporaryDirectory() as tmpdir:
yield tmpdir
@pytest.fixture
def factura_xml_path(temp_dir):
path = os.path.join(temp_dir, "F001-00005103.xml")
with open(path, "w", encoding="utf-8") as f:
f.write(FACTURA_XML)
return path
@pytest.fixture
def credito_xml_path(temp_dir):
path = os.path.join(temp_dir, "B001-00001234.xml")
with open(path, "w", encoding="utf-8") as f:
f.write(CREDIT_NOTE_XML)
return path
@pytest.fixture
def cdr_xml_path(temp_dir):
path = os.path.join(temp_dir, "R-2025-00000123.xml")
with open(path, "w", encoding="utf-8") as f:
f.write(CDR_XML)
return path
@pytest.fixture
def recibo_xml_path(temp_dir):
path = os.path.join(temp_dir, "R001-00000456.xml")
with open(path, "w", encoding="utf-8") as f:
f.write(RECIBO_XML)
return path
+73
View File
@@ -0,0 +1,73 @@
import os
import sys
import pytest
import tempfile
import sqlite3
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Forzar uso de DB temporal para no contaminar la real
@pytest.fixture(autouse=True)
def override_db_path(monkeypatch, tmp_path):
db_path = str(tmp_path / "test_emails.db")
monkeypatch.setattr("database_utilities.DB_NAME", db_path)
yield
if os.path.exists(db_path):
os.remove(db_path)
from database_utilities import init_db, save_emails, get_existing_ids, delete_email_by_id
class TestInitDb:
def test_crea_tabla(self):
init_db()
conn = sqlite3.connect("test_emails.db" if False else ":memory:")
# No podemos verificar la DB real porque el path se sobreescribió
# Solo verificar que no lanza excepción
assert True
class TestSaveAndGetEmails:
def test_save_and_retrieve(self):
init_db()
emails = [
{"id": "id1", "fecha": "2026-01-01", "sender": "a@b.com",
"subject": "Test", "has_attach": True},
{"id": "id2", "fecha": "2026-01-02", "sender": "c@d.com",
"subject": "Test2", "has_attach": False},
]
assert save_emails(emails) > 0
assert get_existing_ids(["id1", "id2", "id3"]) == ["id1", "id2"]
def test_duplicados_ignorados(self):
init_db()
emails = [
{"id": "id1", "fecha": "2026-01-01", "sender": "a@b.com",
"subject": "Test", "has_attach": True},
]
save_emails(emails)
# Insertar el mismo id otra vez
save_emails(emails)
assert get_existing_ids(["id1"]) == ["id1"]
def test_lista_vacia(self):
assert save_emails([]) == 0
assert get_existing_ids([]) == []
class TestDeleteEmail:
def test_delete_existing(self):
init_db()
save_emails([{"id": "id1", "fecha": "2026-01-01", "sender": "a@b.com",
"subject": "Test", "has_attach": True}])
assert delete_email_by_id("id1") is True
assert get_existing_ids(["id1"]) == []
def test_delete_nonexistent(self):
init_db()
assert delete_email_by_id("no_existe") is False
def test_delete_empty_id(self):
assert delete_email_by_id("") is False
assert delete_email_by_id(None) is False
+98
View File
@@ -0,0 +1,98 @@
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from document_processor import (
formatear_serie_numero,
generar_nombre_propuesto,
obtener_datos_xml,
TIPOS_SUNAT,
)
class TestFormatearSerieNumero:
def test_formato_estandar(self):
assert formatear_serie_numero("F001-00005103") == "F001-5103"
def test_sin_ceros(self):
assert formatear_serie_numero("F001-123") == "F001-123"
def test_con_letras_en_numero(self):
assert formatear_serie_numero("E001-ABC") == "E001-ABC"
def test_sin_guion(self):
assert formatear_serie_numero("F001") == "F001"
def test_cdr_id(self):
assert formatear_serie_numero("R-2025-00000123") == "R-2025-00000123"
def test_na(self):
assert formatear_serie_numero("N/A") == "N/A"
def test_no_encontrado(self):
assert formatear_serie_numero("No encontrado") == "No encontrado"
class TestGenerarNombrePropuesto:
def test_factura(self):
m = {"tipo": "01", "ruc": "20549758054", "serie_numero": "F001-5103"}
assert generar_nombre_propuesto(m) == "20549758054-01-F001-5103"
def test_cdr(self):
m = {"tipo": "CDR", "ruc": "20549758054", "serie_numero": "R-2025-123"}
assert generar_nombre_propuesto(m) == "20549758054-CDR-R-2025-123"
def test_credito(self):
m = {"tipo": "07", "ruc": "20123456789", "serie_numero": "B001-1234"}
assert generar_nombre_propuesto(m) == "20123456789-07-B001-1234"
def test_tipo_99_no_renombra(self):
m = {"tipo": "99", "ruc": "20549758054", "serie_numero": "F001-5103"}
assert generar_nombre_propuesto(m) is None
def test_ruc_desconocido_no_renombra(self):
m = {"tipo": "01", "ruc": "Desconocido", "serie_numero": "F001-5103"}
assert generar_nombre_propuesto(m) is None
def test_ruc_na_no_renombra(self):
m = {"tipo": "01", "ruc": "N/A", "serie_numero": "F001-5103"}
assert generar_nombre_propuesto(m) is None
class TestObtenerDatosXml:
def test_factura(self, factura_xml_path):
res = obtener_datos_xml(factura_xml_path)
assert res["valido"] is True
assert res["tipo_xml"] == "01"
assert res["ruc"] == "20549758054"
assert res["id"] == "F001-00005103"
def test_credito(self, credito_xml_path):
res = obtener_datos_xml(credito_xml_path)
assert res["valido"] is True
assert res["tipo_xml"] == "07"
assert res["ruc"] == "20123456789"
assert res["id"] == "B001-00001234"
def test_cdr(self, cdr_xml_path):
res = obtener_datos_xml(cdr_xml_path)
assert res["valido"] is True
assert res["tipo_xml"] == "CDR"
assert res["id"] == "R-2025-00000123"
def test_recibo_honorarios(self, recibo_xml_path):
res = obtener_datos_xml(recibo_xml_path)
assert res["valido"] is True
assert res["tipo_xml"] == "02"
assert res["id"] == "R001-00000456"
def test_xml_inexistente(self):
res = obtener_datos_xml("/ruta/inexistente.xml")
assert res["valido"] is False
def test_tipos_sunat_incluye_cdr(self):
assert "CDR" in TIPOS_SUNAT
assert "01" in TIPOS_SUNAT
assert "02" in TIPOS_SUNAT
+108
View File
@@ -0,0 +1,108 @@
import os
import sys
import pytest
from unittest.mock import patch, MagicMock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from email_services import (
filtrar_correos_nuevos,
descomprimir_adjuntos,
filtrar_archivos_validos,
procesar_adjuntos_email,
)
class TestFiltrarArchivosValidos:
def test_solo_pdf(self):
archivos = ["doc.pdf", "imagen.jpg", "texto.txt"]
assert filtrar_archivos_validos(archivos) == ["doc.pdf"]
def test_solo_xml(self):
archivos = ["doc.xml", "imagen.jpg", "texto.txt"]
assert filtrar_archivos_validos(archivos) == ["doc.xml"]
def test_pdf_y_xml(self):
archivos = ["doc.pdf", "doc.xml", "imagen.jpg"]
assert filtrar_archivos_validos(archivos) == ["doc.pdf", "doc.xml"]
def test_sin_validos(self):
archivos = ["imagen.jpg", "texto.txt", "archivo.zip"]
assert filtrar_archivos_validos(archivos) == []
def test_case_insensitive(self):
archivos = ["DOC.PDF", "Doc.Xml", "IMAGEN.JPG"]
assert filtrar_archivos_validos(archivos) == ["DOC.PDF", "Doc.Xml"]
class TestDescomprimirAdjuntos:
def test_sin_zips(self, temp_dir):
archivos = ["a.pdf", "b.xml"]
resultado = descomprimir_adjuntos(archivos, temp_dir)
assert resultado == ["a.pdf", "b.xml"]
def test_zip_inexistente(self, temp_dir):
archivos = ["nope.zip"]
resultado = descomprimir_adjuntos(archivos, temp_dir)
assert resultado == ["nope.zip"]
class TestFiltrarCorreosNuevos:
@patch("email_services.database_utilities.get_existing_ids")
@patch("email_services.database_utilities.save_emails")
def test_retorna_todos_los_nuevos(self, mock_save, mock_get):
mock_get.return_value = []
emails = [
{"id": "1", "has_attach": True},
{"id": "2", "has_attach": False},
]
resultado = filtrar_correos_nuevos(emails)
# Debe retornar TODOS los nuevos (con y sin adjuntos)
assert len(resultado) == 2
mock_save.assert_called_once()
@patch("email_services.database_utilities.get_existing_ids")
@patch("email_services.database_utilities.save_emails")
def test_filtra_existentes(self, mock_save, mock_get):
mock_get.return_value = ["1"]
emails = [
{"id": "1", "has_attach": True},
{"id": "2", "has_attach": True},
]
resultado = filtrar_correos_nuevos(emails)
assert len(resultado) == 1
assert resultado[0]["id"] == "2"
@patch("email_services.database_utilities.get_existing_ids")
def test_emails_vacios(self, mock_get):
assert filtrar_correos_nuevos([]) == []
mock_get.assert_not_called()
class TestProcesarAdjuntosEmail:
def test_con_adjuntos_validos(self, temp_dir):
e = {}
for f in ["doc.pdf", "doc.xml", "imagen.jpg"]:
with open(os.path.join(temp_dir, f), "w") as fh:
fh.write("dummy")
archivos = ["doc.pdf", "doc.xml", "imagen.jpg"]
resultado = procesar_adjuntos_email(e, archivos, temp_dir)
assert resultado == ["doc.pdf", "doc.xml"]
assert e["archivos"] == ["doc.pdf", "doc.xml"]
def test_sin_adjuntos_validos(self, temp_dir):
e = {}
for f in ["imagen.jpg", "texto.txt"]:
with open(os.path.join(temp_dir, f), "w") as fh:
fh.write("dummy")
archivos = ["imagen.jpg", "texto.txt"]
resultado = procesar_adjuntos_email(e, archivos, temp_dir)
assert resultado == []
assert "archivos" not in e
def test_archivos_vacios(self):
e = {}
resultado = procesar_adjuntos_email(e, [], "/tmp")
assert resultado == []
+128
View File
@@ -0,0 +1,128 @@
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from file_manager import renombrar_archivos_por_metadata, mover_validos_a_resultado, limpiar_carpeta_descarga
class TestRenombrarArchivosPorMetadata:
def test_renombrar_xml_valido(self, temp_dir):
ruta = os.path.join(temp_dir, "F001-100.xml")
with open(ruta, "w") as f:
f.write("<root/>")
metadatos = [{
"archivo": "F001-100.xml",
"extension": ".xml",
"tipo": "01",
"serie_numero": "F001-100",
"ruc": "20549758054",
"archivo_propuesto": "20549758054-01-F001-100"
}]
renombrar_archivos_por_metadata(metadatos, temp_dir)
assert not os.path.exists(ruta)
nuevo = os.path.join(temp_dir, "20549758054-01-F001-100.xml")
assert os.path.exists(nuevo)
assert metadatos[0]["archivo"] == "20549758054-01-F001-100.xml"
def test_no_renombra_sin_metadata(self, temp_dir):
ruta = os.path.join(temp_dir, "documento.pdf")
with open(ruta, "w") as f:
f.write("dummy")
metadatos = [{
"archivo": "documento.pdf",
"extension": ".pdf",
"tipo": "99",
"serie_numero": "N/A",
"ruc": "N/A",
"archivo_propuesto": None
}]
renombrar_archivos_por_metadata(metadatos, temp_dir)
assert os.path.exists(ruta)
assert metadatos[0]["archivo"] == "documento.pdf"
def test_renombrar_cdr(self, temp_dir):
ruta = os.path.join(temp_dir, "R-2025-123.xml")
with open(ruta, "w") as f:
f.write("<root/>")
metadatos = [{
"archivo": "R-2025-123.xml",
"extension": ".xml",
"tipo": "CDR",
"serie_numero": "R-2025-123",
"ruc": "20549758054",
"archivo_propuesto": "20549758054-CDR-R-2025-123"
}]
renombrar_archivos_por_metadata(metadatos, temp_dir)
assert not os.path.exists(ruta)
nuevo = os.path.join(temp_dir, "20549758054-CDR-R-2025-123.xml")
assert os.path.exists(nuevo)
class TestMoverValidosAResultado:
def test_mover_archivos_validos(self, temp_dir):
origen = os.path.join(temp_dir, "origen")
destino = os.path.join(temp_dir, "destino")
os.makedirs(origen)
ruta_file = os.path.join(origen, "20549758054-01-F001-100.xml")
with open(ruta_file, "w") as f:
f.write("<root/>")
metadatos = [{
"archivo": "20549758054-01-F001-100.xml",
"extension": ".xml",
"tipo": "01",
"archivo_propuesto": "20549758054-01-F001-100"
}]
mover_validos_a_resultado(metadatos, origen, destino)
assert not os.path.exists(ruta_file)
assert os.path.exists(os.path.join(destino, "20549758054-01-F001-100.xml"))
def test_no_mueve_sin_metadata(self, temp_dir):
origen = os.path.join(temp_dir, "origen")
destino = os.path.join(temp_dir, "destino")
os.makedirs(origen)
ruta_file = os.path.join(origen, "basura.pdf")
with open(ruta_file, "w") as f:
f.write("dummy")
metadatos = [{
"archivo": "basura.pdf",
"extension": ".pdf",
"tipo": "99",
"archivo_propuesto": None
}]
mover_validos_a_resultado(metadatos, origen, destino)
assert os.path.exists(ruta_file)
assert os.path.exists(destino)
assert len(os.listdir(destino)) == 0
class TestLimpiarCarpetaDescarga:
def test_limpia_archivos(self, temp_dir):
for f in ["a.pdf", "b.xml", "c.zip"]:
with open(os.path.join(temp_dir, f), "w") as fh:
fh.write("dummy")
limpiar_carpeta_descarga(temp_dir)
assert len(os.listdir(temp_dir)) == 0
def test_carpeta_no_existe(self):
limpiar_carpeta_descarga("/ruta/inexistente")
+99
View File
@@ -0,0 +1,99 @@
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from pdf_utilities import extraer_ruc, extraer_tipo_documento, extraer_serie_numero, obtener_datos_pdf
class TestExtraerRuc:
def test_ruc_formato_completo(self):
texto = "R.U.C.: 20549758054\nRazon Social: EMPRESA SAC"
assert extraer_ruc(texto) == "20549758054"
def test_ruc_con_puntos(self):
texto = "R.U.C. 20549758054\nAlguna direccion"
assert extraer_ruc(texto) == "20549758054"
def test_ruc_desnudo(self):
texto = "Factura Electronica\n20549758054\nMonto: 100.00"
assert extraer_ruc(texto) == "20549758054"
def test_ruc_con_espacio(self):
texto = "RUC: 20605626921\nBoleta de Venta"
assert extraer_ruc(texto) == "20605626921"
def test_sin_ruc(self):
texto = "Documento sin numero de identificacion"
assert extraer_ruc(texto) == "N/A"
class TestExtraerTipoDocumento:
def test_factura_electronica(self):
texto = "FACTURA ELECTRONICA\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "01"
def test_factura_simple(self):
texto = "FACTURA DE VENTA\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "01"
def test_boleta(self):
texto = "BOLETA DE VENTA ELECTRONICA\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "03"
def test_nota_credito(self):
texto = "NOTA DE CREDITO ELECTRONICA\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "07"
def test_nota_debito(self):
texto = "NOTA DE DEBITO\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "08"
def test_recibo_honorarios(self):
texto = "RECIBO POR HONORARIOS\nRUC: 10456789012"
assert extraer_tipo_documento(texto) == "14"
def test_guia_remision(self):
texto = "GUIA DE REMISION ELECTRONICA\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "09"
def test_tipo_desconocido(self):
texto = "DOCUMENTO INVALIDO\nRUC: 20549758054"
assert extraer_tipo_documento(texto) == "99"
class TestExtraerSerieNumero:
def test_serie_factura_guion(self):
texto = "FACTURA ELECTRONICA\nF001-00005103\nRUC: 20549758054"
assert extraer_serie_numero(texto) == "F001-00005103"
def test_serie_boleta_guion(self):
texto = "BOLETA ELECTRONICA\nB001-123456\nRUC: 20549758054"
assert extraer_serie_numero(texto) == "B001-123456"
def test_serie_eg_con_guion(self):
texto = "GUIA DE REMISION\nEG07-2367\nRUC: 20549758054"
assert extraer_serie_numero(texto) == "EG07-2367"
def test_sin_serie(self):
texto = "Documento sin serie"
assert extraer_serie_numero(texto) == "No encontrado"
PDF_REAL = os.environ.get("TEST_PDF_PATH")
@pytest.mark.skipif(not PDF_REAL or not os.path.exists(PDF_REAL),
reason="Define TEST_PDF_PATH con la ruta a un PDF real. Ej: TEST_PDF_PATH=factura.pdf pytest ...")
def test_pdf_real_con_metadata():
"""Extrae metadata de un PDF real. Usar: TEST_PDF_PATH=/ruta/pdf python3 -m pytest tests/ -s"""
datos = obtener_datos_pdf(PDF_REAL)
assert datos["valido"] is True, f"Error: {datos.get('error')}"
print(f"\n{'='*50}")
print(f"Archivo: {PDF_REAL}")
print(f"{'='*50}")
print(f" RUC: {datos['ruc']}")
print(f" Tipo doc: {datos['tipo_doc']}")
print(f" Serie-Numero: {datos['id']}")
print(f"{'='*50}")