backend/api/customs/tests.py

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 = (
    "<?xml version='1.0' encoding='UTF-8'?>"
    '<S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body>'
    '<ns9:consultarPartidaRespuesta xmlns:ns9="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpartida">'
    "<tieneError>false</tieneError><ns9:partida/></ns9:consultarPartidaRespuesta>"
    "</S:Body></S:Envelope>"
)

XML_ERROR_VUCEM = (
    "<?xml version='1.0' encoding='UTF-8'?>"
    '<S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body>'
    '<ns9:consultarPartidaRespuesta xmlns:ns9="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpartida">'
    "<tieneError>true</tieneError></ns9:consultarPartidaRespuesta>"
    "</S:Body></S:Envelope>"
)

XML_ECO_REQUEST = (
    "<?xml version='1.0' encoding='UTF-8'?>"
    '<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"'
    ' xmlns:con="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpartida"><soapenv:Body>'
    "<con:consultarPartidaPeticion><con:peticion/></con:consultarPartidaPeticion>"
    "</soapenv:Body></soapenv:Envelope>"
)


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<d2<d3 y archivos distintos."""
        from django.utils import timezone
        import datetime
        d1 = self._doc(f"vu_ED_{self.APP}_EDOC001_aaa.pdf", 5)
        d2 = self._doc(f"vu_ED_{self.APP}_EDOC001_bbb.pdf", 5)
        d3 = self._doc(f"vu_ED_{self.APP}_EDOC001_ccc.pdf", 5)
        base = timezone.now()
        Document.objects.filter(id=d1.id).update(created_at=base - datetime.timedelta(days=2))
        Document.objects.filter(id=d2.id).update(created_at=base - datetime.timedelta(days=1))
        Document.objects.filter(id=d3.id).update(created_at=base)
        return d1, d2, d3

    def test_dedup_conserva_mas_reciente_y_es_idempotente(self):
        from unittest.mock import patch
        d1, d2, d3 = self._tres_copias_edoc()
        self.assertEqual(Document.objects.filter(edocument=self.edoc, document_type_id=5).count(), 3)
        with patch('api.customs.management.commands.dedup_documents.storage_service') as st:
            st.file_exists.return_value = True
            call_command('dedup_documents', pedimento=str(self.pedimento.id), stdout=StringIO())
            self.assertEqual(st.delete_file.call_count, 2)  # borró los 2 viejos de MinIO
        restantes = list(Document.objects.filter(edocument=self.edoc, document_type_id=5))
        self.assertEqual(len(restantes), 1)
        self.assertEqual(restantes[0].id, d3.id)  # conservó el más reciente
        # idempotente: re-correr no borra nada
        with patch('api.customs.management.commands.dedup_documents.storage_service') as st2:
            st2.file_exists.return_value = True
            call_command('dedup_documents', pedimento=str(self.pedimento.id), stdout=StringIO())
            self.assertEqual(st2.delete_file.call_count, 0)

    def test_dedup_dry_run_no_borra(self):
        from unittest.mock import patch
        self._tres_copias_edoc()
        with patch('api.customs.management.commands.dedup_documents.storage_service') as st:
            call_command('dedup_documents', pedimento=str(self.pedimento.id), dry_run=True, stdout=StringIO())
            self.assertEqual(st.delete_file.call_count, 0)
        self.assertEqual(Document.objects.filter(edocument=self.edoc, document_type_id=5).count(), 3)

    def test_dedup_conserva_el_que_tiene_archivo_en_storage(self):
        from unittest.mock import patch
        d1, d2, d3 = self._tres_copias_edoc()  # d3 el más reciente
        # En storage solo existe el de d1 (el más viejo); los más nuevos no.
        with patch('api.customs.management.commands.dedup_documents.storage_service') as st:
            st.file_exists.side_effect = lambda nombre: nombre.endswith('_aaa.pdf')
            call_command('dedup_documents', pedimento=str(self.pedimento.id), stdout=StringIO())
        restantes = list(Document.objects.filter(edocument=self.edoc, document_type_id=5))
        self.assertEqual(len(restantes), 1)
        self.assertEqual(restantes[0].id, d1.id)  # conservó el único con archivo válido