276 lines
13 KiB
Python
276 lines
13 KiB
Python
|
|
from django.urls import reverse
|
|
from django.test import TestCase
|
|
from rest_framework.test import APITestCase, APIClient
|
|
from rest_framework import status
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from unittest.mock import patch, MagicMock
|
|
from api.organization.models import Organizacion, UsoAlmacenamiento
|
|
from api.cuser.models import CustomUser
|
|
from api.customs.models import Pedimento
|
|
from api.licence.models import Licencia
|
|
from api.customs.views import is_same_document, get_clean_base_filename
|
|
from .models import Document, DocumentType
|
|
import io
|
|
|
|
class DocumentViewSetTests(APITestCase):
|
|
def setUp(self):
|
|
self.org = Organizacion.objects.create(nombre="OrgTest", is_active=True, is_verified=True)
|
|
self.pedimento = Pedimento.objects.create(organizacion=self.org, numero="123456")
|
|
self.admin = CustomUser.objects.create_user(username="admin", password="adminpass", organizacion=self.org)
|
|
self.admin.groups.create(name="admin")
|
|
self.superuser = CustomUser.objects.create_superuser(username="superuser", password="superpass")
|
|
self.client = APIClient()
|
|
|
|
def test_list_documents_only_own_org(self):
|
|
doc1 = Document.objects.create(organizacion=self.org, pedimento=self.pedimento, archivo="documents/test1.pdf", size=100, extension="pdf")
|
|
org2 = Organizacion.objects.create(nombre="OrgTest2", is_active=True, is_verified=True)
|
|
ped2 = Pedimento.objects.create(organizacion=org2, numero="654321")
|
|
Document.objects.create(organizacion=org2, pedimento=ped2, archivo="documents/test2.pdf", size=200, extension="pdf")
|
|
self.client.force_authenticate(user=self.admin)
|
|
url = reverse('Document-list')
|
|
response = self.client.get(url)
|
|
ids = [d['id'] for d in response.data]
|
|
self.assertIn(str(doc1.id), ids)
|
|
self.assertEqual(len(ids), 1)
|
|
|
|
def test_create_document_success(self):
|
|
self.client.force_authenticate(user=self.admin)
|
|
file_content = b"dummy pdf content"
|
|
archivo = SimpleUploadedFile("test.pdf", file_content, content_type="application/pdf")
|
|
url = reverse('Document-list')
|
|
data = {
|
|
"pedimento": str(self.pedimento.id),
|
|
"archivo": archivo,
|
|
"size": len(file_content),
|
|
"extension": "pdf"
|
|
}
|
|
response = self.client.post(url, data, format='multipart')
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
doc = Document.objects.get(id=response.data['id'])
|
|
self.assertEqual(doc.organizacion, self.org)
|
|
|
|
def test_update_document_size(self):
|
|
doc = Document.objects.create(organizacion=self.org, pedimento=self.pedimento, archivo="documents/test1.pdf", size=100, extension="pdf")
|
|
self.client.force_authenticate(user=self.admin)
|
|
url = reverse('Document-detail', args=[doc.id])
|
|
file_content = b"new content"
|
|
archivo = SimpleUploadedFile("test2.pdf", file_content, content_type="application/pdf")
|
|
data = {
|
|
"archivo": archivo,
|
|
"size": len(file_content),
|
|
"extension": "pdf"
|
|
}
|
|
response = self.client.patch(url, data, format='multipart')
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
doc.refresh_from_db()
|
|
self.assertEqual(doc.size, len(file_content))
|
|
|
|
def test_delete_document_frees_storage(self):
|
|
doc = Document.objects.create(organizacion=self.org, pedimento=self.pedimento, archivo="documents/test1.pdf", size=100, extension="pdf")
|
|
UsoAlmacenamiento.objects.create(organizacion=self.org, espacio_utilizado=100)
|
|
self.client.force_authenticate(user=self.admin)
|
|
url = reverse('Document-detail', args=[doc.id])
|
|
response = self.client.delete(url)
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
uso = UsoAlmacenamiento.objects.get(organizacion=self.org)
|
|
self.assertEqual(uso.espacio_utilizado, 0)
|
|
|
|
def test_permission_denied_for_other_org(self):
|
|
org2 = Organizacion.objects.create(nombre="OrgTest2", is_active=True, is_verified=True)
|
|
ped2 = Pedimento.objects.create(organizacion=org2, numero="654321")
|
|
doc2 = Document.objects.create(organizacion=org2, pedimento=ped2, archivo="documents/test2.pdf", size=200, extension="pdf")
|
|
self.client.force_authenticate(user=self.admin)
|
|
url = reverse('Document-detail', args=[doc2.id])
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
|
|
|
def test_superuser_can_access_all(self):
|
|
org2 = Organizacion.objects.create(nombre="OrgTest2", is_active=True, is_verified=True)
|
|
ped2 = Pedimento.objects.create(organizacion=org2, numero="654321")
|
|
doc2 = Document.objects.create(organizacion=org2, pedimento=ped2, archivo="documents/test2.pdf", size=200, extension="pdf")
|
|
self.client.force_authenticate(user=self.superuser)
|
|
url = reverse('Document-detail', args=[doc2.id])
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
def test_protected_download_requires_auth(self):
|
|
doc = Document.objects.create(organizacion=self.org, pedimento=self.pedimento, archivo="documents/test1.pdf", size=100, extension="pdf")
|
|
url = reverse('descargar-documento', args=[doc.id])
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests unitarios para las funciones helper de comparación de documentos
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class DocumentNameHelperTests(TestCase):
|
|
"""Verifica que get_clean_base_filename e is_same_document manejan
|
|
correctamente el sufijo UUID de 8 chars que añade storage_service."""
|
|
|
|
def test_strips_uuid_suffix(self):
|
|
self.assertEqual(get_clean_base_filename('informe_a1b2c3d4.pdf'), 'informe')
|
|
|
|
def test_no_suffix_unchanged(self):
|
|
self.assertEqual(get_clean_base_filename('informe.pdf'), 'informe')
|
|
|
|
def test_is_same_document_matches_stored_uuid_name(self):
|
|
"""El archivo guardado tiene sufijo, el nuevo no — deben coincidir."""
|
|
doc = MagicMock()
|
|
doc.archivo.name = 'org_1/documents/24-01-3420-1234567/informe_a1b2c3d4.pdf'
|
|
doc.extension = 'pdf'
|
|
self.assertTrue(is_same_document(doc, 'informe.pdf'))
|
|
|
|
def test_is_same_document_different_name_no_match(self):
|
|
doc = MagicMock()
|
|
doc.archivo.name = 'org_1/documents/ped/informe_a1b2c3d4.pdf'
|
|
doc.extension = 'pdf'
|
|
self.assertFalse(is_same_document(doc, 'otro.pdf'))
|
|
|
|
def test_is_same_document_different_extension_no_match(self):
|
|
doc = MagicMock()
|
|
doc.archivo.name = 'org_1/documents/ped/informe_a1b2c3d4.pdf'
|
|
doc.extension = 'pdf'
|
|
self.assertFalse(is_same_document(doc, 'informe.xml'))
|
|
|
|
def test_both_clean_names_equal(self):
|
|
"""Dos archivos con UUID distintos pero mismo nombre base deben coincidir."""
|
|
doc = MagicMock()
|
|
doc.archivo.name = 'org_1/documents/ped/pedimento_a1b2c3d4.xml'
|
|
doc.extension = 'xml'
|
|
self.assertTrue(is_same_document(doc, 'pedimento_b5c6d7e8.xml'))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests de integración para bulk-upload (DocumentViewSet.bulk_upload)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BulkUploadReplaceTests(APITestCase):
|
|
"""Verifica que bulk-upload reemplaza documentos existentes en vez de duplicar
|
|
y que no quedan archivos residuales en el storage."""
|
|
|
|
def setUp(self):
|
|
self.licencia = Licencia.objects.create(nombre="Lic100GB", almacenamiento=100)
|
|
self.org = Organizacion.objects.create(
|
|
nombre="OrgBulkUpload",
|
|
licencia=self.licencia,
|
|
is_active=True,
|
|
is_verified=True,
|
|
)
|
|
self.user = CustomUser.objects.create_user(
|
|
username="bulkuploaduser", password="pass", organizacion=self.org
|
|
)
|
|
self.pedimento = Pedimento.objects.create(
|
|
organizacion=self.org,
|
|
pedimento="1234567",
|
|
pedimento_app="24-01-3420-1234567",
|
|
)
|
|
self.doc_type = DocumentType.objects.get_or_create(nombre="Documento General")[0]
|
|
self.url = reverse("Document-bulk-upload")
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def _post_file(self, filename, content=b"contenido de prueba"):
|
|
archivo = SimpleUploadedFile(filename, content, content_type="application/pdf")
|
|
return self.client.post(
|
|
self.url,
|
|
{"pedimento_id": str(self.pedimento.id), "files": [archivo]},
|
|
format="multipart",
|
|
)
|
|
|
|
@patch("api.record.views.storage_service")
|
|
def test_new_file_creates_document(self, mock_st):
|
|
"""Subir un archivo nuevo crea exactamente un Document."""
|
|
mock_st.save_document.return_value = "org_1/documents/ped/informe_a1b2c3d4.pdf"
|
|
|
|
response = self._post_file("informe.pdf")
|
|
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(Document.objects.filter(pedimento=self.pedimento).count(), 1)
|
|
mock_st.delete_file.assert_not_called()
|
|
|
|
@patch("api.record.views.storage_service")
|
|
def test_duplicate_replaces_not_creates(self, mock_st):
|
|
"""Re-subir el mismo archivo debe actualizar el Document existente,
|
|
no crear uno nuevo."""
|
|
old_path = "org_1/documents/24-01-3420-1234567/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 = "org_1/documents/24-01-3420-1234567/informe_b5c6d7e8.pdf"
|
|
mock_st.save_document.return_value = new_path
|
|
mock_st.delete_file.return_value = True
|
|
|
|
response = self._post_file("informe.pdf", b"contenido actualizado")
|
|
|
|
self.assertIn(response.status_code, [status.HTTP_201_CREATED, status.HTTP_207_MULTI_STATUS])
|
|
docs = Document.objects.filter(pedimento=self.pedimento)
|
|
# Un único Document — sin duplicados
|
|
self.assertEqual(docs.count(), 1)
|
|
# Es el mismo registro (mismo UUID)
|
|
self.assertEqual(docs.first().id, old_doc.id)
|
|
# El campo archivo fue actualizado
|
|
old_doc.refresh_from_db()
|
|
self.assertEqual(old_doc.archivo.name, new_path)
|
|
|
|
@patch("api.record.views.storage_service")
|
|
def test_replace_deletes_old_storage_file(self, mock_st):
|
|
"""Al reemplazar, delete_file debe llamarse con la ruta del archivo viejo."""
|
|
old_path = "org_1/documents/24-01-3420-1234567/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.return_value = "org_1/documents/24-01-3420-1234567/informe_b5c6d7e8.pdf"
|
|
mock_st.delete_file.return_value = True
|
|
|
|
self._post_file("informe.pdf")
|
|
|
|
mock_st.delete_file.assert_called_once_with(old_path)
|
|
|
|
@patch("api.record.views.storage_service")
|
|
def test_different_filename_creates_new_document(self, mock_st):
|
|
"""Archivo con nombre diferente debe crear un Document adicional."""
|
|
Document.objects.create(
|
|
organizacion=self.org,
|
|
pedimento=self.pedimento,
|
|
document_type=self.doc_type,
|
|
archivo="org_1/documents/ped/informe_a1b2c3d4.pdf",
|
|
size=500,
|
|
extension="pdf",
|
|
)
|
|
mock_st.save_document.return_value = "org_1/documents/ped/otro_b5c6d7e8.pdf"
|
|
|
|
self._post_file("otro.pdf")
|
|
|
|
self.assertEqual(Document.objects.filter(pedimento=self.pedimento).count(), 2)
|
|
mock_st.delete_file.assert_not_called()
|
|
|
|
@patch("api.record.views.storage_service")
|
|
def test_multiple_files_no_cross_replacement(self, mock_st):
|
|
"""Subir dos archivos distintos en la misma petición crea dos Documents."""
|
|
mock_st.save_document.side_effect = [
|
|
"org_1/documents/ped/a_a1b2c3d4.pdf",
|
|
"org_1/documents/ped/b_a1b2c3d4.pdf",
|
|
]
|
|
archivos = [
|
|
SimpleUploadedFile("a.pdf", b"contenido a", content_type="application/pdf"),
|
|
SimpleUploadedFile("b.pdf", b"contenido b", content_type="application/pdf"),
|
|
]
|
|
self.client.post(
|
|
self.url,
|
|
{"pedimento_id": str(self.pedimento.id), "files": archivos},
|
|
format="multipart",
|
|
)
|
|
self.assertEqual(Document.objects.filter(pedimento=self.pedimento).count(), 2)
|
|
mock_st.delete_file.assert_not_called()
|