✨ generate_permissions

manage:
- possession (direct, creation, basket)
- heritage
- areas association
- requests ({USER} special syntax)

2 files changed, 277 insertions, 5 deletions

diff --git a/archaeological_context_records/models.py b/archaeological_context_records/models.py
index a88a7d161..1bb09717d 100644
--- a/archaeological_context_records/models.py
+++ b/archaeological_context_records/models.py
@@ -25,7 +25,7 @@ from django.conf import settings
 from django.contrib.gis.db import models
 from django.contrib.postgres.indexes import GinIndex
 from django.contrib.sites.models import Site
-from django.db import connection, transaction, OperationalError, IntegrityError
+from django.db import transaction, OperationalError, IntegrityError
 from django.db.models import Q
 from django.db.models.signals import post_delete, post_save, m2m_changed
 from django.urls import reverse
@@ -37,7 +37,7 @@ from ishtar_common.utils import (
     cached_label_changed,
     m2m_historization_changed,
     post_save_geo,
-    task,
+    SearchAltName,
 )
 
 from ishtar_common.models import (
@@ -58,7 +58,6 @@ from ishtar_common.models import (
     get_current_profile,
     document_attached_changed,
     HistoryModel,
-    SearchAltName,
     GeoItem,
     CompleteIdentifierItem,
     SearchVectorConfig,
@@ -826,6 +825,7 @@ class ContextRecord(
         "town_label_with_areas",
     ]
     UPPER_GEO = ["operation", "archaeological_site"]
+    UPPER_PERMISSIONS = [(Operation, "operation_id")]
 
     history = HistoricalRecords(bases=[HistoryModel])
     objects = UUIDModelManager()
@@ -1214,6 +1214,10 @@ class ContextRecord(
         return actions
 
     @classmethod
+    def get_limit_to_area_query(cls, town_ids):
+        return Q(operation__towns__pk__in=town_ids)
+
+    @classmethod
     def get_query_owns(cls, ishtaruser):
         return (
             cls._construct_query_own(
diff --git a/archaeological_context_records/tests.py b/archaeological_context_records/tests.py
index 8a16facf5..f550d23ce 100644
--- a/archaeological_context_records/tests.py
+++ b/archaeological_context_records/tests.py
@@ -44,7 +44,8 @@ from ishtar_common.models import (
 )
 
 from ishtar_common import forms_common
-from archaeological_operations.tests import OperationInitTest, ImportTest
+from archaeological_operations.tests import OperationInitTest, ImportTest, \
+                                            TestPermissionRequest
 from archaeological_operations import models as models_ope
 from archaeological_operations.views import RELATION_FORMSET_EXTRA_FORM
 from archaeological_context_records import models
@@ -942,10 +943,11 @@ class ContextRecordSearchTest(ContextRecordInit, TestCase, SearchText):
         self._test_search(c, result, context="Text period search")
 
 
-class ContextRecordPermissionTest(ContextRecordInit, TestCase):
+class ContextRecordOldPermissionTest(ContextRecordInit, TestCase):
     fixtures = CONTEXT_RECORD_TOWNS_FIXTURES
 
     def setUp(self):
+        print("Theses tests should fail on v5")
         profile_type = ProfileType.objects.create(
             label="xxCollaborateur",
             txt_idx="xxcollaborator",
@@ -1056,6 +1058,272 @@ class ContextRecordPermissionTest(ContextRecordInit, TestCase):
         self.assertRedirects(response, "/")
 
 
+class ContextRecordPermissionTest(ContextRecordInit, TestPermissionRequest,
+                                  TestCase):
+    fixtures = CONTEXT_RECORD_TOWNS_FIXTURES
+
+    def setUp(self):
+        IshtarSiteProfile.objects.create()
+
+        self.setup_permission_requests(
+            "ope",
+            "operation",
+            permissions=["view_own_operation", "change_own_operation"],
+            create_profiles=False
+        )
+        self.setup_permission_requests(
+            "cr",
+            "contextrecord",
+            permissions=["view_own_contextrecord", "change_own_contextrecord"],
+            perm_requests=['id="new-*"', 'excavator="{USER}"']
+        )
+
+        self.users = {}
+        username, password, user = create_superuser()
+        self.users["superuser"] = (username, password, user)
+
+        upstream_username, upstream_password, upstream_user = create_user()
+        UserProfile.objects.create(
+            profile_type=self.profile_types["cr_upstream"],
+            person=upstream_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["upstream"] = (upstream_username, upstream_password, upstream_user)
+
+        # nosec: hard coded password for test purposes
+        associated_username, associated_password, associated_user = create_user(  # nosec
+            username="vador", password="darth"
+        )
+        profile = UserProfile.objects.create(
+            profile_type=self.profile_types["cr_associated_items"],
+            person=associated_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["associated"] = (
+            associated_username, associated_password, associated_user
+        )
+
+        # nosec: hard coded password for test purposes
+        areas_username, areas_password, areas_user = create_user(  # nosec
+            username="luke", password="iamyourfather"
+        )
+        profile = UserProfile.objects.create(
+            profile_type=self.profile_types["cr_areas"],
+            person=areas_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["areas"] = (
+            areas_username, areas_password, areas_user
+        )
+
+        town = Town.objects.create(name="Tatouine", numero_insee="66000")
+        area = Area.objects.create(label="Galaxie", txt_idx="galaxie")
+        area.towns.add(town)
+        profile.areas.add(area)
+
+        # nosec: hard coded password for test purposes
+        simple_up_username, simple_up_password, simple_up_user = create_user(  # nosec
+            username="r2d2", password="bipbip"
+        )
+        UserProfile.objects.create(
+            profile_type=self.profile_types["cr_upstream"],
+            person=simple_up_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["simple_upstream"] = (
+            simple_up_username, simple_up_password, simple_up_user
+        )
+
+        # nosec: hard coded password for test purposes
+        request_username, request_password, request_user = create_user(  # nosec
+            username="c6po", password="bopbop"
+        )
+        UserProfile.objects.create(
+            profile_type=self.profile_types["cr_request_1"],
+            person=request_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["request"] = (
+            request_username, request_password, request_user
+        )
+
+        # nosec: hard coded password for test purposes
+        request2_username, request2_password, request2_user = create_user(  # nosec
+            username="cowboy", password="bebop"
+        )
+        UserProfile.objects.create(
+            profile_type=self.profile_types["cr_request_2"],
+            person=request2_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["request2"] = (
+            request2_username, request2_password, request2_user
+        )
+
+        # nosec: hard coded password for test purposes
+        req_area_username, req_area_password, req_area_user = create_user(  # nosec
+            username="chewee", password="bwawa"
+        )
+        profile = UserProfile.objects.create(
+            profile_type=self.profile_types["cr_request_areas_1"],
+            person=req_area_user.ishtaruser.person,
+            current=True,
+        )
+        self.users["request_areas"] = (
+            req_area_username, req_area_password, req_area_user
+        )
+        profile.areas.add(area)
+
+        self.orgas = self.create_orgas(user)
+        self.operations = self.create_operation(user, self.orgas[0],
+                                                values={"code_patriarche": "OPE01"})
+        self.operations += self.create_operation(upstream_user, self.orgas[0],
+                                                 values={"code_patriarche": "OPE02"})
+        self.operations[0].towns.add(town)
+
+        self.operations[1].ishtar_users.add(simple_up_user.ishtaruser)
+
+        self.create_context_record(
+            user=user, data={"label": "new-CR1", "operation": self.operations[0]}
+        )
+        self.create_context_record(
+            user=user, data={"label": "new-CR2", "operation": self.operations[1]}
+        )
+        self.create_context_record(
+            user=associated_user,
+            data={"label": "old-CR3", "operation": self.operations[0]}
+        )
+        self.cr_1 = self.context_records[0]
+        self.cr_2 = self.context_records[1]
+        self.cr_2.ishtar_users.add(associated_user.ishtaruser)
+
+        associated_user.ishtaruser.generate_permission()
+        areas_user.ishtaruser.generate_permission()
+        simple_up_user.ishtaruser.generate_permission()
+        request_user.ishtaruser.generate_permission()
+        req_area_user.ishtaruser.generate_permission()
+        request2_user.ishtaruser.generate_permission()
+
+        # upstream with associated request for operation
+        gp = Group.objects.get(name="ope_xxx")
+        self.profile_types["cr_upstream"].groups.add(gp)
+        self.profile_types["cr_upstream"].permission_requests.add(
+            self.permission_requests["ope_associated_items"]
+        )
+        upstream_user.ishtaruser.generate_permission()
+
+    def test_own_search(self):
+        # no result when no authentification
+        c = Client()
+        response = c.get(reverse("get-contextrecord"))
+        self.assertTrue(not json.loads(response.content.decode()))
+
+        url = reverse("get-contextrecord")
+        # possession + direct "history_creator_id"
+        # two context record available via operation
+        self._test_search(
+            url,
+            'possession + direct "history_creator_id"',
+            self.users["associated"],
+            2
+        )
+
+        # upstream + direct "history_creator_id"
+        # one context record available via operation - can view operation
+        self._test_search(
+            url,
+            'upstream + direct "history_creator_id"',
+            self.users["upstream"],
+            1
+        )
+
+        # upstream: no upstream permission only via direct association
+        # one context record available via operation
+        self._test_search(
+            url,
+            'upstream: no upstream permission only via direct association',
+            self.users["simple_upstream"],
+            1
+        )
+
+        # area filter
+        # only one "own" operation available
+        self._test_search(
+            url,
+            'areas filter',
+            self.users["areas"],
+            2
+        )
+
+        # request
+        # 2 context record available via request
+        self._test_search(
+            url,
+            'request',
+            self.users["request"],
+            2
+        )
+
+        # request limited by area
+        # 2 context record available via request but only one match the area
+        self._test_search(
+            url,
+            'request limited by area',
+            self.users["request_areas"],
+            1
+        )
+        # request with {USER}
+        # one context record available via request
+        self._test_search(
+            url,
+            'request2',
+            self.users["request2"],
+            0
+        )
+        __, __, request2_user = self.users["request2"]
+        cr2 = self.context_records[2]
+        cr2.excavator = request2_user.ishtaruser.person
+        cr2.save()
+        request2_user.ishtaruser.generate_permission()
+        self._test_search(
+            url,
+            'request2',
+            self.users["request2"],
+            1
+        )
+
+    def test_own_modify(self):
+        # no result when no authentification
+        c = Client()
+        response = c.get(reverse("record_modify", args=[self.cr_2.pk]))
+        self.assertRedirects(response, "/")
+
+        modif_url = "/record_modification/operation-record_modification"
+
+        # upstream
+        c = Client()
+        upstream_username, upstream_password, upstream_user = self.users["upstream"]
+        c.login(username=upstream_username, password=upstream_password)
+        response = c.get(reverse("record_modify", args=[self.cr_2.pk]), follow=True)
+        self.assertRedirects(response, modif_url)
+        response = c.get(modif_url)
+
+        self.assertEqual(response.status_code, 200)
+        response = c.get(reverse("record_modify", args=[self.cr_1.pk]), follow=True)
+        self.assertRedirects(response, "/")
+
+        # area filter
+        c = Client()
+        areas_username, areas_password, areas_user = self.users["areas"]
+        c.login(username=areas_username, password=areas_password)
+        response = c.get(reverse("record_modify", args=[self.cr_1.pk]), follow=True)
+        self.assertRedirects(response, modif_url)
+        response = c.get(modif_url)
+        self.assertEqual(response.status_code, 200)
+        response = c.get(reverse("record_modify", args=[self.cr_2.pk]), follow=True)
+        self.assertRedirects(response, "/")
+
+
 class RecordRelationsTest(ContextRecordInit, TestCase):
     fixtures = OPERATION_TOWNS_FIXTURES
     model = models.ContextRecord