Serialization - Import/Export: manage put an release of locks

2 files changed, 82 insertions, 5 deletions

diff --git a/archaeological_operations/serializers.py b/archaeological_operations/serializers.py
index 6fd0f10d6..91b0b1b0f 100644
--- a/archaeological_operations/serializers.py
+++ b/archaeological_operations/serializers.py
@@ -57,7 +57,8 @@ def operation_serialization(archive=False, return_empty_types=False,
                             archive_name=None, operation_queryset=None,
                             site_queryset=None, cr_queryset=None,
                             find_queryset=None, warehouse_queryset=None,
-                            get_queryset=False, no_geo=True):
+                            get_queryset=False, no_geo=True,
+                            put_locks=False, lock_user=None):
     result_queryset = {}
     if operation_queryset:
         operation_ids = operation_queryset.values_list("id", flat=True)
@@ -124,6 +125,16 @@ def operation_serialization(archive=False, return_empty_types=False,
 
     result = generic_get_results(OPERATION_MODEL_LIST, "operations",
                                  result_queryset=result_queryset, no_geo=no_geo)
+
+    if put_locks:
+        for model in OPERATION_MODEL_LIST:
+            if not hasattr(model, "locked"):
+                continue
+            q = model.objects
+            if result_queryset and model.__name__ in result_queryset:
+                q = result_queryset[model.__name__]
+            q.update(locked=True, lock_user=lock_user)
+
     full_archive = archive_serialization(
         result, archive_dir="operations", archive=archive,
         return_empty_types=return_empty_types, archive_name=archive_name,
diff --git a/archaeological_operations/tests.py b/archaeological_operations/tests.py
index 72cecee95..6c20df85a 100644
--- a/archaeological_operations/tests.py
+++ b/archaeological_operations/tests.py
@@ -753,6 +753,7 @@ class SerializationTest(GenericSerializationTest, TestCase):
         operation = create_operation(self.user,
                                      values={"code_patriarche": "66666"})
         ope2 = create_operation(self.user, values={"code_patriarche": "66667"})
+        self.operations = [operation, ope2]
         models.RecordRelations.objects.create(
             left_record=operation, right_record=ope2,
             relation_type=models.RelationType.objects.all()[0]
@@ -815,6 +816,41 @@ class SerializationTest(GenericSerializationTest, TestCase):
         )
         self.assertEqual(len(rel_json), 0)
 
+    def reinit_lock(self):
+        for operation in self.operations:
+            operation = models.Operation.objects.get(pk=operation.pk)
+            operation.locked = False
+            operation.lock_user = None
+            operation.save()
+
+    def test_lock(self):
+        self.reinit_lock()
+        self.generic_serialization_test(
+            serializers.operation_serialization, no_test=True,
+            kwargs={"put_locks": False})
+        for operation in self.operations:
+            operation = models.Operation.objects.get(pk=operation.pk)
+            self.assertFalse(operation.locked)
+            self.assertIsNone(operation.lock_user)
+
+        self.reinit_lock()
+        self.generic_serialization_test(
+            serializers.operation_serialization, no_test=True,
+            kwargs={"put_locks": True})
+        for operation in self.operations:
+            operation = models.Operation.objects.get(pk=operation.pk)
+            self.assertTrue(operation.locked)
+            self.assertIsNone(operation.lock_user)
+
+        self.reinit_lock()
+        self.generic_serialization_test(
+            serializers.operation_serialization, no_test=True,
+            kwargs={"put_locks": True, "lock_user": self.user})
+        for operation in self.operations:
+            operation = models.Operation.objects.get(pk=operation.pk)
+            self.assertTrue(operation.locked)
+            self.assertEqual(operation.lock_user, self.user)
+
     def test_restore(self):
         current_number, zip_filename = self.generic_restore_test_genzip(
             serializers.OPERATION_MODEL_LIST,
@@ -822,6 +858,31 @@ class SerializationTest(GenericSerializationTest, TestCase):
         self.generic_restore_test(zip_filename, current_number,
                                   serializers.OPERATION_MODEL_LIST)
 
+    def test_unlock_on_restore(self):
+        current_number, zip_filename = self.generic_restore_test_genzip(
+            serializers.OPERATION_MODEL_LIST,
+            serializers.operation_serialization,
+            kwargs={"put_locks": True, "lock_user": self.user})
+
+        self.generic_restore_test(zip_filename, current_number,
+                                  serializers.OPERATION_MODEL_LIST,
+                                  delete_existing=False)
+        for operation in self.operations:
+            operation = models.Operation.objects.get(
+                code_patriarche=operation.code_patriarche)
+            self.assertTrue(operation.locked)
+            self.assertEqual(operation.lock_user, self.user)
+
+        self.generic_restore_test(zip_filename, current_number,
+                                  serializers.OPERATION_MODEL_LIST,
+                                  delete_existing=False,
+                                  release_locks=True)
+        for operation in self.operations:
+            operation = models.Operation.objects.get(
+                code_patriarche=operation.code_patriarche)
+            self.assertFalse(operation.locked)
+            self.assertIsNone(operation.lock_user)
+
     def test_historization_on_restore(self):
         current_number, zip_filename = self.generic_restore_test_genzip(
             serializers.OPERATION_MODEL_LIST,
@@ -1620,18 +1681,23 @@ class LockTest(TestCase, OperationInitTest):
         step = 'selec-operation_modification'
         response = cls_wiz.wizard_post(
             self.client, url, step, {'pk': self.operation.pk})
-        self.assertIn("locked for edition", response.content.decode(),
+        msg = str(_("This operation is locked for edition.")
+                  ).replace("'", "'")
+        self.assertIn(msg, response.content.decode(),
                       msg="wizard lock for edition not effective")
 
     def test_qa_lock(self):
         url = reverse('operation-qa-bulk-update', args=[self.operation.pk])
         response = self.client.get(url)
-        self.assertRedirects(response, reverse('qa-not-available'))
+        self.assertRedirects(response, reverse('qa-not-available',
+                                               args=["locked"]))
 
     def test_sheet_lock(self):
         url = reverse('show-operation', kwargs={'pk': self.operation.pk})
         response = self.client.get(url)
-        self.assertIn('This item has been locked', response.content.decode(),
+        msg = str(_("This item has been locked. Edition is disabled.")
+                  ).replace("'", "\'")
+        self.assertIn(msg, response.content.decode(),
                       msg="lock not displayed on sheet")
 
 
@@ -3279,7 +3345,7 @@ class OperationQATest(OperationInitTest, TestCase):
             reverse('operation-qa-bulk-update-confirm', args=[pks]),
             {'qa_operation_type': operation_type.pk}
         )
-        self.assertRedirects(response, '/qa-not-available/')
+        self.assertRedirects(response, '/qa-not-available/locked/')
 
 
 class DocumentQATest(OperationInitTest, TestCase):