""" 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.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.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, } # 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) 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, } # 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) 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) 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) 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()) 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)