Further tests

* restructures compensation/tests into subtests for ecoaccount and compensation
* adds tests for ema workflow
* improved test data setup

compensation/tests/compensation/__init__.py

@@ -0,0 +1,7 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 07.02.22
+
+"""

compensation/tests/compensation/test_views.py

@@ -2,11 +2,11 @@
 Author: Michel Peltriaux
 Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
 Contact: michel.peltriaux@sgdnord.rlp.de
-Created on: 27.10.21
+Created on: 07.02.22
 
 """
+from django.test.client import Client
 from django.urls import reverse
-from django.test import Client
 
 from konova.settings import DEFAULT_GROUP
 from konova.tests.test_views import BaseViewTestCase
@@ -223,184 +223,3 @@ class CompensationViewTestCase(BaseViewTestCase):
         self.assert_url_fail(client, fail_urls)
         self.assert_url_success(client, success_urls)
 
-
-class EcoAccountViewTestCase(CompensationViewTestCase):
-    """
-    These tests focus on proper returned views depending on the user's groups privileges and login status
-
-    EcoAccounts can inherit the same tests used for compensations.
-
-    """
-    comp_state = None
-    comp_action = None
-
-    @classmethod
-    def setUpTestData(cls) -> None:
-        super().setUpTestData()
-        state = cls.create_dummy_states()
-        cls.eco_account.before_states.set([state])
-        cls.eco_account.after_states.set([state])
-
-        action = cls.create_dummy_action()
-        cls.eco_account.actions.set([action])
-
-        # Prepare urls
-        cls.index_url = reverse("compensation:acc:index", args=())
-        cls.new_url = reverse("compensation:acc:new", args=())
-        cls.new_id_url = reverse("compensation:acc:new-id", args=())
-        cls.detail_url = reverse("compensation:acc:detail", args=(cls.eco_account.id,))
-        cls.log_url = reverse("compensation:acc:log", args=(cls.eco_account.id,))
-        cls.edit_url = reverse("compensation:acc:edit", args=(cls.eco_account.id,))
-        cls.remove_url = reverse("compensation:acc:remove", args=(cls.eco_account.id,))
-        cls.report_url = reverse("compensation:acc:report", args=(cls.eco_account.id,))
-        cls.state_new_url = reverse("compensation:acc:new-state", args=(cls.eco_account.id,))
-        cls.action_new_url = reverse("compensation:acc:new-action", args=(cls.eco_account.id,))
-        cls.deadline_new_url = reverse("compensation:acc:new-deadline", args=(cls.eco_account.id,))
-        cls.new_doc_url = reverse("compensation:acc:new-doc", args=(cls.eco_account.id,))
-        cls.state_remove_url = reverse("compensation:acc:state-remove", args=(cls.eco_account.id, cls.comp_state.id,))
-        cls.action_remove_url = reverse("compensation:acc:action-remove", args=(cls.eco_account.id, cls.comp_action.id,))
-
-    def test_logged_in_no_groups_shared(self):
-        """ Check correct status code for all requests
-
-        Assumption: User logged in and has no groups and data is shared
-
-        Returns:
-
-        """
-        client = Client()
-        client.login(username=self.superuser.username, password=self.superuser_pw)
-        self.superuser.groups.set([])
-        self.eco_account.share_with_list([self.superuser])
-
-        # Since the user has no groups, it does not matter that data has been shared. There SHOULD not be any difference
-        # to a user without access, since the important permissions are missing
-        success_urls = [
-            self.index_url,
-            self.detail_url,
-            self.report_url,
-        ]
-        fail_urls = [
-            self.new_url,
-            self.new_id_url,
-            self.log_url,
-            self.edit_url,
-            self.remove_url,
-            self.state_new_url,
-            self.action_new_url,
-            self.deadline_new_url,
-            self.state_remove_url,
-            self.action_remove_url,
-            self.new_doc_url,
-        ]
-
-        self.assert_url_success(client, success_urls)
-        self.assert_url_fail(client, fail_urls)
-
-    def test_logged_in_no_groups_unshared(self):
-        """ Check correct status code for all requests
-
-        Assumption: User logged in and has no groups and data is shared
-
-        Returns:
-
-        """
-        client = Client()
-        client.login(username=self.superuser.username, password=self.superuser_pw)
-        self.superuser.groups.set([])
-        self.eco_account.share_with_list([])
-
-        # Since the user has no groups, it does not matter that data is unshared. There SHOULD not be any difference
-        # to a user having shared access, since all important permissions are missing
-        success_urls = [
-            self.index_url,
-            self.detail_url,
-            self.report_url,
-        ]
-        fail_urls = [
-            self.new_url,
-            self.new_id_url,
-            self.log_url,
-            self.edit_url,
-            self.remove_url,
-            self.state_new_url,
-            self.action_new_url,
-            self.deadline_new_url,
-            self.state_remove_url,
-            self.action_remove_url,
-            self.new_doc_url,
-        ]
-
-        self.assert_url_success(client, success_urls)
-        self.assert_url_fail(client, fail_urls)
-
-    def test_logged_in_default_group_shared(self):
-        """ Check correct status code for all requests
-
-        Assumption: User logged in, is default group member and data is shared
-        --> Default group necessary since all base functionalities depend on this group membership
-
-        Returns:
-
-        """
-        client = Client()
-        client.login(username=self.superuser.username, password=self.superuser_pw)
-        group = self.groups.get(name=DEFAULT_GROUP)
-        self.superuser.groups.set([group])
-        # Sharing is inherited by base intervention for compensation. Therefore configure the interventions share state
-        self.eco_account.share_with_list([self.superuser])
-
-        success_urls = [
-            self.index_url,
-            self.detail_url,
-            self.report_url,
-            self.new_url,
-            self.new_id_url,
-            self.edit_url,
-            self.state_new_url,
-            self.action_new_url,
-            self.deadline_new_url,
-            self.state_remove_url,
-            self.action_remove_url,
-            self.new_doc_url,
-            self.log_url,
-            self.remove_url,
-        ]
-        self.assert_url_success(client, success_urls)
-
-    def test_logged_in_default_group_unshared(self):
-        """ Check correct status code for all requests
-
-        Assumption: User logged in, is default group member and data is NOT shared
-        --> Default group necessary since all base functionalities depend on this group membership
-
-        Returns:
-
-        """
-        client = Client()
-        client.login(username=self.superuser.username, password=self.superuser_pw)
-        group = self.groups.get(name=DEFAULT_GROUP)
-        self.superuser.groups.set([group])
-        self.eco_account.share_with_list([])
-
-        success_urls = [
-            self.index_url,
-            self.detail_url,
-            self.report_url,
-            self.new_id_url,
-            self.new_url,
-        ]
-        fail_urls = [
-            self.edit_url,
-            self.state_new_url,
-            self.action_new_url,
-            self.deadline_new_url,
-            self.state_remove_url,
-            self.action_remove_url,
-            self.new_doc_url,
-            self.log_url,
-            self.remove_url,
-        ]
-        self.assert_url_fail(client, fail_urls)
-        self.assert_url_success(client, success_urls)
-

