✅ Media exporter: tests

1 files changed, 154 insertions, 1 deletions

diff --git a/archaeological_operations/tests.py b/archaeological_operations/tests.py
index 57e71b256..688af87ad 100644
--- a/archaeological_operations/tests.py
+++ b/archaeological_operations/tests.py
@@ -16,12 +16,15 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 # See the file COPYING for details.
+
 from bs4 import BeautifulSoup
 import json
 import datetime
 # nosec: call an explicit bin for testing purpose
 from subprocess import Popen, PIPE  # nosec
 from io import BytesIO
+import os
+from PIL import Image
 import tempfile
 import locale
 from unittest.mock import patch
@@ -88,6 +91,7 @@ from ishtar_common.models import (
     Document,
     ValueFormater,
     Regexp,
+    MediaExporter
 )
 from ishtar_common.models_rest import ApiUser, ApiSearchModel, ApiSheetFilter
 from archaeological_files.models import File, FileType
@@ -3523,7 +3527,7 @@ class OperationSearchTest(TestCase, OperationInitTest, SearchText):
 
 class TestPermissionQuery:
     def setup_permission_queries(self, prefix, model_name, permissions,
-                                  perm_requests=None, create_profiles=True):
+                                 perm_requests=None, create_profiles=True):
         content_type = ContentType.objects.get(model=model_name)
         if not hasattr(self, "permission_queries"):
             self.permission_queries = {}
@@ -3994,6 +3998,155 @@ class OperationPermissionTest(TestCase, TestPermissionQuery, OperationInitTest):
         self.assertRedirects(response, "/")
 
 
+class OperationExportMediaTest(TestCase, TestPermissionQuery, OperationInitTest):
+    fixtures = FILE_FIXTURES
+
+    def setUp(self):
+        IshtarSiteProfile.objects.get_or_create(slug="default", active=True)
+        self.username, self.password, self.user = create_superuser()
+        self.alt_username, self.alt_password, self.alt_user = create_user()
+
+        self.orgas = self.create_orgas(self.user)
+        self.operations = self.create_operation(self.alt_user, self.orgas[0])
+        self.operation = self.operations[-1]
+        image_path = os.path.join(
+            settings.LIB_BASE_PATH, "ishtar_common", "tests", "test.png"
+        )
+        with open(image_path, "rb") as image:
+            document = models.Document.objects.create(
+                title="Test 01",
+                associated_file=SimpleUploadedFile("test.txt", b"no real content"),
+                image=SimpleUploadedFile(
+                    name="test.png",
+                    content=image.read(),
+                    content_type="image/png",
+                ),
+            )
+        self.documents = [document]
+        self.operation.documents.add(document)
+        model, __ = ImporterModel.objects.get_or_create(
+            klass="archaeological_operations.models.Operation"
+        )
+        self.exporter = MediaExporter.objects.create(
+            name="Operation media export",
+            slug="operation-media-export",
+            associated_model=model,
+            files_to_export="I",
+            cascade=False,
+            available=True,
+            query="",
+            thumbnail_for_images=False
+        )
+        self.export_url = reverse("export-media", kwargs={
+            "exporter": self.exporter.slug,
+            "item_pk": self.operation.pk
+        })
+
+    def test_permissions(self):
+        c = Client()
+
+        response = c.get(self.export_url)
+        redirect_url = f'/accounts/login/?next={self.export_url}'
+        self.assertRedirects(
+            response, redirect_url, status_code=302,
+            target_status_code=200, fetch_redirect_response=True)
+
+        c.login(username=self.username, password=self.password)
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 200)
+
+        c.login(username=self.alt_username, password=self.alt_password)
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 403)
+
+        # has the permission to export but... cannot view this item
+        self.exporter.users.add(self.alt_user.ishtaruser)
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 403)
+
+        self.setup_permission_queries(
+            "ope",
+            "operation",
+            permissions=["view_own_operation"],
+        )
+        UserProfile.objects.create(
+            profile_type=self.profile_types["ope_associated_items"],
+            person=self.alt_user.ishtaruser.person,
+            current=True,
+        )
+        self.alt_user.ishtaruser.generate_permission()
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 200)
+
+        # can view the item and no restrictions on users
+        self.exporter.users.remove(self.alt_user.ishtaruser)
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 200)
+
+        # can view the item, restrictions on users and not in users
+        self.exporter.users.add(self.user.ishtaruser)
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 403)
+
+        # can view the item, restrictions on profiles and not in profiles
+        self.exporter.users.remove(self.user.ishtaruser)
+        collaborator = ProfileType.objects.create(txt_idx="test-operation")
+        self.exporter.profile_types.add(collaborator)
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 403)
+
+        # can view the item, restrictions on profiles and in profiles
+        self.exporter.users.remove(self.user.ishtaruser)
+        self.exporter.profile_types.add(self.profile_types["ope_associated_items"])
+        response = c.get(self.export_url)
+        self.assertEqual(response.status_code, 200)
+
+    def _test_files(self, response, filenames):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            f, z = None, None
+            try:
+                f = BytesIO(response.content)
+                z = zipfile.ZipFile(f)
+                self.assertIsNone(z.testzip())
+                z.extractall(tmpdir)
+                name_list = z.namelist()
+                for filename, is_image in filenames:
+                    self.assertIn(filename, name_list)
+                    if not is_image:
+                        continue
+                    image = f"{tmpdir}/{filename}"
+                    with Image.open(image) as im:
+                        self.assertIsNone(im.verify())
+            finally:
+                if z:
+                    z.close()
+                if f:
+                    f.close()
+
+    def test_content(self):
+        c = Client()
+        c.login(username=self.username, password=self.password)
+
+        response = c.get(self.export_url)
+        self._test_files(response, [('image_00001.png', True)])
+
+        self.exporter.files_to_export = 'A'
+        self.exporter.save()
+        response = c.get(self.export_url)
+        self._test_files(
+            response,
+            [('image_00001.png', True), ('file_00001.txt', False), ]
+        )
+
+        self.exporter.thumbnail_for_images = True
+        self.exporter.save()
+        response = c.get(self.export_url)
+        self._test_files(
+            response,
+            [('image_00001.jpg', True), ('file_00001.txt', False), ]
+        )
+
+
 class LabelTest(TestCase, OperationInitTest):
     fixtures = FILE_FIXTURES