✨ permissions refactoring: settings for upstream items management

7 files changed, 182 insertions, 64 deletions

diff --git a/archaeological_finds/models_finds.py b/archaeological_finds/models_finds.py
index 035f7717d..8ea623d98 100644
--- a/archaeological_finds/models_finds.py
+++ b/archaeological_finds/models_finds.py
@@ -2017,7 +2017,6 @@ class Find(
         "weight_string",
     ]
     UPPER_PERMISSIONS = [
-        (Operation, "base_finds__context_record__operation_id"),
         (ContextRecord, "base_finds__context_record_id"),
         (("archaeological_warehouse", "Warehouse"), "container__location_id"),
         (("archaeological_warehouse", "Warehouse"), "container_ref__responsibility_id"),
diff --git a/archaeological_finds/models_treatments.py b/archaeological_finds/models_treatments.py
index 45dc26c16..a26df8b6a 100644
--- a/archaeological_finds/models_treatments.py
+++ b/archaeological_finds/models_treatments.py
@@ -207,6 +207,16 @@ class Treatment(
     ]
     PARENT_SEARCH_VECTORS = ["person", "organization"]
     GET_VALUES_M2M = ["treatment_types"]
+    UPPER_PERMISSIONS = [
+        (("archaeological_finds", "treatmentfile"),
+         "file_id"),
+        (("archaeological_finds", "find"),
+         "downstream__pk"),
+        (("archaeological_finds", "find"),
+         "upstream__pk"),
+        (("archaeological_finds", "find"),
+         "finds__pk"),
+    ]
     objects = ExternalIdManager()
 
     label = models.CharField(_("Label"), blank=True, null=True, max_length=200)
@@ -1150,6 +1160,10 @@ class TreatmentFile(
         "documents__associated_file__isnull",
         "documents__associated_url__isnull",
     ]
+    UPPER_PERMISSIONS = [
+        (("archaeological_finds", "find"),
+         "associated_basket__items__pk"),
+    ]
     # alternative names of fields for searches
     ALT_NAMES = {
         "name": SearchAltName(
diff --git a/archaeological_operations/models.py b/archaeological_operations/models.py
index dd265fb1d..3f4f252c3 100644
--- a/archaeological_operations/models.py
+++ b/archaeological_operations/models.py
@@ -447,6 +447,10 @@ class ArchaeologicalSite(
         "documents__associated_file__isnull",
         "documents__associated_url__isnull",
     ]
+    UPPER_PERMISSIONS = [
+        (("archaeological_operations", "operation"),
+         "operations__pk"),
+    ]
     ALT_NAMES = {
         "reference": SearchAltName(
             pgettext_lazy("key for text search", "reference"), "reference__iexact"
@@ -1321,6 +1325,10 @@ class Operation(
         "cached_remains",
         "cached_label",
     ]
+    UPPER_PERMISSIONS = [
+        (("archaeological_files", "file"),
+         "associated_file_id"),
+    ]
     objects = UUIDModelManager()
 
     # alternative names of fields for searches
@@ -2825,7 +2833,8 @@ post_save.connect(post_save_cache, sender=ActType)
 post_delete.connect(post_save_cache, sender=ActType)
 
 
-class AdministrativeAct(DocumentItem, BaseHistorizedItem, OwnPerms, ValueGetter, SheetItem):
+class AdministrativeAct(DocumentItem, BaseHistorizedItem, OwnPerms, ValueGetter,
+                        SheetItem):
     SHOW_URL = "show-administrativeact"
     TABLE_COLS = [
         "full_ref",
@@ -3068,6 +3077,16 @@ class AdministrativeAct(DocumentItem, BaseHistorizedItem, OwnPerms, ValueGetter,
         "operation__towns__numero_insee__startswith": "_get_department_code",
         "associated_file__towns__numero_insee__startswith": "_get_department_code",
     }
+    UPPER_PERMISSIONS = [
+        (("archaeological_operations", "operation"),
+         "operation_id"),
+        (("archaeological_files", "file"),
+         "associated_file_id"),
+        (("archaeological_finds", "treatmentfile"),
+         "treatment_file_id"),
+        (("archaeological_finds", "treatment"),
+         "treatment_id"),
+    ]
 
     # fields
     act_type = models.ForeignKey(
diff --git a/ishtar_common/admin.py b/ishtar_common/admin.py
index 465a9f152..91a036ad9 100644
--- a/ishtar_common/admin.py
+++ b/ishtar_common/admin.py
@@ -795,20 +795,33 @@ class PermissionRequestAdminForm(forms.ModelForm):
             klass = ct.model_class()
             if not klass:
                 continue
-            if getattr(klass, "SHOW_URL", None) and hasattr(klass, "history_creator_id"):
+            if (getattr(klass, "SHOW_URL", None) and
+                hasattr(klass, "history_creator_id")) or \
+                    ct.model in ("geovectordata", "document"):
                 choices.append((ct.pk, ct.name))
         self.fields["model"].choices = [("", "-" * 9)] + list(
-                sorted(choices, key=lambda x: x[1])
+            sorted(choices, key=lambda x: x[1])
         )
 
     def clean(self):
+        super().clean()
+        model = self.cleaned_data["model"]
+        include_upstream_items = self.cleaned_data.get("include_upstream_items", None)
+        if include_upstream_items and model.model in ("organization", "person", "file",
+                                                      "warehouse"):
+            raise forms.ValidationError(
+                _("This model do have attached upstream items. Uncheck "
+                  "\"Include upstream items\" field.")
+            )
         limit_to_attached_areas = self.cleaned_data.get("limit_to_attached_areas", "")
         if not limit_to_attached_areas:
             return self.cleaned_data
-        model = self.cleaned_data["model"]
         if model.model not in ("operation", "contextrecord", "find",
                                "archaeologicalsite", "file"):
-            raise forms.ValidationError(_("This model do not accept area limitation."))
+            raise forms.ValidationError(
+                _("This model do not accept area limitation. Uncheck "
+                  "\"Limit request to attached areas\" field.")
+            )
         return self.cleaned_data
 
 
diff --git a/ishtar_common/models.py b/ishtar_common/models.py
index 045bab1cc..eb9e73150 100644
--- a/ishtar_common/models.py
+++ b/ishtar_common/models.py
@@ -3607,12 +3607,14 @@ class UserProfile(models.Model):
         item_ids = []
         model_class = content_type.model_class()
         if permission_request.include_associated_items:
-            item_ids += model_class.objects.filter(
-                ishtar_users__pk=ishtar_user.pk
-            ).values_list("pk", flat=True)
-            item_ids += model_class.objects.filter(
-                history_creator_id=ishtar_user.pk
-            ).values_list("pk", flat=True)
+            if hasattr(model_class, "ishtar_users"):
+                item_ids += model_class.objects.filter(
+                    ishtar_users__pk=ishtar_user.pk
+                ).values_list("pk", flat=True)
+            if hasattr(model_class, "history_creator_id"):
+                item_ids += model_class.objects.filter(
+                    history_creator_id=ishtar_user.pk
+                ).values_list("pk", flat=True)
             if content_type.model == "find" and \
                     permission_type in ("view", "change"):
                 Find = apps.get_model("archaeological_finds", "Find")
@@ -4510,6 +4512,28 @@ class Document(
         "towns",
         "areas",
     ]
+    UPPER_PERMISSIONS = [
+        (("archaeological_files", "file"),
+         "files__pk"),
+        (("archaeological_operations", "operation"),
+         "operations__pk"),
+        (("archaeological_operations", "archaeologicalsite"),
+         "sites__pk"),
+        (("archaeological_operations", "administrativeact"),
+         "administrativeacts__pk"),
+        (("archaeological_context_records", "contextrecord"),
+         "context_records__pk"),
+        (("archaeological_finds", "find"),
+         "finds__pk"),
+        (("archaeological_finds", "treatmentfile"),
+         "treatment_files__pk"),
+        (("archaeological_finds", "treatment"),
+         "treatments__pk"),
+        (("archaeological_warehouse", "warehouse"),
+         "warehouses__pk"),
+        (("archaeological_warehouse", "container"),
+         "containers__pk"),
+    ]
     SLUG = "document"
     LINK_SPLIT = "<||>"
 
diff --git a/ishtar_common/models_common.py b/ishtar_common/models_common.py
index dc48fa9e5..aa52d82c7 100644
--- a/ishtar_common/models_common.py
+++ b/ishtar_common/models_common.py
@@ -22,7 +22,7 @@ from unidecode import unidecode
 from django import forms
 from django.apps import apps
 from django.conf import settings
-from django.contrib.auth.models import Permission, User
+from django.contrib.auth.models import User
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.contrib.gis.db import models
@@ -60,8 +60,6 @@ from simple_history.signals import (
     pre_create_historical_record,
 )
 
-from guardian.models import UserObjectPermission
-
 from ishtar_common.data_importer import post_importer_action, ImporterError
 from ishtar_common.model_managers import TypeManager
 from ishtar_common.model_merging import merge_model_objects
@@ -1528,7 +1526,7 @@ class BaseHistorizedItem(
     FullSearch,
     Imported,
     JsonData,
-    FixAssociated,
+    FixAssociated
 ):
     """
     Historized item with external ID management.
@@ -1541,7 +1539,6 @@ class BaseHistorizedItem(
     EXTERNAL_ID_KEY = ""
     EXTERNAL_ID_DEPENDENCIES = []
     HISTORICAL_M2M = []
-    UPPER_PERMISSIONS = []
 
     history_modifier = models.ForeignKey(
         User,
@@ -1614,52 +1611,6 @@ class BaseHistorizedItem(
     def get_verbose_name(cls):
         return cls._meta.verbose_name
 
-    @classmethod
-    def get_ids_from_upper_permissions(cls, user_id, base_permissions):
-        if not cls.UPPER_PERMISSIONS:
-            return []
-        ProfileType = apps.get_model("ishtar_common", "ProfileType")
-        item_ids = []
-        for model, attr in cls.UPPER_PERMISSIONS:
-            if isinstance(model, tuple):
-                app_label, model_name = model
-                model = apps.get_model(app_label, model_name)
-            permissions = list(set([
-                "_".join(permission.codename.split("_")[:-1])
-                + f"_{model._meta.model_name}"
-                for permission in base_permissions
-            ]))
-            q = ProfileType.objects.filter(
-                user_profiles__person__ishtaruser=user_id,
-                groups__permissions__codename__in=permissions
-            )
-            lst = []
-            if not q.count():
-                # no permissions associated for upstream model get direct attachement
-                lst = model.objects.filter(
-                    ishtar_users__pk=user_id
-                ).values_list("pk", flat=True)
-            else:
-                perms = []
-                for codename in permissions:
-                    perms += [
-                        perm
-                        for perm in Permission.objects.filter(
-                            codename=codename).all()
-                    ]
-                lst = []
-                for permission in perms:
-                    lst += list(
-                        UserObjectPermission.objects.filter(
-                            permission=permission,
-                            user_id=user_id
-                        ).values_list("object_pk", flat=True)
-                    )
-            item_ids += cls.objects.filter(
-                **{f"{attr}__in": lst}
-            ).values_list("pk", flat=True)
-        return list(set(item_ids))
-
     def is_locked(self, user=None):
         if not user:
             return self.locked
@@ -2317,6 +2268,21 @@ class GeoVectorData(Imported, OwnPerms):
         "related_items_archaeological_warehouse_container",
     ]
 
+    UPPER_PERMISSIONS = [
+        (("archaeological_operations", "operation"),
+         "related_items_archaeological_operations_operation__pk"),
+        (("archaeological_operations", "archaeologicalsite"),
+         "related_items_archaeological_operations_archaeologicalsite__pk"),
+        (("archaeological_context_records", "contextrecord"),
+         "related_items_archaeological_context_records_contextrecord__pk"),
+        (("archaeological_finds", "find"),
+         "related_items_archaeological_finds_basefind__find__pk"),
+        (("archaeological_warehouse", "warehouse"),
+         "related_items_archaeological_warehouse_warehouse__pk"),
+        (("archaeological_warehouse", "container"),
+         "related_items_archaeological_warehouse_container__pk"),
+    ]
+
     buffer = models.FloatField(
         _("Buffer"), blank=True, null=True
     )
@@ -3110,7 +3076,7 @@ class PermissionRequest(models.Model):
         help_text=_("All items associated items match the request")
     )
     include_upstream_items = models.BooleanField(
-        _("Include upstream items"), default=True,
+        _("Include upstream items"), default=False,
         help_text=_(
             "All items associated by upstream link math the request. "
             "For instance, match is done for all finds associated with own "
diff --git a/ishtar_common/utils.py b/ishtar_common/utils.py
index bbed4e8a4..5536bc84b 100644
--- a/ishtar_common/utils.py
+++ b/ishtar_common/utils.py
@@ -414,6 +414,89 @@ class OwnPerms:
     """
     Manage special permissions for object's owner
     """
+    UPPER_PERMISSIONS = []
+
+    @classmethod
+    def _has_permission_query_for_upper_permissions(
+            cls, base_permissions, model, user_id):
+        ProfileType = apps.get_model("ishtar_common", "ProfileType")
+        permissions = list(set([
+            "_".join(permission.codename.split("_")[:-1])
+            + f"_{model._meta.model_name}"
+            for permission in base_permissions
+        ]))
+        q = ProfileType.objects.filter(
+            user_profiles__person__ishtaruser=user_id,
+            groups__permissions__codename__in=permissions
+        )
+        return q, permissions
+
+    @classmethod
+    def get_ids_from_upper_permissions(cls, user_id, base_permissions):
+        if not cls.UPPER_PERMISSIONS:
+            return []
+        UserObjectPermission = apps.get_model("guardian", "UserObjectPermission")
+        item_ids = []
+        full_permissions = []
+        for base_permission in base_permissions:
+            if "_own_" not in base_permission.codename:
+                full_permissions.append(base_permission)
+                continue
+            codename = base_permission.codename.replace("_own", "")
+            try:
+                full_permissions.append(
+                    Permission.objects.get(
+                        codename=codename,
+                        content_type=base_permission.content_type
+                    )
+                )
+            except Permission.DoesNotExist:
+                continue
+        for model, attr in cls.UPPER_PERMISSIONS:
+            if isinstance(model, tuple):
+                app_label, model_name = model
+                model = apps.get_model(app_label, model_name)
+
+            # check if has full permission
+            q_full, __ = cls._has_permission_query_for_upper_permissions(
+                full_permissions, model, user_id
+            )
+            has_full_permission = bool(q_full.count())
+            if has_full_permission:
+                item_ids += cls.objects.filter(
+                    **{f"{attr}__isnull": False}
+                ).values_list("pk", flat=True)
+                continue
+
+            q, permissions = cls._has_permission_query_for_upper_permissions(
+                base_permissions, model, user_id
+            )
+            lst = []
+            if not q.count():
+                # no permissions associated for upstream model get direct attachement
+                lst = model.objects.filter(
+                    ishtar_users__pk=user_id
+                ).values_list("pk", flat=True)
+            else:
+                perms = []
+                for codename in permissions:
+                    perms += [
+                        perm
+                        for perm in Permission.objects.filter(
+                            codename=codename).all()
+                    ]
+                lst = []
+                for permission in perms:
+                    lst += list(
+                        UserObjectPermission.objects.filter(
+                            permission=permission,
+                            user_id=user_id
+                        ).values_list("object_pk", flat=True)
+                    )
+            item_ids += cls.objects.filter(
+                **{f"{attr}__in": lst}
+            ).values_list("pk", flat=True)
+        return list(set(item_ids))
 
     @classmethod
     def get_query_owns(cls, ishtaruser):