compensation/tests/compensation/test_workflow.py

@@ -2,7 +2,7 @@
 Author: Michel Peltriaux
 Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
 Contact: michel.peltriaux@sgdnord.rlp.de
-Created on: 11.11.21
+Created on: 07.02.22
 
 """
 import datetime
@@ -11,7 +11,7 @@ from django.contrib.gis.geos import MultiPolygon
 from django.urls import reverse
 
 from compensation.models import Compensation
-from konova.settings import ETS_GROUP, ZB_GROUP
+from konova.settings import ZB_GROUP, ETS_GROUP
 from konova.tests.test_views import BaseWorkflowTestCase
 from user.models import UserAction
 
@@ -55,6 +55,7 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
             "geom": test_geom.geojson,
             "intervention": self.intervention.id,
         }
+        pre_creation_intervention_log_count = self.intervention.log.count()
 
         # Preserve the current number of intervention's compensations
         num_compensations = self.intervention.compensations.count()
@@ -66,6 +67,13 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
         self.assertEqual(new_compensation.identifier, test_id)
         self.assertEqual(new_compensation.title, test_title)
         self.assert_equal_geometries(new_compensation.geometry.geom, test_geom)
+        self.assertEqual(new_compensation.log.count(), 1)
+
+        # Expect logs to be set
+        self.assertEqual(pre_creation_intervention_log_count + 1, self.intervention.log.count())
+        self.assertEqual(new_compensation.log.count(), 1)
+        self.assertEqual(self.intervention.log.first().action, UserAction.EDITED)
+        self.assertEqual(new_compensation.log.first().action, UserAction.CREATED)
 
     def test_new_from_intervention(self):
         """ Test the creation of a compensation from a given intervention
