Manage own permissions with areas for context records (refs #4060)

5 files changed, 138 insertions, 15 deletions

diff --git a/archaeological_context_records/models.py b/archaeological_context_records/models.py
index 90c3d5563..996304231 100644
--- a/archaeological_context_records/models.py
+++ b/archaeological_context_records/models.py
@@ -429,10 +429,16 @@ class ContextRecord(BulkUpdatedItem, BaseHistorizedItem,
 
     @classmethod
     def get_query_owns(cls, ishtaruser):
+        profile = ishtaruser.current_profile
+        town_ids = []
+        if profile:
+            town_ids = [town['pk']
+                        for town in profile.query_towns.values('pk').all()]
         return (Q(operation__scientist=ishtaruser.person) |
                 Q(operation__in_charge=ishtaruser.person) |
                 Q(operation__collaborators__pk=ishtaruser.person.pk) |
-                Q(history_creator=ishtaruser.user_ptr)) \
+                Q(history_creator=ishtaruser.user_ptr) |
+                Q(operation__towns__pk__in=town_ids)) \
             & Q(operation__end_date__isnull=True)
 
     @classmethod
diff --git a/archaeological_context_records/tests.py b/archaeological_context_records/tests.py
index 78d21dd90..aad8b8b6d 100644
--- a/archaeological_context_records/tests.py
+++ b/archaeological_context_records/tests.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
-# Copyright (C) 2015-2017 Étienne Loks  <etienne.loks_AT_peacefrogsDOTnet>
+# Copyright (C) 2015-2018 Étienne Loks  <etienne.loks_AT_peacefrogsDOTnet>
 
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -22,11 +22,13 @@ import json
 from StringIO import StringIO
 
 from django.conf import settings
+from django.contrib.auth.models import Permission
 from django.core.exceptions import ValidationError, ImproperlyConfigured
 from django.core.urlresolvers import reverse
 from django.test.client import Client
 
-from ishtar_common.models import IshtarSiteProfile, ImporterModel
+from ishtar_common.models import IshtarSiteProfile, ImporterModel, \
+    UserProfile, ProfileType, Town, Area
 
 from archaeological_operations.tests import OperationInitTest, \
     ImportTest, FILE_TOWNS_FIXTURES, FILE_FIXTURES, OPERATION_TOWNS_FIXTURES
@@ -34,7 +36,7 @@ from archaeological_operations import models as models_ope
 from archaeological_context_records import models
 
 from ishtar_common.tests import WizardTest, WizardTestFormData as FormData, \
-    create_superuser, TestCase
+    create_superuser, create_user, TestCase
 
 from archaeological_context_records import views
 
@@ -167,19 +169,26 @@ class ImportContextRecordTest(ImportTest, TestCase):
 
 
 class ContextRecordInit(OperationInitTest):
-    def create_context_record(self, user=None, data={}, force=False):
+    def create_context_record(self, data=None, user=None, force=False):
+        if not data:
+            data = {}
+
         if not getattr(self, 'context_records', None):
             self.context_records = []
+
         default = {'label': "Context record"}
         if force or not data.get('operation') \
                 or not models.Operation.objects.filter(
                     pk=data['operation'].pk).count():
-            data['operation'] = self.get_default_operation(force=force)
+            data['operation'] = self.get_default_operation(force=force,
+                                                           user=user)
         if not data.get('parcel') or not data['parcel'].pk \
             or not models.Parcel.objects.filter(
                 pk=data['parcel'].pk).count():
             data['parcel'] = self.get_default_parcel(force=force)
-        if not data.get('history_modifier'):
+        if user:
+            data['history_modifier'] = user
+        elif not data.get('history_modifier'):
             data['history_modifier'] = self.get_default_user()
 
         default.update(data)
@@ -580,6 +589,102 @@ class ContextRecordSearchTest(ContextRecordInit, TestCase):
         self.assertEqual(json.loads(response.content)['recordsTotal'], 1)
 
 
+class ContextRecordPermissionTest(ContextRecordInit, TestCase):
+    fixtures = CONTEXT_RECORD_TOWNS_FIXTURES
+
+    def setUp(self):
+        IshtarSiteProfile.objects.create()
+        self.username, self.password, self.user = create_superuser()
+        self.alt_username, self.alt_password, self.alt_user = create_user()
+        self.alt_user.user_permissions.add(Permission.objects.get(
+            codename='view_own_contextrecord'))
+        self.alt_user.user_permissions.add(Permission.objects.get(
+            codename='change_own_contextrecord'))
+        self.alt_username2, self.alt_password2, self.alt_user2 = create_user(
+            username='luke', password='iamyourfather'
+        )
+        profile = UserProfile.objects.create(
+            profile_type=ProfileType.objects.get(txt_idx='collaborator'),
+            person=self.alt_user2.ishtaruser.person,
+            current=True
+        )
+
+        town = Town.objects.create(name='Tatouine', numero_insee='66000')
+        area = Area.objects.create(label='Galaxie', txt_idx='galaxie')
+        area.towns.add(town)
+
+        self.orgas = self.create_orgas(self.user)
+        self.operations = self.create_operation(self.user, self.orgas[0])
+        self.operations += self.create_operation(self.alt_user, self.orgas[0])
+        self.operations[1].towns.add(town)
+
+        self.create_context_record(user=self.user, data={"label": u"CR 1",
+                                         "operation": self.operations[0]})
+        self.create_context_record(
+            user=self.alt_user,
+            data={"label": u"CR 2", "operation": self.operations[1]})
+        self.cr_1 = self.context_records[0]
+        self.cr_2 = self.context_records[1]
+
+
+        profile.areas.add(area)
+
+    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))
+
+        # possession
+        c = Client()
+        c.login(username=self.alt_username, password=self.alt_password)
+        response = c.get(reverse('get-contextrecord'))
+        # only one "own" context record available
+        self.assertTrue(json.loads(response.content))
+        self.assertEqual(json.loads(response.content)['recordsTotal'], 1)
+
+        # area filter
+        c = Client()
+        c.login(username=self.alt_username2, password=self.alt_password2)
+        response = c.get(reverse('get-contextrecord'))
+        # only one "own" operation available
+        self.assertTrue(json.loads(response.content))
+        self.assertEqual(json.loads(response.content)['recordsTotal'], 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'
+
+        # possession
+        c = Client()
+        c.login(username=self.alt_username, password=self.alt_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()
+        c.login(username=self.alt_username2, password=self.alt_password2)
+        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, "/")
+
+
 class RecordRelationsTest(ContextRecordInit, TestCase):
     fixtures = OPERATION_TOWNS_FIXTURES
     model = models.ContextRecord
diff --git a/archaeological_context_records/views.py b/archaeological_context_records/views.py
index 898610007..4f4b06993 100644
--- a/archaeological_context_records/views.py
+++ b/archaeological_context_records/views.py
@@ -21,7 +21,7 @@ import json
 
 from django.core.urlresolvers import reverse
 from django.db.models import Q
-from django.http import HttpResponse
+from django.http import HttpResponse, HttpResponseRedirect
 from django.shortcuts import redirect
 from django.utils.translation import ugettext_lazy as _
 
@@ -29,6 +29,9 @@ from ishtar_common.forms_common import AuthorFormset, SourceForm
 from ishtar_common.views import get_item, show_item, revert_item
 from archaeological_operations.views import site_extra_context
 from ishtar_common.wizards import SearchWizard
+
+from ishtar_common.utils import put_session_message
+
 from wizards import *
 from forms import *
 import models
@@ -122,7 +125,16 @@ record_modification_wizard = RecordModifWizard.as_view([
 
 
 def record_modify(request, pk):
-    record_modification_wizard(request)
+    try:
+        record_modification_wizard(request)
+    except IndexError:  # no step available
+        put_session_message(
+            request.session.session_key,
+            _(u"You don't have sufficient permissions to do this action."),
+            'warning'
+        )
+        return HttpResponseRedirect("/")
+
     RecordModifWizard.session_set_value(
         request, 'selec-record_modification', 'pk', pk, reset=True)
     return redirect(reverse('record_modification',
diff --git a/archaeological_operations/tests.py b/archaeological_operations/tests.py
index 299ed49f8..6a96ebdf4 100644
--- a/archaeological_operations/tests.py
+++ b/archaeological_operations/tests.py
@@ -893,14 +893,14 @@ class OperationInitTest(object):
         self.operations.append(create_operation(user, orga))
         return self.operations
 
-    def get_default_operation(self, force=False):
+    def get_default_operation(self, force=False, user=None):
         if force:
-            return self.create_operation()[-1]
-        ope = self.create_operation()[0]
+            return self.create_operation(user=user)[-1]
+        ope = self.create_operation(user=user)[0]
         if models.Operation.objects.filter(pk=ope.pk).count():
             return ope
         self.operations.pop(0)
-        return self.create_operation()[-1]
+        return self.create_operation(user=user)[-1]
 
     def tearDown(self):
         # cleanup for further test
@@ -1614,7 +1614,6 @@ class OperationPermissionTest(TestCase, OperationInitTest):
         self.assertRedirects(response, "/")
 
 
-
 class DashboardTest(TestCase, OperationInitTest):
     fixtures = FILE_FIXTURES
 
diff --git a/ishtar_common/wizards.py b/ishtar_common/wizards.py
index 9e77a0dda..8e7cd0c74 100644
--- a/ishtar_common/wizards.py
+++ b/ishtar_common/wizards.py
@@ -178,10 +178,10 @@ class Wizard(NamedUrlWizardView):
 
     def dispatch(self, request, *args, **kwargs):
         self.current_right = kwargs.get('current_right', None)
-
         step = kwargs.get('step', None)
         # check that the current object is really owned by the current user
         if step and self.current_right and '_own_' in self.current_right:
+
             # reinit default dispatch of a wizard - not clean...
             self.request = request
             self.session = request.session
@@ -194,6 +194,7 @@ class Wizard(NamedUrlWizardView):
             current_object = self.get_current_object()
             ishtaruser = request.user.ishtaruser \
                 if hasattr(request.user, 'ishtaruser') else None
+
             # not the fisrt step and current object is not owned
             if self.steps and self.steps.first != step and\
                     current_object and not current_object.is_own(ishtaruser):