compensation/tests/test_workflow.py

@@ -0,0 +1,135 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 11.11.21
+
+"""
+import datetime
+
+from django.urls import reverse
+
+from compensation.models import Compensation
+from konova.settings import DEFAULT_GROUP, ETS_GROUP, ZB_GROUP
+from konova.tests.test_views import BaseWorkflowTestCase
+from user.models import UserAction
+
+
+class CompensationWorkflowTestCase(BaseWorkflowTestCase):
+    @classmethod
+    def setUpTestData(cls):
+        super().setUpTestData()
+
+        # Give the user shared access to the dummy intervention -> inherits the access to the compensation
+        cls.intervention.users.add(cls.superuser)
+
+        # Make sure the intervention itself would be fine with valid data
+        cls.intervention = cls.fill_out_intervention(cls.intervention)
+
+        # Make sure the compensation is linked to the intervention
+        cls.intervention.compensations.set([cls.compensation])
+
+    def setUp(self) -> None:
+        # Delete all existing compensations, which might be created by tests
+        Compensation.objects.all().delete()
+
+        # Create a fresh dummy (non-valid) compensation before each test
+        self.compensation = self.create_dummy_compensation()
+
+    def test_checkability(self):
+        """
+        This tests if the checkability of the compensation (which is defined by the linked intervention's checked
+        attribute) is triggered by the quality of it's data (e.g. not all fields filled)
+
+        We expect a compensation, missing required data, linked to an intervention to fail the intervention's quality
+        check performed in the checking action.
+
+        Returns:
+
+        """
+        # Add proper privilege for the user
+        self.superuser.groups.add(self.groups.get(name=ZB_GROUP))
+
+        # Prepare url and form data
+        url = reverse("intervention:check", args=(self.intervention.id,))
+        post_data = {
+            "checked_intervention": True,
+            "checked_comps": True,
+        }
+
+        # Make sure the intervention is not checked
+        self.assertIsNone(self.intervention.checked)
+
+        # Run the request --> expect fail, since the compensation is not valid, yet
+        self.client_user.post(url, post_data)
+
+        # Check that the intervention is still not recorded
+        self.assertIsNone(self.intervention.checked)
+
+        # Now fill out the data for a compensation
+        self.compensation = self.fill_out_compensation(self.compensation)
+
+        # Rerun the request
+        self.client_user.post(url, post_data)
+
+        # Expect the linked intervention now to be checked
+        # Attention: We can only test the date part of the timestamp,
+        # since the delay in microseconds would lead to fail
+        self.intervention.refresh_from_db()
+        checked = self.intervention.checked
+        self.assertIsNotNone(checked)
+        self.assertEqual(self.superuser, checked.user)
+        self.assertEqual(UserAction.CHECKED, checked.action)
+        self.assertEqual(datetime.date.today(), checked.timestamp.date())
+
+        # Expect the user action to be in the log
+        self.assertIn(checked, self.compensation.log.all())
+
+    def test_recordability(self):
+        """
+        This tests if the recordability of the compensation (which is defined by the linked intervention's recorded
+        attribute) is triggered by the quality of it's data (e.g. not all fields filled)
+
+        We expect a compensation, missing required data, linked to an intervention to fail the intervention's quality
+        check performed in the recording action.
+
+        Returns:
+
+        """
+        # Add proper privilege for the user
+        self.superuser.groups.add(self.groups.get(name=ETS_GROUP))
+
+        # Prepare url and form data
+        record_url = reverse("intervention:record", args=(self.intervention.id,))
+        post_data = {
+            "confirm": True,
+        }
+
+        # Make sure the intervention is not recorded
+        self.assertIsNone(self.intervention.recorded)
+
+        # Run the request --> expect fail, since the compensation is not valid, yet
+        self.client_user.post(record_url, post_data)
+
+        # Check that the intervention is still not recorded
+        self.assertIsNone(self.intervention.recorded)
+
+        # Now fill out the data for a compensation
+        self.compensation = self.fill_out_compensation(self.compensation)
+
+        # Rerun the request
+        self.client_user.post(record_url, post_data)
+
+        # Expect the linked intervention 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.intervention.refresh_from_db()
+        recorded = self.intervention.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.compensation.log.all())
+

intervention/forms/modalForms.py

@@ -235,16 +235,7 @@ class CheckModalForm(BaseModalForm):
 
         """
         with transaction.atomic():