@@ -83,6 +91,7 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
             "title": test_title,
             "geom": test_geom.geojson,
         }
+        pre_creation_intervention_log_count = self.intervention.log.count()
 
         # Preserve the current number of intervention's compensations
         num_compensations = self.intervention.compensations.count()
@@ -95,6 +104,12 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
         self.assertEqual(new_compensation.title, test_title)
         self.assert_equal_geometries(new_compensation.geometry.geom, test_geom)
 
+        # Expect logs to be set
+        self.assertEqual(new_compensation.log.count(), 1)
+        self.assertEqual(new_compensation.log.first().action, UserAction.CREATED)
+        self.assertEqual(pre_creation_intervention_log_count + 1, self.intervention.log.count())
+        self.assertEqual(self.intervention.log.first().action, UserAction.EDITED)
+
     def test_edit(self):
         """ Checks that the editing of a compensation works
 
@@ -103,6 +118,7 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
         """
         url = reverse("compensation:edit", args=(self.compensation.id,))
         self.compensation = self.fill_out_compensation(self.compensation)
+        pre_edit_log_count = self.compensation.log.count()
 
         new_title = self.create_dummy_string()
         new_identifier = self.create_dummy_string()
@@ -138,6 +154,10 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
 
         self.assert_equal_geometries(self.compensation.geometry.geom, new_geometry)
 
+        # Expect logs to be set
+        self.assertEqual(pre_edit_log_count + 1, self.compensation.log.count())
+        self.assertEqual(self.compensation.log.first().action, UserAction.EDITED)
+
     def test_checkability(self):
         """
         This tests if the checkability of the compensation (which is defined by the linked intervention's checked
@@ -152,6 +172,8 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
         # Add proper privilege for the user
         self.superuser.groups.add(self.groups.get(name=ZB_GROUP))
 
+        pre_check_log_count = self.compensation.log.count()
+
         # Prepare url and form data
         url = reverse("intervention:check", args=(self.intervention.id,))
         post_data = {
@@ -186,6 +208,7 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
 
         # Expect the user action to be in the log
         self.assertIn(checked, self.compensation.log.all())
+        self.assertEqual(pre_check_log_count + 1, self.compensation.log.count())
 
     def test_recordability(self):
         """
