447 lines
18 KiB
Python
447 lines
18 KiB
Python
"""
|
|
Tests para generate_report_document (T2026-04-001).
|
|
|
|
Ejecución:
|
|
python manage.py test api.reports.tests
|
|
python manage.py test api.reports.tests.TestEstadoHelper
|
|
"""
|
|
import io
|
|
import uuid
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import openpyxl
|
|
from django.contrib.auth import get_user_model
|
|
from django.db.models import Q
|
|
from django.test import TestCase
|
|
|
|
from api.customs.models import Cove, EDocument, Importador, Partida, Pedimento
|
|
from api.licence.models import Licencia
|
|
from api.organization.models import Organizacion
|
|
from api.reports.models import ReportDocument
|
|
from api.reports.tasks.report_document import (
|
|
_apply_user_rfc_filter,
|
|
_estado,
|
|
generate_report_document,
|
|
)
|
|
|
|
User = get_user_model()
|
|
|
|
FAKE_PATH = 'reports/test/reporte.xlsx'
|
|
|
|
# ── fixtures ──────────────────────────────────────────────────────────────────
|
|
|
|
def _licencia(nombre='Plan Test'):
|
|
return Licencia.objects.create(nombre=nombre, almacenamiento=10)
|
|
|
|
|
|
def _org(nombre='Org Test'):
|
|
lic = _licencia(f'Lic {nombre}')
|
|
return Organizacion.objects.create(nombre=nombre, is_active=True, is_verified=True, licencia=lic)
|
|
|
|
|
|
def _user(org, username='tuser', rfcs=None):
|
|
u = User.objects.create_user(username=username, password='pass', organizacion=org)
|
|
if rfcs:
|
|
u.rfc.set(rfcs)
|
|
return u
|
|
|
|
|
|
def _imp(org, rfc='RFC000000001', nombre='Importador Test'):
|
|
return Importador.objects.create(rfc=rfc, nombre=nombre, organizacion=org)
|
|
|
|
|
|
def _ped(org, imp=None, num='0000001'):
|
|
return Pedimento.objects.create(
|
|
pedimento=num,
|
|
pedimento_app=f'25-160-3910-{num}',
|
|
organizacion=org,
|
|
contribuyente=imp,
|
|
aduana='160',
|
|
patente='3910',
|
|
regimen='ITE',
|
|
clave_pedimento='A1',
|
|
)
|
|
|
|
|
|
def _reporte(user, org_id, extra=None):
|
|
filtros = {'organizacion_id': str(org_id)}
|
|
if extra:
|
|
filtros.update(extra)
|
|
return ReportDocument.objects.create(
|
|
user=user, filters=filtros, status='pending', report_type='cumplimiento'
|
|
)
|
|
|
|
|
|
def _excel_desde_mock(mock_save):
|
|
"""Parsea el workbook que recibió storage_service.save_report."""
|
|
uf = mock_save.call_args[1]['file']
|
|
return openpyxl.load_workbook(io.BytesIO(uf.read()))
|
|
|
|
|
|
def _docs_col(ws):
|
|
"""Devuelve {documento: estatus} leyendo columnas 9 y 10 del worksheet."""
|
|
return {
|
|
ws.cell(row=r, column=9).value: ws.cell(row=r, column=10).value
|
|
for r in range(1, ws.max_row + 1)
|
|
if ws.cell(row=r, column=9).value
|
|
}
|
|
|
|
|
|
def _col1_values(ws):
|
|
"""Devuelve todos los valores no vacíos de la columna 1."""
|
|
return [
|
|
str(ws.cell(row=r, column=1).value)
|
|
for r in range(1, ws.max_row + 1)
|
|
if ws.cell(row=r, column=1).value
|
|
]
|
|
|
|
|
|
# ── 1. Helpers ────────────────────────────────────────────────────────────────
|
|
|
|
class TestEstadoHelper(TestCase):
|
|
def test_true_retorna_recuperado(self):
|
|
self.assertEqual(_estado(True), 'RECUPERADO')
|
|
|
|
def test_false_retorna_pendiente(self):
|
|
self.assertEqual(_estado(False), 'PENDIENTE')
|
|
|
|
|
|
# ── 2. Filtro de RFC por usuario ──────────────────────────────────────────────
|
|
|
|
class TestApplyUserRfcFilter(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.org = _org()
|
|
cls.imp1 = _imp(cls.org, rfc='RFC000000001')
|
|
cls.imp2 = _imp(cls.org, rfc='RFC000000002')
|
|
|
|
def test_sin_rfcs_asignados_sin_filtro_retorna_q_vacio(self):
|
|
user = _user(self.org, username='u_admin')
|
|
q = _apply_user_rfc_filter(Q(), user, None)
|
|
self.assertEqual(str(q), str(Q()))
|
|
|
|
def test_sin_rfcs_asignados_con_filtro_explicito_aplica_filtro(self):
|
|
user = _user(self.org, username='u_admin2')
|
|
q = _apply_user_rfc_filter(Q(), user, 'RFC000000001')
|
|
self.assertIn('RFC000000001', str(q))
|
|
|
|
def test_con_rfcs_sin_filtro_restringe_a_sus_importadores(self):
|
|
user = _user(self.org, username='u_imp1', rfcs=[self.imp1])
|
|
q = _apply_user_rfc_filter(Q(), user, None)
|
|
self.assertIn('contribuyente', str(q))
|
|
|
|
def test_con_rfcs_pide_el_suyo_se_filtra_por_ese_rfc(self):
|
|
user = _user(self.org, username='u_imp2', rfcs=[self.imp1])
|
|
q = _apply_user_rfc_filter(Q(), user, 'RFC000000001')
|
|
self.assertIn('RFC000000001', str(q))
|
|
|
|
def test_con_rfcs_pide_ajeno_se_usa_el_suyo_no_el_solicitado(self):
|
|
user = _user(self.org, username='u_imp3', rfcs=[self.imp1])
|
|
q = _apply_user_rfc_filter(Q(), user, 'RFC000000002')
|
|
self.assertNotIn('RFC000000002', str(q))
|
|
self.assertIn('contribuyente', str(q))
|
|
|
|
|
|
# ── 3. Tarea completa ─────────────────────────────────────────────────────────
|
|
# Todos los tests en esta clase mockean Redis (publish_task_event) y MinIO
|
|
# (storage_service.save_report) para no depender de infraestructura externa.
|
|
|
|
@patch('api.reports.tasks.report_document.publish_task_event')
|
|
@patch('api.reports.tasks.report_document.storage_service.save_report',
|
|
return_value=FAKE_PATH)
|
|
class TestGenerateReportDocument(TestCase):
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
cls.org = _org('Org Reporte')
|
|
cls.imp = _imp(cls.org, rfc='MTK8610143000', nombre='Servicios TETAKAWI')
|
|
cls.user = _user(cls.org, username='rep_user')
|
|
|
|
def _run(self, report):
|
|
generate_report_document.apply(args=[str(report.id)])
|
|
report.refresh_from_db()
|
|
|
|
# ── 3.1 Sin pedimentos ────────────────────────────────────────────────────
|
|
|
|
def test_sin_pedimentos_genera_excel_vacio_y_status_ready(self, mock_save, mock_pub):
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
self.assertEqual(report.status, 'ready')
|
|
self.assertEqual(report.file, FAKE_PATH)
|
|
mock_save.assert_called_once()
|
|
|
|
# El workbook no debe tener datos de RFCs
|
|
wb = _excel_desde_mock(mock_save)
|
|
ws = wb.active
|
|
col1 = _col1_values(ws)
|
|
self.assertFalse(col1, 'Excel vacío no debe tener contenido en col 1')
|
|
|
|
# ── 3.2 RFC aparece en encabezado ─────────────────────────────────────────
|
|
|
|
def test_rfc_del_importador_aparece_en_excel(self, mock_save, mock_pub):
|
|
_ped(self.org, self.imp, '1000001')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
self.assertEqual(report.status, 'ready')
|
|
wb = _excel_desde_mock(mock_save)
|
|
ws = wb.active
|
|
col1 = ' '.join(_col1_values(ws))
|
|
self.assertIn('MTK8610143000', col1)
|
|
|
|
# ── 3.3 PEDIMENTO COMPLETO ────────────────────────────────────────────────
|
|
|
|
def test_pedimento_completo_recuperado_cuando_existe_expediente(self, mock_save, mock_pub):
|
|
ped = _ped(self.org, self.imp, '1000002')
|
|
ped.existe_expediente = True
|
|
ped.save(update_fields=['existe_expediente'])
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('PEDIMENTO COMPLETO'), 'RECUPERADO')
|
|
|
|
def test_pedimento_completo_pendiente_cuando_no_tiene_expediente(self, mock_save, mock_pub):
|
|
_ped(self.org, self.imp, '1000003') # existe_expediente=False por default
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('PEDIMENTO COMPLETO'), 'PENDIENTE')
|
|
|
|
# ── 3.4 Partidas ──────────────────────────────────────────────────────────
|
|
|
|
def test_partidas_con_estado_correcto(self, mock_save, mock_pub):
|
|
ped = _ped(self.org, self.imp, '1000004')
|
|
Partida.objects.create(
|
|
pedimento=ped, organizacion=self.org, numero_partida=1, descargado=True
|
|
)
|
|
Partida.objects.create(
|
|
pedimento=ped, organizacion=self.org, numero_partida=2, descargado=False
|
|
)
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('PARTIDA1'), 'RECUPERADO')
|
|
self.assertEqual(docs.get('PARTIDA2'), 'PENDIENTE')
|
|
|
|
# ── 3.5 COVEs y acuses ────────────────────────────────────────────────────
|
|
|
|
def test_cove_y_acuse_con_estados_distintos(self, mock_save, mock_pub):
|
|
ped = _ped(self.org, self.imp, '1000005')
|
|
Cove.objects.create(
|
|
pedimento=ped, organizacion=self.org,
|
|
numero_cove='654001',
|
|
cove_descargado=True,
|
|
acuse_cove_descargado=False,
|
|
)
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('COVE654001'), 'RECUPERADO')
|
|
self.assertEqual(docs.get('ACUSE COVE654001'), 'PENDIENTE')
|
|
|
|
# ── 3.6 EDocumentos y acuses ──────────────────────────────────────────────
|
|
|
|
def test_edocumento_y_acuse_con_estados_distintos(self, mock_save, mock_pub):
|
|
ped = _ped(self.org, self.imp, '1000006')
|
|
EDocument.objects.create(
|
|
pedimento=ped, organizacion=self.org,
|
|
numero_edocument='EDOC001',
|
|
edocument_descargado=False,
|
|
acuse_descargado=True,
|
|
)
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('EDOCUMENTOEDOC001'), 'PENDIENTE')
|
|
self.assertEqual(docs.get('ACUSE EDOCUMENTOEDOC001'), 'RECUPERADO')
|
|
|
|
# ── 3.7 Remesa ────────────────────────────────────────────────────────────
|
|
|
|
def test_remesa_recuperada_cuando_document_tipo_15_existe(self, mock_save, mock_pub):
|
|
"""Pedimento.remesas=True y el query de Document devuelve el pedimento_id."""
|
|
ped = Pedimento.objects.create(
|
|
pedimento='1000007', pedimento_app='25-160-3910-1000007',
|
|
organizacion=self.org, contribuyente=self.imp,
|
|
aduana='160', patente='3910', remesas=True,
|
|
)
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
# Patch solo el query de Document dentro del task
|
|
with patch('api.reports.tasks.report_document.Document') as MockDoc:
|
|
mock_qs = MagicMock()
|
|
mock_qs.values_list.return_value = [ped.id]
|
|
MockDoc.objects.filter.return_value = mock_qs
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('REMESA'), 'RECUPERADO')
|
|
|
|
def test_remesa_pendiente_cuando_no_hay_document(self, mock_save, mock_pub):
|
|
"""Pedimento.remesas=True pero el query de Document devuelve lista vacía."""
|
|
Pedimento.objects.create(
|
|
pedimento='1000008', pedimento_app='25-160-3910-1000008',
|
|
organizacion=self.org, contribuyente=self.imp,
|
|
aduana='160', patente='3910', remesas=True,
|
|
)
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
with patch('api.reports.tasks.report_document.Document') as MockDoc:
|
|
mock_qs = MagicMock()
|
|
mock_qs.values_list.return_value = []
|
|
MockDoc.objects.filter.return_value = mock_qs
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertEqual(docs.get('REMESA'), 'PENDIENTE')
|
|
|
|
def test_sin_remesa_no_aparece_fila_remesa(self, mock_save, mock_pub):
|
|
"""Pedimento.remesas=False → no debe aparecer fila REMESA."""
|
|
_ped(self.org, self.imp, '1000009') # remesas=False por default
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
docs = _docs_col(_excel_desde_mock(mock_save).active)
|
|
self.assertNotIn('REMESA', docs)
|
|
|
|
# ── 3.8 Múltiples RFCs ───────────────────────────────────────────────────
|
|
|
|
def test_multiples_rfcs_generan_secciones_separadas(self, mock_save, mock_pub):
|
|
imp2 = _imp(self.org, rfc='TEC140624802', nombre='TEC Importaciones')
|
|
_ped(self.org, self.imp, '1000010')
|
|
_ped(self.org, imp2, '1000011')
|
|
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
self.assertEqual(report.status, 'ready')
|
|
contenido = ' '.join(_col1_values(_excel_desde_mock(mock_save).active))
|
|
self.assertIn('MTK8610143000', contenido)
|
|
self.assertIn('TEC140624802', contenido)
|
|
|
|
# ── 3.9 Restricción por RFC de usuario ───────────────────────────────────
|
|
|
|
def test_importador_solo_ve_sus_pedimentos(self, mock_save, mock_pub):
|
|
imp2 = _imp(self.org, rfc='XYZ999999999', nombre='Externo')
|
|
_ped(self.org, self.imp, '1000012')
|
|
_ped(self.org, imp2, '1000013')
|
|
|
|
user_restr = _user(self.org, username='u_restr', rfcs=[self.imp])
|
|
report = _reporte(user_restr, self.org.id)
|
|
self._run(report)
|
|
|
|
self.assertEqual(report.status, 'ready')
|
|
contenido = ' '.join(_col1_values(_excel_desde_mock(mock_save).active))
|
|
self.assertIn('MTK8610143000', contenido)
|
|
self.assertNotIn('XYZ999999999', contenido)
|
|
|
|
# ── 3.10 Formato del archivo ──────────────────────────────────────────────
|
|
|
|
def test_archivo_descargado_es_xlsx_valido(self, mock_save, mock_pub):
|
|
_ped(self.org, self.imp, '1000014')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
uf = mock_save.call_args[1]['file']
|
|
self.assertTrue(uf.name.endswith('.xlsx'), f'Esperado .xlsx, recibido: {uf.name}')
|
|
try:
|
|
wb = openpyxl.load_workbook(io.BytesIO(uf.read()))
|
|
self.assertIsNotNone(wb)
|
|
except Exception as exc:
|
|
self.fail(f'Excel no es válido: {exc}')
|
|
|
|
def test_cabeceras_de_columna_presentes(self, mock_save, mock_pub):
|
|
_ped(self.org, self.imp, '1000015')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
ws = _excel_desde_mock(mock_save).active
|
|
cabeceras = None
|
|
for r in range(1, ws.max_row + 1):
|
|
if ws.cell(row=r, column=1).value == 'Año':
|
|
cabeceras = [ws.cell(row=r, column=c).value for c in range(1, 11)]
|
|
break
|
|
|
|
self.assertIsNotNone(cabeceras, 'No se encontró la fila de cabeceras')
|
|
for col in ('Año', 'Aduana', 'Patente', 'Pedimento', 'Documento', 'Estatus'):
|
|
self.assertIn(col, cabeceras, f'Cabecera "{col}" no encontrada')
|
|
|
|
# ── 3.11 Progreso en Redis ────────────────────────────────────────────────
|
|
|
|
def test_se_publican_eventos_de_progreso(self, mock_save, mock_pub):
|
|
_ped(self.org, self.imp, '1000016')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
self.assertGreaterEqual(mock_pub.call_count, 4, 'Se esperan mínimo 4 eventos')
|
|
|
|
def test_ultimo_evento_es_completed_con_100(self, mock_save, mock_pub):
|
|
_ped(self.org, self.imp, '1000017')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
ultimo = mock_pub.call_args_list[-1]
|
|
self.assertEqual(ultimo[0][1], 'completed')
|
|
self.assertEqual(ultimo[1].get('progress'), 100)
|
|
|
|
# ── 3.12 Manejo de errores ────────────────────────────────────────────────
|
|
|
|
def test_storage_none_deja_status_error(self, mock_save, mock_pub):
|
|
"""storage_service.save_report retorna None → report queda en error."""
|
|
mock_save.return_value = None
|
|
_ped(self.org, self.imp, '1000018')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
self.assertEqual(report.status, 'error')
|
|
self.assertIn('almacenamiento', report.error_message)
|
|
|
|
def test_storage_none_publica_evento_failed(self, mock_save, mock_pub):
|
|
mock_save.return_value = None
|
|
_ped(self.org, self.imp, '1000019')
|
|
report = _reporte(self.user, self.org.id)
|
|
self._run(report)
|
|
|
|
statuses = [c[0][1] for c in mock_pub.call_args_list]
|
|
self.assertIn('failed', statuses)
|
|
self.assertNotIn('completed', statuses)
|
|
|
|
def test_excepcion_guarda_traceback_en_error_message(self, mock_save, mock_pub):
|
|
"""Una excepción inesperada debe incluir traceback en error_message."""
|
|
mock_save.side_effect = RuntimeError('Fallo simulado de MinIO')
|
|
_ped(self.org, self.imp, '1000020')
|
|
report = _reporte(self.user, self.org.id)
|
|
|
|
try:
|
|
generate_report_document.apply(args=[str(report.id)])
|
|
except RuntimeError:
|
|
pass # apply() re-raise la excepción
|
|
|
|
report.refresh_from_db()
|
|
self.assertEqual(report.status, 'error')
|
|
self.assertIn('Fallo simulado de MinIO', report.error_message)
|
|
self.assertIn('Traceback', report.error_message)
|
|
|
|
def test_excepcion_publica_evento_failed(self, mock_save, mock_pub):
|
|
mock_save.side_effect = RuntimeError('Error MinIO')
|
|
_ped(self.org, self.imp, '1000021')
|
|
report = _reporte(self.user, self.org.id)
|
|
|
|
try:
|
|
generate_report_document.apply(args=[str(report.id)])
|
|
except RuntimeError:
|
|
pass
|
|
|
|
statuses = [c[0][1] for c in mock_pub.call_args_list]
|
|
self.assertIn('failed', statuses)
|