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

@@ -0,0 +1,225 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 07.02.22
+
+"""
+from django.test.client import Client
+from django.urls import reverse
+
+from konova.settings import DEFAULT_GROUP
+from konova.tests.test_views import BaseViewTestCase
+
+
+class CompensationViewTestCase(BaseViewTestCase):
+    """
+    These tests focus on proper returned views depending on the user's groups privileges and login status
+
+    """
+
+    @classmethod
+    def setUpTestData(cls) -> None:
+        super().setUpTestData()
+        state = cls.create_dummy_states()
+        cls.compensation.before_states.set([state])
+        cls.compensation.after_states.set([state])
+
+        action = cls.create_dummy_action()
+        cls.compensation.actions.set([action])
+
+        # Prepare urls
+        cls.index_url = reverse("compensation:index", args=())
+        cls.new_url = reverse("compensation:new", args=(cls.intervention.id,))
+        cls.new_id_url = reverse("compensation:new-id", args=())
+        cls.detail_url = reverse("compensation:detail", args=(cls.compensation.id,))
+        cls.log_url = reverse("compensation:log", args=(cls.compensation.id,))
+        cls.edit_url = reverse("compensation:edit", args=(cls.compensation.id,))
+        cls.remove_url = reverse("compensation:remove", args=(cls.compensation.id,))
+        cls.report_url = reverse("compensation:report", args=(cls.compensation.id,))
+        cls.state_new_url = reverse("compensation:new-state", args=(cls.compensation.id,))
+        cls.action_new_url = reverse("compensation:new-action", args=(cls.compensation.id,))
+        cls.deadline_new_url = reverse("compensation:new-deadline", args=(cls.compensation.id,))
+        cls.new_doc_url = reverse("compensation:new-doc", args=(cls.compensation.id,))
+
+        cls.state_remove_url = reverse("compensation:state-remove", args=(cls.compensation.id, cls.comp_state.id,))
+        cls.action_remove_url = reverse("compensation:action-remove", args=(cls.compensation.id, cls.comp_action.id,))
+
+    def test_anonymous_user(self):
+        """ Check correct status code for all requests
+
+        Assumption: User not logged in
+
+        Returns:
+
+        """
+        client = Client()
+
+        success_urls = [
+            self.report_url,
+        ]
+        fail_urls = [
+            self.index_url,
+            self.detail_url,
+            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_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.intervention.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([])
+        # Sharing is inherited by base intervention for compensation. Therefore configure the interventions share state
+        self.intervention.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.intervention.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])
+        # Sharing is inherited by base intervention for compensation. Therefore configure the interventions share state
+        self.intervention.share_with_list([])
+
+        success_urls = [
+            self.index_url,
+            self.detail_url,
+            self.report_url,
+            self.new_id_url,
+        ]
+        fail_urls = [
+            self.new_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_fail(client, fail_urls)
+        self.assert_url_success(client, success_urls)
+

compensation/tests/compensation/test_workflow.py

@@ -0,0 +1,262 @@
+"""
+Author: Michel Peltriaux
+Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
+Contact: michel.peltriaux@sgdnord.rlp.de
+Created on: 07.02.22
+
+"""
+import datetime
+
+from django.contrib.gis.geos import MultiPolygon
+from django.urls import reverse
+
+from compensation.models import Compensation
+from konova.settings import ZB_GROUP, ETS_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.share_with(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:
+        super().setUp()
+        # 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_new(self):
+        """ Test the creation of a compensation
+
+        Returns:
+
+        """
+        # Prepare url and form data to be posted
+        new_url = reverse("compensation:new")
+        test_id = self.create_dummy_string()
+        test_title = self.create_dummy_string()
+        test_geom = self.create_dummy_geometry()
+        post_data = {
+            "identifier": test_id,
+            "title": test_title,
+            "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()
+        self.client_user.post(new_url, post_data)
+
+        self.intervention.refresh_from_db()
+        self.assertEqual(num_compensations + 1, self.intervention.compensations.count())
+        new_compensation = self.intervention.compensations.get(identifier=test_id)
+        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
+
+        Returns:
+
+        """
+        # Prepare url and form data to be posted
+        new_url = reverse("compensation:new", args=(self.intervention.id,))
+        test_id = self.create_dummy_string()
+        test_title = self.create_dummy_string()
+        test_geom = self.create_dummy_geometry()
+        post_data = {
+            "identifier": test_id,
+            "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()
+        self.client_user.post(new_url, post_data)
+
+        self.intervention.refresh_from_db()
+        self.assertEqual(num_compensations + 1, self.intervention.compensations.count())
+        new_compensation = self.intervention.compensations.get(identifier=test_id)
+        self.assertEqual(new_compensation.identifier, test_id)
+        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
+
+        Returns:
+
+        """
+        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()
+        new_comment = self.create_dummy_string()
+        new_geometry = MultiPolygon(srid=4326)  # Create an empty geometry
+
+        check_on_elements = {
+            self.compensation.title: new_title,
+            self.compensation.identifier: new_identifier,
+            self.compensation.comment: new_comment,
+        }
+        for k, v in check_on_elements.items():
+            self.assertNotEqual(k, v)
+
+        post_data = {
+            "identifier": new_identifier,
+            "title": new_title,
+            "intervention": self.intervention.id,  # just keep the intervention as it is
+            "comment": new_comment,
+            "geom": new_geometry.geojson,
+        }
+        self.client_user.post(url, post_data)
+        self.compensation.refresh_from_db()
+
+        check_on_elements = {
+            self.compensation.title: new_title,
+            self.compensation.identifier: new_identifier,
+            self.compensation.comment: new_comment,
+        }
+
+        for k, v in check_on_elements.items():
+            self.assertEqual(k, v)
+
+        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
+        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))
+
+        pre_check_log_count = self.compensation.log.count()
+
+        # 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())
+        self.assertEqual(pre_check_log_count + 1, self.compensation.log.count())
+
+    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))
+        pre_record_log_count = self.compensation.log.count()
+
+        # 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())
+        self.assertEqual(pre_record_log_count + 1, self.compensation.log.count())
+