@@ -200,6 +223,7 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
         """
         # Add proper privilege for the user
         self.superuser.groups.add(self.groups.get(name=ETS_GROUP))
+        pre_record_log_count = self.compensation.log.count()
 
         # Prepare url and form data
         record_url = reverse("intervention:record", args=(self.intervention.id,))
@@ -234,62 +258,5 @@ class CompensationWorkflowTestCase(BaseWorkflowTestCase):
 
         # Expect the user action to be in the log
         self.assertIn(recorded, self.compensation.log.all())
-
+        self.assertEqual(pre_record_log_count + 1, self.compensation.log.count())
-
-class EcoAccountWorkflowTestCase(BaseWorkflowTestCase):
-    @classmethod
-    def setUpTestData(cls):
-        super().setUpTestData()
-
-        # Add user to conservation office group and give shared access to the account
-        cls.superuser.groups.add(cls.groups.get(name=ETS_GROUP))
-        cls.eco_account.share_with_list([cls.superuser])
-
-    def test_deductability(self):
-        """
-        This tests the deductability of an eco account.
-
-        An eco account should only be deductible if it is recorded.
-
-        Returns:
-
-        """
-        # Give user shared access to the dummy intervention, which will be needed here
-        self.intervention.share_with(self.superuser)
-
-        # Prepare data for deduction creation
-        deduct_url = reverse("compensation:acc:new-deduction", args=(self.eco_account.id,))
-        test_surface = 10.00
-        post_data = {
-            "surface": test_surface,
-            "account": self.id,
-            "intervention": self.intervention.id,
-        }
-        # Perform request --> expect to fail
-        self.client_user.post(deduct_url, post_data)
-
-        # Expect that no deduction has been created
-        self.assertEqual(0, self.eco_account.deductions.count())
-        self.assertEqual(0, self.intervention.deductions.count())
-
-        # Now mock the eco account as it would be recorded (with invalid data)
-        # Make sure the deductible surface is high enough for the request
-        self.eco_account.set_recorded(self.superuser)
-        self.eco_account.refresh_from_db()
-        self.eco_account.deductable_surface = test_surface + 1.00
-        self.eco_account.save()
-        self.assertIsNotNone(self.eco_account.recorded)
-        self.assertGreater(self.eco_account.deductable_surface, test_surface)
-
-        # Rerun the request
-        self.client_user.post(deduct_url, post_data)
-
-        # Expect that the deduction has been created
-        self.assertEqual(1, self.eco_account.deductions.count())
-        self.assertEqual(1, self.intervention.deductions.count())
-        deduction = self.eco_account.deductions.first()
-        self.assertEqual(deduction.surface, test_surface)
-        self.assertEqual(deduction.account, self.eco_account)
-        self.assertEqual(deduction.intervention, self.intervention)
-
 

compensation/tests/ecoaccount/__init__.py

@@ -0,0 +1,7 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 07.02.22
+
+"""

compensation/tests/ecoaccount/test_views.py

