227 lines
9.6 KiB
Python
227 lines
9.6 KiB
Python
|
|
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)
|