-            user_action = UserActionLogEntry.objects.create(
-                user=self.user,
-                action=UserAction.CHECKED
-            )
-            # Replace old checked
-            if self.instance.checked:
-                self.instance.checked.delete()
-            self.instance.checked = user_action
-            self.instance.log.add(user_action)
-            self.instance.save()
+            self.instance.toggle_checked(self.user)
 
         # Send message to the SSO server
         messenger = Messenger(

intervention/models.py

@@ -19,7 +19,6 @@ from intervention.utils.quality import InterventionQualityChecker
 from konova.models import BaseObject, Geometry, UuidModel, BaseResource, AbstractDocument, \
     generate_document_file_upload_path, RecordableObject, CheckableObject, ShareableObject
 from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE, LANIS_ZOOM_LUT
-from konova.utils import generators
 from user.models import UserActionLogEntry
 
 
@@ -285,6 +284,45 @@ class Intervention(BaseObject, ShareableObject, RecordableObject, CheckableObjec
         )
         return revoc_docs, regular_docs
 
+    def toggle_recorded(self, user: User):
+        """ Toggle the recorded state
+
+        For interventions the recorded action needs to be added to their compensation objects as well
+
+        Args:
+            user (User): The performing user
+
+        Returns:
+
+        """
+        log_entry = super().toggle_recorded(user)
+
+        # Add this action to the linked compensation logs as well
+        comps = self.compensations.all()
+        for comp in comps:
+            comp.log.add(log_entry)
+
+    def toggle_checked(self, user: User) -> UserActionLogEntry:
+        """ Toggle the checked state
+
+        For interventions the checked action needs to be added to their compensation objects as well
+
+        Args:
+            user (User): The performing user
+
+        Returns:
+
+        """
+        log_entry = super().toggle_checked(user)
+        # Leave if the log_entry is None (means "unchecked")
+        if log_entry is None:
+            return
+        # Add this action to the linked compensation logs as well
+        comps = self.compensations.all()
+        for comp in comps:
+            comp.log.add(log_entry)
+
+
 
 class InterventionDocument(AbstractDocument):
     """

intervention/tests/test_workflow.py

@@ -29,16 +29,6 @@ class InterventionWorkflowTestCase(BaseWorkflowTestCase):
         # Give the user shared access to the dummy intervention
         cls.intervention.users.add(cls.superuser)
 
-    def setUp(self) -> None:
-        """ Setup data before each test run
-
-        Returns:
-
-        """
-        # Set the default group as only group for the user
-        default_group = self.groups.get(name=DEFAULT_GROUP)
-        self.superuser.groups.set([default_group])
-
     def test_new(self):
         """
         Checks a 'normal' case of creating a new intervention.

konova/models.py

@@ -122,11 +122,12 @@ class BaseObject(BaseResource):
             self.save()
 
     def add_log_entry(self, action: UserAction, user: User, comment: str):
-        """ Wraps adding of UserActionLogEntry to self.log
+        """ Wraps adding of UserActionLogEntry to log
 
         Args:
             action (UserAction): The performed UserAction
             user (User): Performing user
+            comment (str): The optional comment
 
         Returns:
 
@@ -349,6 +350,7 @@ class RecordableObject(models.Model):
         self.recorded = None
         self.save()
         self.log.add(action)
+        return action
 
     def set_recorded(self, user: User):
         """ Perform recording
@@ -366,8 +368,9 @@ class RecordableObject(models.Model):
         self.recorded = action
         self.save()
         self.log.add(action)
+        return action
 
-    def toggle_recorded(self, user: User):
+    def toggle_recorded(self, user: User) -> UserActionLogEntry:
         """ Un/Record intervention
 
         Args:
@@ -377,9 +380,10 @@ class RecordableObject(models.Model):
 
         """
         if not self.recorded:
-            self.set_recorded(user)
+            ret_log_entry = self.set_recorded(user)
         else:
-            self.set_unrecorded(user)
+            ret_log_entry = self.set_unrecorded(user)
+        return ret_log_entry
 
 
 class CheckableObject(models.Model):
@@ -392,10 +396,11 @@ class CheckableObject(models.Model):
         help_text="Holds data on user and timestamp of this action",
         related_name="+"
     )
+
     class Meta:
         abstract = True
 
-    def set_unchecked(self, user: User):
+    def set_unchecked(self) -> None:
         """ Perform unrecording
 
         Args:
@@ -403,10 +408,13 @@ class CheckableObject(models.Model):
         Returns:
 
         """
+        # Do not .delete() the checked attribute! Just set it to None, since a delete() would kill it out of the
+        # log history, which is not what we want!
         self.checked = None
         self.save()
+        return None
 
-    def set_checked(self, user: User):
+    def set_checked(self, user: User) -> UserActionLogEntry:
         """ Perform checking
 
         Args:
@@ -422,8 +430,9 @@ class CheckableObject(models.Model):
         self.checked = action
         self.save()
         self.log.add(action)
+        return action
 
-    def toggle_checked(self, user: User):
+    def toggle_checked(self, user: User) -> UserActionLogEntry:
         """ Un/Record intervention
 
         Args:
@@ -433,9 +442,10 @@ class CheckableObject(models.Model):
 
         """
         if not self.checked:
-            self.set_checked(user)
+            ret_log_entry = self.set_checked(user)
         else:
-            self.set_unchecked(user)
+            ret_log_entry = self.set_unchecked()
+        return ret_log_entry
 
 
 class ShareableObject(models.Model):

konova/tests/test_views.py

@@ -18,6 +18,7 @@ from compensation.models import Compensation, CompensationState, CompensationAct
 from intervention.models import LegalData, ResponsibilityData, Intervention
 from konova.management.commands.setup_data import GROUPS_DATA
 from konova.models import Geometry
+from konova.settings import DEFAULT_GROUP
 from user.models import UserActionLogEntry, UserAction
 
 
@@ -49,6 +50,8 @@ class BaseTestCase(TestCase):
         cls.intervention = cls.create_dummy_intervention()
         cls.compensation = cls.create_dummy_compensation()
         cls.eco_account = cls.create_dummy_eco_account()
+        cls.create_dummy_states()
+        cls.create_dummy_action()
         cls.codes = cls.create_dummy_codes()
 
     @classmethod
@@ -204,7 +207,16 @@ class BaseTestCase(TestCase):
         return codes
 
     @staticmethod
-    def fill_out_intervention(intervention: Intervention) -> Intervention:
+    def create_dummy_geometry() -> MultiPolygon:
+        """ Creates some geometry
+
+        Returns:
+
+        """
+        return MultiPolygon(Polygon.from_bbox([-4.526367, 18.354526, -1.801758, 20.591652]))
+
+    @classmethod
+    def fill_out_intervention(cls, intervention: Intervention) -> Intervention:
         """ Adds all required (dummy) data to an intervention
 
         Args:
@@ -224,11 +236,28 @@ class BaseTestCase(TestCase):
         intervention.legal.process_type = KonovaCode.objects.get(id=3)
         intervention.legal.save()
         intervention.legal.laws.set([KonovaCode.objects.get(id=(4))])
-        intervention.geometry.geom = MultiPolygon(Polygon.from_bbox([-4.526367, 18.354526, -1.801758, 20.591652]))
+        intervention.geometry.geom = cls.create_dummy_geometry()
         intervention.geometry.save()
         intervention.save()
         return intervention
 
+    @classmethod
+    def fill_out_compensation(cls, compensation: Compensation) -> Compensation:
+        """ Adds all required (dummy) data to a compensation
+
+        Args:
+            compensation (Compensation): The compensation which shall be filled out
+
+        Returns:
+            compensation (Compensation): The modified compensation
+        """
+        compensation.after_states.add(cls.comp_state)
+        compensation.before_states.add(cls.comp_state)
+        compensation.actions.add(cls.comp_action)
+        compensation.geometry.geom = cls.create_dummy_geometry()
+        compensation.geometry.save()
+        return compensation
+
 
 class BaseViewTestCase(BaseTestCase):
     """ Wraps basic test functionality, reusable for every specialized ViewTestCase
@@ -236,6 +265,9 @@ class BaseViewTestCase(BaseTestCase):
     """
     login_url = None
 
+    class Meta:
+        abstract = True
+
     @classmethod
     def setUpTestData(cls) -> None:
         super().setUpTestData()
@@ -404,6 +436,16 @@ class BaseWorkflowTestCase(BaseTestCase):
         cls.client_user.login(username=cls.superuser.username, password=cls.superuser_pw)
         cls.client_anon = Client()
 
+    def setUp(self) -> None:
+        """ Setup data before each test run
+
+        Returns:
+
+        """
+        # Set the default group as only group for the user
+        default_group = self.groups.get(name=DEFAULT_GROUP)
+        self.superuser.groups.set([default_group])
+
     def assert_object_is_deleted(self, obj):
         """ Provides a quick check whether an object has been removed from the database or not