@@ -0,0 +1,194 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 27.10.21
+
+"""
+from django.urls import reverse
+from django.test import Client
+
+from compensation.tests.compensation.test_views import CompensationViewTestCase
+from konova.settings import DEFAULT_GROUP
+
+
+class EcoAccountViewTestCase(CompensationViewTestCase):
+    """
+    These tests focus on proper returned views depending on the user's groups privileges and login status
+
+    EcoAccounts can inherit the same tests used for compensations.
+
+    """
+    comp_state = None
+    comp_action = None
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        super().setUpTestData()
+        state = cls.create_dummy_states()
+        cls.eco_account.before_states.set([state])
+        cls.eco_account.after_states.set([state])
+
+        action = cls.create_dummy_action()
+        cls.eco_account.actions.set([action])
+
+        # Prepare urls
+        cls.index_url = reverse("compensation:acc:index", args=())
+        cls.new_url = reverse("compensation:acc:new", args=())
+        cls.new_id_url = reverse("compensation:acc:new-id", args=())
+        cls.detail_url = reverse("compensation:acc:detail", args=(cls.eco_account.id,))
+        cls.log_url = reverse("compensation:acc:log", args=(cls.eco_account.id,))
+        cls.edit_url = reverse("compensation:acc:edit", args=(cls.eco_account.id,))
+        cls.remove_url = reverse("compensation:acc:remove", args=(cls.eco_account.id,))
+        cls.report_url = reverse("compensation:acc:report", args=(cls.eco_account.id,))
+        cls.state_new_url = reverse("compensation:acc:new-state", args=(cls.eco_account.id,))
+        cls.action_new_url = reverse("compensation:acc:new-action", args=(cls.eco_account.id,))
+        cls.deadline_new_url = reverse("compensation:acc:new-deadline", args=(cls.eco_account.id,))
+        cls.new_doc_url = reverse("compensation:acc:new-doc", args=(cls.eco_account.id,))
+        cls.state_remove_url = reverse("compensation:acc:state-remove", args=(cls.eco_account.id, cls.comp_state.id,))
+        cls.action_remove_url = reverse("compensation:acc:action-remove", args=(cls.eco_account.id, cls.comp_action.id,))
+
+    def test_logged_in_no_groups_shared(self):
+        """ Check correct status code for all requests
+
+        Assumption: User logged in and has no groups and data is shared
+
+        Returns:
+
+        """
+        client = Client()
+        client.login(username=self.superuser.username, password=self.superuser_pw)
+        self.superuser.groups.set([])
+        self.eco_account.share_with_list([self.superuser])
+
+        # Since the user has no groups, it does not matter that data has been shared. There SHOULD not be any difference
+        # to a user without access, since the important permissions are missing
+        success_urls = [
+            self.index_url,
+            self.detail_url,
+            self.report_url,
+        ]
+        fail_urls = [
+            self.new_url,
+            self.new_id_url,
+            self.log_url,
+            self.edit_url,
+            self.remove_url,
+            self.state_new_url,
+            self.action_new_url,
+            self.deadline_new_url,
+            self.state_remove_url,
+            self.action_remove_url,
+            self.new_doc_url,
+        ]
+
+        self.assert_url_success(client, success_urls)
+        self.assert_url_fail(client, fail_urls)
+
+    def test_logged_in_no_groups_unshared(self):
+        """ Check correct status code for all requests
+
+        Assumption: User logged in and has no groups and data is shared
+
+        Returns:
+
+        """
+        client = Client()
+        client.login(username=self.superuser.username, password=self.superuser_pw)
+        self.superuser.groups.set([])
+        self.eco_account.share_with_list([])
+
+        # Since the user has no groups, it does not matter that data is unshared. There SHOULD not be any difference
+        # to a user having shared access, since all important permissions are missing
+        success_urls = [
+            self.index_url,
+            self.detail_url,
+            self.report_url,
+        ]
+        fail_urls = [
+            self.new_url,
+            self.new_id_url,
+            self.log_url,
+            self.edit_url,
+            self.remove_url,
+            self.state_new_url,
+            self.action_new_url,
+            self.deadline_new_url,
+            self.state_remove_url,
+            self.action_remove_url,
+            self.new_doc_url,
+        ]
+
+        self.assert_url_success(client, success_urls)
+        self.assert_url_fail(client, fail_urls)
+
+    def test_logged_in_default_group_shared(self):
+        """ Check correct status code for all requests
+
+        Assumption: User logged in, is default group member and data is shared
+        --> Default group necessary since all base functionalities depend on this group membership
+
+        Returns:
+
+        """
+        client = Client()
+        client.login(username=self.superuser.username, password=self.superuser_pw)
+        group = self.groups.get(name=DEFAULT_GROUP)
+        self.superuser.groups.set([group])
+        # Sharing is inherited by base intervention for compensation. Therefore configure the interventions share state
+        self.eco_account.share_with_list([self.superuser])
+
+        success_urls = [
+            self.index_url,
+            self.detail_url,
+            self.report_url,
+            self.new_url,
+            self.new_id_url,
+            self.edit_url,
+            self.state_new_url,
+            self.action_new_url,
+            self.deadline_new_url,
+            self.state_remove_url,
+            self.action_remove_url,
+            self.new_doc_url,
+            self.log_url,
+            self.remove_url,
+        ]
+        self.assert_url_success(client, success_urls)
+
+    def test_logged_in_default_group_unshared(self):
+        """ Check correct status code for all requests
+
+        Assumption: User logged in, is default group member and data is NOT shared
+        --> Default group necessary since all base functionalities depend on this group membership
+
+        Returns:
+
+        """
+        client = Client()
+        client.login(username=self.superuser.username, password=self.superuser_pw)
+        group = self.groups.get(name=DEFAULT_GROUP)
+        self.superuser.groups.set([group])
+        self.eco_account.share_with_list([])
+
+        success_urls = [
+            self.index_url,
+            self.detail_url,
+            self.report_url,
+            self.new_id_url,
+            self.new_url,
+        ]
+        fail_urls = [
+            self.edit_url,
+            self.state_new_url,
+            self.action_new_url,
+            self.deadline_new_url,
+            self.state_remove_url,
+            self.action_remove_url,
+            self.new_doc_url,
+            self.log_url,
+            self.remove_url,
+        ]
+        self.assert_url_fail(client, fail_urls)
+        self.assert_url_success(client, success_urls)
+

compensation/tests/ecoaccount/test_workflow.py

@@ -0,0 +1,84 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 11.11.21
+
+"""
+from django.urls import reverse
+
+from konova.settings import ETS_GROUP
+from konova.tests.test_views import BaseWorkflowTestCase
+from user.models import UserAction
+
+
+class EcoAccountWorkflowTestCase(BaseWorkflowTestCase):
+    @classmethod
+    def setUpTestData(cls):
+        super().setUpTestData()
+
+        # Add user to conservation office group and give shared access to the account
+        cls.superuser.groups.add(cls.groups.get(name=ETS_GROUP))
+        cls.eco_account.share_with_list([cls.superuser])
+
+    def test_deductability(self):
+        """
+        This tests the deductability of an eco account.
+
+        An eco account should only be deductible if it is recorded.
+
+        Returns:
+
+        """
+        # Give user shared access to the dummy intervention, which will be needed here
+        self.intervention.share_with(self.superuser)
+        pre_deduction_acc_log_count = self.eco_account.log.count()
+        pre_deduction_int_log_count = self.intervention.log.count()
+
+        # Prepare data for deduction creation
+        deduct_url = reverse("compensation:acc:new-deduction", args=(self.eco_account.id,))
+        test_surface = 10.00
+        post_data = {
+            "surface": test_surface,
+            "account": self.id,
+            "intervention": self.intervention.id,
+        }
+        # Perform request --> expect to fail
+        self.client_user.post(deduct_url, post_data)
+
+        # Expect that no deduction has been created
+        self.assertEqual(0, self.eco_account.deductions.count())
+        self.assertEqual(0, self.intervention.deductions.count())
+        self.assertEqual(pre_deduction_acc_log_count, 0)
+        self.assertEqual(pre_deduction_int_log_count, 0)
+
+        # Now mock the eco account as it would be recorded (with invalid data)
+        # Make sure the deductible surface is high enough for the request
+        self.eco_account.set_recorded(self.superuser)
+        self.eco_account.refresh_from_db()
+        self.eco_account.deductable_surface = test_surface + 1.00
+        self.eco_account.save()
+        self.assertIsNotNone(self.eco_account.recorded)
+        self.assertGreater(self.eco_account.deductable_surface, test_surface)
+        # Expect the recorded entry in the log
+        self.assertEqual(pre_deduction_acc_log_count + 1, self.eco_account.log.count())
+        self.assertTrue(self.eco_account.log.first().action == UserAction.RECORDED)
+
+        # Rerun the request
+        self.client_user.post(deduct_url, post_data)
+
+        # Expect that the deduction has been created
+        self.assertEqual(1, self.eco_account.deductions.count())
+        self.assertEqual(1, self.intervention.deductions.count())
+        deduction = self.eco_account.deductions.first()
+        self.assertEqual(deduction.surface, test_surface)
+        self.assertEqual(deduction.account, self.eco_account)
+        self.assertEqual(deduction.intervention, self.intervention)
+
+        # Expect entries in the log
+        self.assertEqual(pre_deduction_acc_log_count + 2, self.eco_account.log.count())
+        self.assertTrue(self.eco_account.log.first().action == UserAction.EDITED)
+        self.assertEqual(pre_deduction_int_log_count + 1, self.intervention.log.count())
+        self.assertTrue(self.intervention.log.first().action == UserAction.EDITED)
+
+

