konova/compensation/tests/compensation/test_workflow.py

"""
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()

    def setUp(self) -> None:
        super().setUp()

        # Give the user shared access to the dummy intervention -> inherits the access to the compensation
        self.intervention.share_with_user(self.superuser)

        # Make sure the intervention itself would be fine with valid data
        self.intervention = self.fill_out_intervention(self.intervention)

        # Make sure the compensation is linked to the intervention
        self.intervention.compensations.set([self.compensation])
        
        # 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())