from django.urls import reverse from rest_framework.test import APITestCase, APIClient from rest_framework import status from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from unittest.mock import patch from io import BytesIO import zipfile from api.organization.models import Organizacion from api.licence.models import Licencia from .models import Pedimento, TipoOperacion, ProcesamientoPedimento, EDocument User = get_user_model() class CustomsViewsTests(APITestCase): def setUp(self): self.org = Organizacion.objects.create(nombre="OrgTest", is_active=True, is_verified=True) self.org2 = Organizacion.objects.create(nombre="OrgTest2", is_active=True, is_verified=True) self.admin = User.objects.create_user(username="admin", password="adminpass", organizacion=self.org) self.admin.groups.create(name="admin") self.superuser = User.objects.create_superuser(username="superuser", password="superpass") self.importador = User.objects.create_user(username="importador", password="importpass", organizacion=self.org2, is_importador=True, rfc="RFC123456789") self.importador.groups.create(name="importador") self.client = APIClient() def test_admin_sees_only_own_pedimentos(self): from .models import Pedimento p1 = Pedimento.objects.create(pedimento="P1", organizacion=self.org) p2 = Pedimento.objects.create(pedimento="P2", organizacion=self.org2) self.client.force_authenticate(user=self.admin) url = reverse('Pedimento-list') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) pedimentos = [p['pedimento'] for p in response.data] self.assertIn("P1", pedimentos) self.assertNotIn("P2", pedimentos) def test_superuser_sees_all_pedimentos(self): from .models import Pedimento p1 = Pedimento.objects.create(pedimento="P1", organizacion=self.org) p2 = Pedimento.objects.create(pedimento="P2", organizacion=self.org2) self.client.force_authenticate(user=self.superuser) url = reverse('Pedimento-list') response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) pedimentos = [p['pedimento'] for p in response.data] self.assertIn("P1", pedimentos) self.assertIn("P2", pedimentos) def test_importador_cannot_create_pedimento(self): self.client.force_authenticate(user=self.importador) url = reverse('Pedimento-list') data = { "pedimento": "P3", "patente": "1234", "aduana": "001", "regimen": "A1", "clave_pedimento": "A1", "contribuyente": "ImportadorTest" } response = self.client.post(url, data) self.assertNotIn(response.status_code, [status.HTTP_201_CREATED, status.HTTP_200_OK]) def test_list_tipos_operacion(self): url = reverse('TipoOperacion-list') self.client.force_authenticate(user=self.admin) response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) def test_list_procesamientos(self): url = reverse('ProcesamientoPedimento-list') self.client.force_authenticate(user=self.admin) response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) def test_list_edocuments(self): url = reverse('EDocument-list') self.client.force_authenticate(user=self.admin) response = self.client.get(url) self.assertEqual(response.status_code, status.HTTP_200_OK) # --------------------------------------------------------------------------- # Tests de integración para bulk-create (ViewSetPedimento.bulk_create) # Verifica que al re-cargar un pedimento existente sus documentos se actualicen # --------------------------------------------------------------------------- class BulkCreateDocumentReplaceTests(APITestCase): """Verifica que bulk-create actualiza los documentos de pedimentos existentes en vez de ignorarlos, y que no quedan archivos residuales en el storage.""" PEDIMENTO_APP = "24-01-3420-1234567" def setUp(self): self.licencia = Licencia.objects.create(nombre="Lic100GB", almacenamiento=100) self.org = Organizacion.objects.create( nombre="OrgBulkCreate", licencia=self.licencia, is_active=True, is_verified=True, ) self.user = User.objects.create_user( username="bulkcreateuser", password="pass", organizacion=self.org ) self.pedimento = Pedimento.objects.create( organizacion=self.org, pedimento="1234567", pedimento_app=self.PEDIMENTO_APP, ) from api.record.models import DocumentType, Fuente self.doc_type = DocumentType.objects.get_or_create(nombre="Pedimento")[0] # bulk_create usa fuente_id=4 hardcodeado; debe existir en la DB de test Fuente.objects.get_or_create(id=4, defaults={"nombre": "Bulk Create"}) self.url = reverse("Pedimento-bulk-create") self.client.force_authenticate(user=self.user) def _make_zip(self, files_dict): """Crea un ZIP en memoria. files_dict = {nombre_archivo: contenido_bytes}""" buf = BytesIO() with zipfile.ZipFile(buf, "w") as zf: for name, content in files_dict.items(): zf.writestr(name, content) buf.seek(0) return SimpleUploadedFile( f"{self.PEDIMENTO_APP}.zip", buf.read(), content_type="application/zip" ) def _post_zip(self, files_dict): return self.client.post( self.url, {"contribuyente": "XAXX010101000", "archivos": [self._make_zip(files_dict)]}, format="multipart", ) @patch("api.customs.views.storage_service") def test_existing_pedimento_not_duplicated(self, mock_st): """Re-subir un pedimento existente NO debe crear un segundo Pedimento.""" mock_st.save_document_from_path.return_value = "org_1/documents/ped/informe_a1b2c3d4.pdf" self._post_zip({"informe.pdf": b"contenido"}) self.assertEqual( Pedimento.objects.filter( organizacion=self.org, pedimento_app=self.PEDIMENTO_APP ).count(), 1, ) @patch("api.customs.views.storage_service") def test_existing_pedimento_document_replaced_not_duplicated(self, mock_st): """Documento existente con el mismo nombre base se reemplaza, no se duplica.""" from api.record.models import Document old_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_a1b2c3d4.pdf" old_doc = Document.objects.create( organizacion=self.org, pedimento=self.pedimento, document_type=self.doc_type, archivo=old_path, size=500, extension="pdf", ) new_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_b5c6d7e8.pdf" mock_st.save_document_from_path.return_value = new_path mock_st.delete_file.return_value = True self._post_zip({"informe.pdf": b"contenido actualizado"}) docs = Document.objects.filter(pedimento=self.pedimento) # Sin duplicados self.assertEqual(docs.count(), 1) # Mismo registro self.assertEqual(docs.first().id, old_doc.id) # Archivo actualizado old_doc.refresh_from_db() self.assertEqual(old_doc.archivo.name, new_path) @patch("api.customs.views.storage_service") def test_existing_pedimento_stale_file_deleted_from_storage(self, mock_st): """Al reemplazar un documento, el archivo viejo debe eliminarse del storage.""" from api.record.models import Document old_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_a1b2c3d4.pdf" Document.objects.create( organizacion=self.org, pedimento=self.pedimento, document_type=self.doc_type, archivo=old_path, size=500, extension="pdf", ) mock_st.save_document_from_path.return_value = f"org_1/documents/{self.PEDIMENTO_APP}/informe_b5c6d7e8.pdf" mock_st.delete_file.return_value = True self._post_zip({"informe.pdf": b"contenido"}) # delete_file debe haberse llamado con la ruta del archivo viejo mock_st.delete_file.assert_called() called_arg = str(mock_st.delete_file.call_args[0][0]) self.assertIn("informe_a1b2c3d4", called_arg) @patch("api.customs.views.storage_service") def test_existing_pedimento_new_file_added(self, mock_st): """Archivo nuevo en el ZIP se añade al pedimento existente.""" from api.record.models import Document mock_st.save_document_from_path.return_value = "org_1/documents/ped/nuevo_b5c6d7e8.pdf" self._post_zip({"nuevo_documento.pdf": b"contenido nuevo"}) self.assertGreaterEqual( Document.objects.filter(pedimento=self.pedimento).count(), 1 ) @patch("api.customs.views.storage_service") def test_already_existing_count_in_response(self, mock_st): """La respuesta debe indicar que el pedimento ya existía (already_existing_count >= 1).""" mock_st.save_document_from_path.return_value = "org_1/documents/ped/f_a1b2c3d4.pdf" response = self._post_zip({"archivo.pdf": b"contenido"}) self.assertIn(response.status_code, [status.HTTP_200_OK, status.HTTP_207_MULTI_STATUS, status.HTTP_201_CREATED]) data = response.json() self.assertGreaterEqual(data.get("already_existing_count", 0), 1) # --------------------------------------------------------------------------- # Tests del comando fix_partidas_error # Una partida descargado=True solo es válida si alguno de sus documentos # contiene consultarPartidaRespuesta sin tieneError=true. Partidas que solo # tienen el REQUEST (o errores) deben volver a descargado=False. # --------------------------------------------------------------------------- from io import StringIO from types import SimpleNamespace from django.core.management import call_command from django.test import TestCase, SimpleTestCase from core.partida_docs import es_doc_de_partida, patron_regex_partida from api.customs.serializers import PartidaSerializer, CoveSerializer, EDocumentSerializer from api.customs.models import Partida from api.record.models import Document, DocumentType XML_RESPUESTA_VALIDA = ( "" '' '' "false" "" ) XML_ERROR_VUCEM = ( "" '' '' "true" "" ) XML_ECO_REQUEST = ( "" '' "" "" ) class _FakeMinioObject: """Simula el objeto retornado por minio get_object.""" def __init__(self, content): self._content = content def read(self): return self._content def close(self): pass def release_conn(self): pass class FixPartidasErrorCommandTests(TestCase): PED_APP = "24-01-3420-1234567" def setUp(self): from api.customs.models import Partida from api.record.models import DocumentType self.licencia = Licencia.objects.create(nombre="LicFixPartidas", almacenamiento=100) self.org = Organizacion.objects.create( nombre="OrgFixPartidas", licencia=self.licencia, is_active=True, is_verified=True ) # Pedimento VÁLIDO (no malformado): el comando ya no se limita a malformados self.pedimento = Pedimento.objects.create( organizacion=self.org, pedimento="1234567", pedimento_app=self.PED_APP, aduana="034", patente="3420", numero_operacion="12345678", ) self.partida = Partida.objects.create( pedimento=self.pedimento, organizacion=self.org, numero_partida=1, descargado=True, ) self.type_resp = DocumentType.objects.get_or_create(id=1, defaults={"nombre": "XML"})[0] self.type_req = DocumentType.objects.get_or_create(id=17, defaults={"nombre": "PT Request"})[0] self.type_err = DocumentType.objects.get_or_create(id=18, defaults={"nombre": "PT Error"})[0] # Storage simulado: dict path -> bytes self.storage = {} patcher = patch("api.customs.management.commands.fix_partidas_error.minio_client") self.minio = patcher.start() self.addCleanup(patcher.stop) self.minio._bucket_name = "test-bucket" self.minio.file_exists.side_effect = lambda name: name in self.storage self.minio._client.get_object.side_effect = ( lambda bucket, name: _FakeMinioObject(self.storage[name]) ) self.minio.upload_file.side_effect = ( lambda name, file_data=None, content_type=None: self.storage.__setitem__( name, file_data.read() ) ) self.minio.delete_file.side_effect = lambda name: self.storage.pop(name, None) def _doc(self, filename, doc_type, content=None): from api.record.models import Document path = f"org/{self.PED_APP}/{filename}" doc = Document.objects.create( organizacion=self.org, pedimento=self.pedimento, document_type=doc_type, archivo=path, size=100, extension="xml", ) if content is not None: self.storage[path] = content.encode("utf-8") return doc def _run(self, **kwargs): out = StringIO() call_command("fix_partidas_error", stdout=out, stderr=StringIO(), **kwargs) return out.getvalue() def test_partida_solo_request_se_marca_no_descargada(self): """El caso reportado: descargado=True pero solo existe el XML del REQUEST.""" self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST) self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() self.assertFalse(self.partida.descargado) def test_partida_sin_documentos_se_marca_no_descargada(self): """descargado=True sin ningún documento tampoco es una descarga real.""" self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() self.assertFalse(self.partida.descargado) def test_partida_con_respuesta_valida_permanece_descargada(self): """Con consultarPartidaRespuesta sin error la partida no se toca.""" self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST) self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_RESPUESTA_VALIDA) self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() self.assertTrue(self.partida.descargado) def test_doc_con_error_vucem_se_renombra_y_marca_no_descargada(self): """tieneError=true: doc → type 18 con sufijo _ERROR y partida → False.""" doc = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ERROR_VUCEM) old_path = doc.archivo.name self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() doc.refresh_from_db() self.assertFalse(self.partida.descargado) self.assertEqual(doc.document_type_id, 18) self.assertTrue(doc.archivo.name.endswith(f"vu_PT_{self.PED_APP}_1_ERROR.xml")) self.assertTrue(doc.vu) self.assertNotIn(old_path, self.storage) self.assertIn(doc.archivo.name, self.storage) def test_eco_de_request_guardado_como_respuesta_se_reclasifica(self): """Un eco de consultarPartidaPeticion guardado como respuesta se reclasifica a type 17 sin chocar con el REQUEST real existente.""" self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST) doc = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ECO_REQUEST) self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() doc.refresh_from_db() self.assertFalse(self.partida.descargado) self.assertEqual(doc.document_type_id, 17) # El nombre sin índice ya lo usa el REQUEST real → debe ir con _1 self.assertTrue(doc.archivo.name.endswith(f"vu_PT_{self.PED_APP}_1_REQUEST_1.xml")) def test_doc_ausente_sin_canario_no_cambia_partida(self): """Archivo ausente y NINGÚN archivo del pedimento en storage: posible storage equivocado (p. ej. dev) → sin cambios.""" self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, content=None) self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() self.assertTrue(self.partida.descargado) def test_registro_fantasma_con_storage_real_se_marca_no_descargada(self): """Document type 1 en BD sin archivo en storage, pero el REQUEST sí existe físicamente (canario): el storage es el correcto, el registro es fantasma → la partida no tiene XML de partida → descargado=False.""" self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST) fantasma = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, content=None) self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() fantasma.refresh_from_db() self.assertFalse(self.partida.descargado) # El registro fantasma se reporta pero no se modifica ni se borra self.assertEqual(fantasma.document_type_id, 1) def test_storage_inaccesible_no_cambia_partida(self): """Excepción al consultar storage (conexión caída): sin cambios.""" self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ERROR_VUCEM) self.minio.file_exists.side_effect = Exception("connection refused") self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() self.assertTrue(self.partida.descargado) def test_naming_legacy_valida_partida(self): """Documentos con nomenclatura legacy (partida al final) también validan.""" self._doc("vu_PT_010Imp_034_3420_1234567_1.xml", self.type_resp, XML_RESPUESTA_VALIDA) self._run(pedimento=str(self.pedimento.id)) self.partida.refresh_from_db() self.assertTrue(self.partida.descargado) def test_dry_run_no_modifica(self): """--dry-run reporta pero no toca BD ni storage.""" doc = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ERROR_VUCEM) self._run(pedimento=str(self.pedimento.id), dry_run=True) self.partida.refresh_from_db() doc.refresh_from_db() self.assertTrue(self.partida.descargado) self.assertEqual(doc.document_type_id, 1) self.assertIn(doc.archivo.name, self.storage) def test_universo_general_incluye_pedimentos_validos(self): """Sin --pedimento ni --solo-malformados también procesa pedimentos bien formados.""" self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST) self._run() self.partida.refresh_from_db() self.assertFalse(self.partida.descargado) def test_solo_malformados_excluye_pedimentos_validos(self): """Con --solo-malformados un pedimento bien formado no se procesa.""" self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST) self._run(solo_malformados=True) self.partida.refresh_from_db() self.assertTrue(self.partida.descargado) def test_no_confunde_partida_1_con_11(self): """La asignación por nombre no debe mezclar partida 1 con partida 11.""" from api.customs.management.commands.fix_partidas_error import Command docs = [ SimpleNamespace(id=1, document_type_id=1, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_1.xml")), SimpleNamespace(id=2, document_type_id=17, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_1_REQUEST.xml")), SimpleNamespace(id=3, document_type_id=1, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_11.xml")), SimpleNamespace(id=4, document_type_id=17, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_11_REQUEST.xml")), ] cmd = Command() ids_p1 = {d.id for d in cmd._docs_de_partida(docs, self.PED_APP, 1)} ids_p11 = {d.id for d in cmd._docs_de_partida(docs, self.PED_APP, 11)} self.assertEqual(ids_p1, {1, 2}) self.assertEqual(ids_p11, {3, 4}) class PartidaDocsHelperTests(SimpleTestCase): """Matching documento→partida (core.partida_docs), sin BD.""" APP = "24-01-3420-1234567" def test_es_doc_de_partida_cubre_los_tres_formatos(self): for nombre in ( f"documents/vu_PT_{self.APP}_1", # #1 sin extensión f"documents/vu_PT_{self.APP}_1.xml", # #2 f"documents/vu_PT_{self.APP}_1_a1b2c3.xml", # #3 sufijo hex del storage f"documents/vu_PT_{self.APP}_1_REQUEST.xml", # REQUEST (coincide por nombre) ): self.assertTrue(es_doc_de_partida(nombre, self.APP, 1), nombre) def test_es_doc_de_partida_no_confunde_1_con_11_ni_100(self): for n in (11, 12, 100): self.assertFalse(es_doc_de_partida(f"vu_PT_{self.APP}_{n}.xml", self.APP, 1)) self.assertFalse(es_doc_de_partida(f"vu_PT_{self.APP}_{n}_x.xml", self.APP, 1)) # a la inversa: la 11 sí coincide con la 11, no con la 1 self.assertTrue(es_doc_de_partida(f"vu_PT_{self.APP}_11.xml", self.APP, 11)) self.assertFalse(es_doc_de_partida(f"vu_PT_{self.APP}_1.xml", self.APP, 11)) def test_es_doc_de_partida_case_insensitive_y_con_ruta(self): self.assertTrue(es_doc_de_partida(f"ORG/X/VU_PT_{self.APP}_1.XML", self.APP, 1)) def test_es_doc_de_partida_vacio_o_none(self): self.assertFalse(es_doc_de_partida("", self.APP, 1)) self.assertFalse(es_doc_de_partida(None, self.APP, 1)) def test_es_doc_de_partida_legacy_segun_flag(self): legacy = "org/x/vu_PT_010Imp_034_3420_1234567_1.xml" # número de partida al final self.assertTrue(es_doc_de_partida(legacy, self.APP, 1, incluir_legacy=True)) self.assertFalse(es_doc_de_partida(legacy, self.APP, 1, incluir_legacy=False)) # el formato legacy tampoco confunde la 1 con la 11 legacy11 = "org/x/vu_PT_010Imp_034_3420_1234567_11.xml" self.assertFalse(es_doc_de_partida(legacy11, self.APP, 1, incluir_legacy=True)) def test_patron_regex_partida_semantica(self): import re rx = re.compile(patron_regex_partida(self.APP, 1), re.IGNORECASE) self.assertTrue(rx.search(f"documents/vu_PT_{self.APP}_1")) self.assertTrue(rx.search(f"documents/vu_PT_{self.APP}_1.xml")) self.assertTrue(rx.search(f"documents/vu_PT_{self.APP}_1_a1b2.xml")) self.assertFalse(rx.search(f"documents/vu_PT_{self.APP}_11.xml")) self.assertFalse(rx.search(f"documents/vu_PT_{self.APP}_100.xml")) # pedimento_app se trata como literal: otro pedimento no coincide self.assertFalse(rx.search("documents/vu_PT_99-99-9999-9999999_1.xml")) # legacy solo cuando se pide rxl = re.compile(patron_regex_partida(self.APP, 1, incluir_legacy=True), re.IGNORECASE) self.assertTrue(rxl.search("vu_PT_010Imp_034_3420_1234567_1.xml")) class PartidaDocumentosSerializerTests(TestCase): """get_documentos (PartidaSerializer) y el prefetch de PartidaViewSet asignan los documentos correctos a cada partida por nombre de archivo.""" APP = "24-01-3420-1234567" def setUp(self): from api.organization.models import Organizacion from api.licence.models import Licencia from .models import Pedimento self.licencia = Licencia.objects.create(nombre="LicPartDocs", almacenamiento=100) self.org = Organizacion.objects.create( nombre="OrgPartDocs", licencia=self.licencia, is_active=True, is_verified=True ) self.pedimento = Pedimento.objects.create( organizacion=self.org, pedimento="1234567", pedimento_app=self.APP, aduana="034", patente="3420", numero_operacion="12345678", ) self.p1 = Partida.objects.create( pedimento=self.pedimento, organizacion=self.org, numero_partida=1, descargado=True ) self.p11 = Partida.objects.create( pedimento=self.pedimento, organizacion=self.org, numero_partida=11, descargado=True ) self.type_resp = DocumentType.objects.get_or_create(id=1, defaults={"nombre": "XML"})[0] self.type_req = DocumentType.objects.get_or_create(id=17, defaults={"nombre": "PT Request"})[0] def _doc(self, filename, doc_type=None): return Document.objects.create( organizacion=self.org, pedimento=self.pedimento, document_type=doc_type or self.type_resp, archivo=f"documents/{filename}", size=100, extension="xml", ) def _blob(self, data): return " ".join(d["archivo"] for d in data["documentos"]) def test_get_documentos_tres_formatos_sin_confundir(self): self._doc(f"vu_PT_{self.APP}_1") # #1 sin extensión self._doc(f"vu_PT_{self.APP}_1.xml") # #2 self._doc(f"vu_PT_{self.APP}_1_a1b2c3.xml") # #3 hex self._doc(f"vu_PT_{self.APP}_1_REQUEST.xml", self.type_req) # tipo 17: no debe salir self._doc(f"vu_PT_{self.APP}_11.xml") # partida 11 self._doc(f"vu_PT_{self.APP}_11_a1b2c3.xml") # partida 11 data = PartidaSerializer(self.p1).data # sin contexto -> camino fallback blob = self._blob(data) # los 3 documentos tipo-1 de la partida 1; el REQUEST (17) excluido self.assertEqual(len(data["documentos"]), 3) self.assertIn(f"vu_PT_{self.APP}_1.xml", blob) self.assertIn(f"vu_PT_{self.APP}_1_a1b2c3.xml", blob) self.assertNotIn("_11", blob) # no arrastra la partida 11 self.assertNotIn("REQUEST", blob) # no incluye el REQUEST tipo 17 def test_get_documentos_incluye_legacy(self): self._doc("vu_PT_010Imp_034_3420_1234567_1.xml") # legacy tipo 1, número al final data = PartidaSerializer(self.p1).data self.assertEqual(len(data["documentos"]), 1) def test_get_documentos_via_fk(self): # save() liga la FK al crear; get_documentos lee por la FK real. d1 = self._doc(f"vu_PT_{self.APP}_1.xml") self._doc(f"vu_PT_{self.APP}_11.xml") d1.refresh_from_db() self.assertEqual(d1.partida_id, self.p1.id) # FK ligada en save() data = PartidaSerializer(self.p1).data self.assertEqual(len(data["documentos"]), 1) self.assertIn(f"vu_PT_{self.APP}_1.xml", self._blob(data)) self.assertNotIn("_11", self._blob(data)) def test_patron_regex_partida_en_bd(self): d1 = self._doc(f"vu_PT_{self.APP}_1") d2 = self._doc(f"vu_PT_{self.APP}_1.xml") d3 = self._doc(f"vu_PT_{self.APP}_1_a1b2c3.xml") d_otro = self._doc(f"vu_PT_{self.APP}_11.xml") ids = set( Document.objects.filter( pedimento=self.pedimento, archivo__iregex=patron_regex_partida(self.APP, 1), ).values_list("id", flat=True) ) self.assertEqual(ids, {d1.id, d2.id, d3.id}) self.assertNotIn(d_otro.id, ids) def test_prefetch_documentos_vu_evita_n_plus_1(self): from django.db.models import Prefetch self._doc(f"vu_PT_{self.APP}_1.xml") self._doc(f"vu_PT_{self.APP}_11.xml") prefetch = Prefetch( 'documents', queryset=Document.objects.filter(document_type_id=1).select_related('pedimento'), to_attr='documentos_vu', ) partidas = list( Partida.objects.filter(pedimento=self.pedimento) .order_by('numero_partida').prefetch_related(prefetch) ) # Serializar la lista no dispara consultas extra: todo viene del prefetch. with self.assertNumQueries(0): data = PartidaSerializer(partidas, many=True).data por_num = {d['numero_partida']: d['documentos'] for d in data} self.assertEqual(len(por_num[1]), 1) self.assertEqual(len(por_num[11]), 1) class DocumentLinksHelperTests(SimpleTestCase): """Resolver tipo→sección y matcher por frontera (core.document_links), sin BD.""" # pedimento_app con guiones bajos (caso real): no se puede extraer la llave # partiendo por '_'; por eso se itera la entidad con su número exacto. APP = "0101_230_1703_3004804" def test_seccion_de_tipo(self): from core.document_links import seccion_de_tipo self.assertEqual(seccion_de_tipo(1), 'partida') self.assertEqual(seccion_de_tipo(8), 'cove') self.assertEqual(seccion_de_tipo(7), 'cove') # acuse cove self.assertEqual(seccion_de_tipo(5), 'edocument') self.assertEqual(seccion_de_tipo(4), 'edocument') # acuse edoc self.assertIsNone(seccion_de_tipo(2)) # PC nativo self.assertIsNone(seccion_de_tipo(3)) # remesa nativo self.assertIsNone(seccion_de_tipo(None)) def test_coincide_cove(self): from core.document_links import coincide self.assertTrue(coincide(f"documents/vu_COVE_{self.APP}_654001.xml", 'cove', self.APP, "654001")) self.assertTrue(coincide(f"documents/vu_AC_COVE_{self.APP}_654001.pdf", 'cove', self.APP, "654001")) self.assertTrue(coincide(f"documents/vu_COVE_{self.APP}_654001_REQUEST.xml", 'cove', self.APP, "654001")) # colisión de prefijo: 654001 no debe matchear 6540012 self.assertFalse(coincide(f"documents/vu_COVE_{self.APP}_6540012.xml", 'cove', self.APP, "654001")) def test_coincide_edocument(self): from core.document_links import coincide self.assertTrue(coincide(f"documents/vu_ED_{self.APP}_EDOC001.pdf", 'edocument', self.APP, "EDOC001")) self.assertTrue(coincide(f"documents/vu_AC_{self.APP}_EDOC001.pdf", 'edocument', self.APP, "EDOC001")) self.assertFalse(coincide(f"documents/vu_ED_{self.APP}_EDOC0011.pdf", 'edocument', self.APP, "EDOC001")) class DocumentFKResolutionTests(TestCase): """save()-resolución de FK por sección, lectura cove/edoc por FK y backfill.""" APP = "0101_230_1703_3004804" # con guiones bajos def setUp(self): from api.organization.models import Organizacion from api.licence.models import Licencia from .models import Pedimento, Cove, EDocument self.licencia = Licencia.objects.create(nombre="LicFK", almacenamiento=100) self.org = Organizacion.objects.create( nombre="OrgFK", licencia=self.licencia, is_active=True, is_verified=True ) self.pedimento = Pedimento.objects.create( organizacion=self.org, pedimento="1234567", pedimento_app=self.APP, aduana="034", patente="3420", numero_operacion="12345678", ) self.partida = Partida.objects.create( pedimento=self.pedimento, organizacion=self.org, numero_partida=3, descargado=True ) self.cove = Cove.objects.create( pedimento=self.pedimento, organizacion=self.org, numero_cove="654001" ) self.edoc = EDocument.objects.create( pedimento=self.pedimento, organizacion=self.org, numero_edocument="EDOC001" ) for tid, nombre in [(1, "PT"), (8, "COVE"), (7, "AC_COVE"), (5, "ED"), (4, "AC_ED"), (2, "PC"), (19, "COVE_REQ")]: DocumentType.objects.get_or_create(id=tid, defaults={"nombre": nombre}) def _doc(self, filename, type_id): return Document.objects.create( organizacion=self.org, pedimento=self.pedimento, document_type_id=type_id, archivo=f"documents/{filename}", size=100, extension="xml", ) def test_save_liga_fk_por_seccion(self): d_pt = self._doc(f"vu_PT_{self.APP}_3.xml", 1) d_cv = self._doc(f"vu_COVE_{self.APP}_654001.xml", 8) d_accv = self._doc(f"vu_AC_COVE_{self.APP}_654001.pdf", 7) d_ed = self._doc(f"vu_ED_{self.APP}_EDOC001.pdf", 5) d_aced = self._doc(f"vu_AC_{self.APP}_EDOC001.pdf", 4) d_pc = self._doc(f"vu_PC_{self.APP}.xml", 2) # nativo: sin FK for d in (d_pt, d_cv, d_accv, d_ed, d_aced, d_pc): d.refresh_from_db() self.assertEqual(d_pt.partida_id, self.partida.id) self.assertEqual(d_cv.cove_id, self.cove.id) self.assertEqual(d_accv.cove_id, self.cove.id) # acuse cove → cove padre self.assertEqual(d_ed.edocument_id, self.edoc.id) self.assertEqual(d_aced.edocument_id, self.edoc.id) # acuse edoc → edoc padre self.assertIsNone(d_pc.partida_id) self.assertIsNone(d_pc.cove_id) self.assertIsNone(d_pc.edocument_id) def test_lectura_cove_edoc_por_fk(self): self._doc(f"vu_COVE_{self.APP}_654001.xml", 8) self._doc(f"vu_AC_COVE_{self.APP}_654001.pdf", 7) self._doc(f"vu_COVE_{self.APP}_654001_REQUEST.xml", 19) # request: excluido en lectura self._doc(f"vu_ED_{self.APP}_EDOC001.pdf", 5) cove_data = CoveSerializer(self.cove).data edoc_data = EDocumentSerializer(self.edoc).data self.assertEqual(len(cove_data["documentos"]), 2) # cove + acuse, sin request self.assertEqual(len(edoc_data["documentos"]), 1) def test_backfill_liga_filas_existentes(self): d_pt = self._doc(f"vu_PT_{self.APP}_3.xml", 1) d_cv = self._doc(f"vu_COVE_{self.APP}_654001.xml", 8) # Simular filas viejas sin ligar (save() las ligó; las desligamos en BD). Document.objects.filter(id__in=[d_pt.id, d_cv.id]).update( partida=None, cove=None, edocument=None ) # dry-run no escribe call_command("backfill_document_links", pedimento=str(self.pedimento.id), dry_run=True, stdout=StringIO()) d_pt.refresh_from_db() self.assertIsNone(d_pt.partida_id) # ejecución real liga call_command("backfill_document_links", pedimento=str(self.pedimento.id), stdout=StringIO()) d_pt.refresh_from_db() d_cv.refresh_from_db() self.assertEqual(d_pt.partida_id, self.partida.id) self.assertEqual(d_cv.cove_id, self.cove.id) # idempotente: re-ejecutar no rompe ni cambia call_command("backfill_document_links", pedimento=str(self.pedimento.id), stdout=StringIO()) d_pt.refresh_from_db() self.assertEqual(d_pt.partida_id, self.partida.id) def _tres_copias_edoc(self): """3 copias del mismo edoc (type 5) con created_at d1