ema/tests/test_views.py

@@ -5,11 +5,12 @@ Contact: michel.peltriaux@sgdnord.rlp.de
 Created on: 26.10.21
 
 """
+
 from django.db.models import Q
 from django.urls import reverse
 from django.test.client import Client
 
-from compensation.tests.test_views import CompensationViewTestCase
+from compensation.tests.compensation.test_views import CompensationViewTestCase
 from ema.models import Ema
 from intervention.models import Responsibility
 from konova.models import Geometry

ema/tests/test_workflow.py

@@ -0,0 +1,165 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 08.02.22
+
+"""
+import datetime
+
+from django.contrib.gis.geos import MultiPolygon
+from django.core.exceptions import ObjectDoesNotExist
+from django.urls import reverse
+
+from ema.models import Ema
+from konova.settings import ETS_GROUP
+from konova.tests.test_views import BaseWorkflowTestCase
+from user.models import UserAction
+
+
+class EmaWorkflowTestCase(BaseWorkflowTestCase):
+    ema = None
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        super().setUpTestData()
+
+    def setUp(self) -> None:
+        super().setUp()
+        # Create a fresh dummy (non-valid) compensation before each test
+        self.ema = self.create_dummy_ema()
+
+    def test_new(self):
+        """ Test the creation of an Ema
+
+        Returns:
+
+        """
+        self.superuser.groups.add(self.groups.get(name=ETS_GROUP))
+        # Prepare url and form data to be posted
+        new_url = reverse("ema:new")
+        test_id = self.create_dummy_string()
+        test_title = self.create_dummy_string()
+        test_geom = self.create_dummy_geometry()
+        test_conservation_office = self.get_conservation_office_code()
+        post_data = {
+            "identifier": test_id,
+            "title": test_title,
+            "geom": test_geom.geojson,
+            "conservation_office": test_conservation_office.id
+        }
+        self.client_user.post(new_url, post_data)
+
+        try:
+            ema = Ema.objects.get(
+                identifier=test_id
+            )
+        except ObjectDoesNotExist:
+            self.fail(msg="Ema not created")
+
+        self.assertEqual(ema.identifier, test_id)
+        self.assertEqual(ema.title, test_title)
+        self.assert_equal_geometries(ema.geometry.geom, test_geom)
+        self.assertEqual(ema.log.count(), 1)
+
+        # Expect logs to be set
+        self.assertEqual(ema.log.count(), 1)
+        self.assertEqual(ema.log.first().action, UserAction.CREATED)
+
+    def test_edit(self):
+        """ Checks that the editing of an Ema works
+
+        Returns:
+
+        """
+        self.superuser.groups.add(self.groups.get(name=ETS_GROUP))
+        self.ema.users.add(self.superuser)
+        url = reverse("ema:edit", args=(self.ema.id,))
+        self.ema = self.fill_out_ema(self.ema)
+        pre_edit_log_count = self.ema.log.count()
+
+        new_title = self.create_dummy_string()
+        new_identifier = self.create_dummy_string()
+        new_comment = self.create_dummy_string()
+        new_geometry = MultiPolygon(srid=4326)  # Create an empty geometry
+        test_conservation_office = self.get_conservation_office_code()
+
+        check_on_elements = {
+            self.ema.title: new_title,
+            self.ema.identifier: new_identifier,
+            self.ema.comment: new_comment,
+        }
+        for k, v in check_on_elements.items():
+            self.assertNotEqual(k, v)
+
+        post_data = {
+            "identifier": new_identifier,
+            "title": new_title,
+            "comment": new_comment,
+            "geom": new_geometry.geojson,
+            "conservation_office": test_conservation_office.id
+        }
+        self.client_user.post(url, post_data)
+        self.ema.refresh_from_db()
+
+        check_on_elements = {
+            self.ema.title: new_title,
+            self.ema.identifier: new_identifier,
+            self.ema.comment: new_comment,
+        }
+
+        for k, v in check_on_elements.items():
+            self.assertEqual(k, v)
+
+        self.assert_equal_geometries(self.ema.geometry.geom, new_geometry)
+
+        # Expect logs to be set
+        self.assertEqual(pre_edit_log_count + 1, self.ema.log.count())
+        self.assertEqual(self.ema.log.first().action, UserAction.EDITED)
+
+    def test_recordability(self):
+        """
+        This tests if the recordability of the Ema is triggered by the quality of it's data (e.g. not all fields filled)
+
+        Returns:
+
+        """
+        # Add proper privilege for the user
+        self.superuser.groups.add(self.groups.get(name=ETS_GROUP))
+        self.ema.users.add(self.superuser)
+        pre_record_log_count = self.ema.log.count()
+
+        # Prepare url and form data
+        record_url = reverse("ema:record", args=(self.ema.id,))
+        post_data = {
+            "confirm": True,
+        }
+
+        # Make sure the ema is not recorded
+        self.assertIsNone(self.ema.recorded)
+
+        # Run the request --> expect fail, since the Ema is not valid, yet
+        self.client_user.post(record_url, post_data)
+
+        # Check that the Ema is still not recorded
+        self.assertIsNone(self.ema.recorded)
+
+        # Now fill out the data for a compensation
+        self.ema = self.fill_out_ema(self.ema)
+
+        # Rerun the request
+        self.client_user.post(record_url, post_data)
+
+        # Expect the Ema now to be recorded
+        # Attention: We can only test the date part of the timestamp,
+        # since the delay in microseconds would lead to fail
+        self.ema.refresh_from_db()
+        recorded = self.ema.recorded
+        self.assertIsNotNone(recorded)
+        self.assertEqual(self.superuser, recorded.user)
+        self.assertEqual(UserAction.RECORDED, recorded.action)
+        self.assertEqual(datetime.date.today(), recorded.timestamp.date())
+
+        # Expect the user action to be in the log
+        self.assertIn(recorded, self.ema.log.all())
+        self.assertEqual(pre_record_log_count + 1, self.ema.log.count())

