#19 Tests
* adds workflow tests for compensation checking and recording * improves related code
This commit is contained in:
		
							parent
							
								
									796990ffbc
								
							
						
					
					
						commit
						ffadfa2f47
					
				
							
								
								
									
										135
									
								
								compensation/tests/test_workflow.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										135
									
								
								compensation/tests/test_workflow.py
									
									
									
									
									
										Normal file
									
								
							@ -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())
 | 
			
		||||
 | 
			
		||||
@ -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(
 | 
			
		||||
 | 
			
		||||
@ -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):
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
@ -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.
 | 
			
		||||
 | 
			
		||||
@ -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):
 | 
			
		||||
 | 
			
		||||
@ -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
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
		Loading…
	
	
			
			x
			
			
		
	
		Reference in New Issue
	
	Block a user