konova/models/object.py

@@ -132,7 +132,7 @@ class BaseObject(BaseResource):
 
             self.save()
 
-    def mark_as_edited(self, performing_user: User, edit_comment: str = None):
+    def mark_as_edited(self, performing_user: User, request: HttpRequest = None, edit_comment: str = None):
         """ In case the object or a related object changed the log history needs to be updated
 
         Args:

konova/tests/test_views.py

@@ -7,6 +7,7 @@ Created on: 26.10.21
 """
 import datetime
 
+from codelist.settings import CODELIST_CONSERVATION_OFFICE_ID
 from ema.models import Ema
 from user.models import User
 from django.contrib.auth.models import Group
@@ -15,7 +16,7 @@ from django.core.exceptions import ObjectDoesNotExist
 from django.test import TestCase, Client
 from django.urls import reverse
 
-from codelist.models import KonovaCode
+from codelist.models import KonovaCode, KonovaCodeList
 from compensation.models import Compensation, CompensationState, CompensationAction, EcoAccount, EcoAccountDeduction
 from intervention.models import Legal, Responsibility, Intervention
 from konova.management.commands.setup_data import GROUPS_DATA
@@ -236,10 +237,10 @@ class BaseTestCase(TestCase):
 
         """
         codes = KonovaCode.objects.bulk_create([
-            KonovaCode(id=1, is_selectable=True, long_name="Test1"),
+            KonovaCode(id=1, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test1"),
-            KonovaCode(id=2, is_selectable=True, long_name="Test2"),
+            KonovaCode(id=2, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test2"),
-            KonovaCode(id=3, is_selectable=True, long_name="Test3"),
+            KonovaCode(id=3, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test3"),
-            KonovaCode(id=4, is_selectable=True, long_name="Test4"),
+            KonovaCode(id=4, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test4"),
         ])
         return codes
 
@@ -298,6 +299,37 @@ class BaseTestCase(TestCase):
         compensation.geometry.save()
         return compensation
 
+    @classmethod
+    def get_conservation_office_code(cls):
+        """ Returns a dummy KonovaCode as conservation office code
+
+        Returns:
+
+        """
+        codelist = KonovaCodeList.objects.get_or_create(
+            id=CODELIST_CONSERVATION_OFFICE_ID
+        )[0]
+        code = KonovaCode.objects.get(id=2)
+        codelist.codes.add(code)
+        return code
+
+    @classmethod
+    def fill_out_ema(cls, ema):
+        """ Adds all required (dummy) data to an Ema
+
+        Returns:
+        """
+        ema.responsible.conservation_office = cls.get_conservation_office_code()
+        ema.responsible.conservation_file_number = "test"
+        ema.responsible.handler = "handler"
+        ema.responsible.save()
+        ema.after_states.add(cls.comp_state)
+        ema.before_states.add(cls.comp_state)
+        ema.actions.add(cls.comp_action)
+        ema.geometry.geom = cls.create_dummy_geometry()
+        ema.geometry.save()
+        return ema
+
     def assert_equal_geometries(self, geom1: MultiPolygon, geom2: MultiPolygon):
         """ Assert for geometries to be equal