""" 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.core.exceptions import ObjectDoesNotExist from django.urls import reverse from compensation.models import EcoAccount, EcoAccountDeduction from konova.settings import ETS_GROUP, DEFAULT_GROUP from konova.tests.test_views import BaseWorkflowTestCase from user.models import UserAction class EcoAccountWorkflowTestCase(BaseWorkflowTestCase): @classmethod def setUpTestData(cls): super().setUpTestData() def setUp(self) -> None: super().setUp() # Add user to conservation office group and give shared access to the account self.superuser.groups.add(self.groups.get(name=DEFAULT_GROUP)) self.superuser.groups.add(self.groups.get(name=ETS_GROUP)) self.eco_account.share_with_user_list([self.superuser]) def test_new(self): """ Test the creation of an EcoAccount Returns: """ # Prepare url and form data to be posted new_url = reverse("compensation:acc:new") test_id = self.create_dummy_string() test_title = self.create_dummy_string() test_geom = self.create_dummy_geometry() geom_json = self.create_geojson(test_geom) test_deductable_surface = 1000 test_conservation_office = self.get_conservation_office_code() post_data = { "identifier": test_id, "title": test_title, "geom": geom_json, "surface": test_deductable_surface, "conservation_office": test_conservation_office.id } self.client_user.post(new_url, post_data) try: acc = EcoAccount.objects.get( identifier=test_id ) except ObjectDoesNotExist: self.fail(msg="EcoAccount not created") self.assertEqual(acc.identifier, test_id) self.assertEqual(acc.title, test_title) self.assertEqual(acc.deductable_surface, test_deductable_surface) self.assertEqual(acc.deductable_rest, test_deductable_surface) self.assert_equal_geometries(acc.geometry.geom, test_geom) self.assertEqual(acc.log.count(), 1) self.assertEqual(acc.created, acc.modified) # Expect logs to be set self.assertEqual(acc.log.count(), 1) self.assertEqual(acc.log.first().action, UserAction.CREATED) def test_edit(self): """ Checks that the editing of an EcoAccount works Returns: """ self.eco_account.share_with_user(self.superuser) url = reverse("compensation:acc:edit", args=(self.eco_account.id,)) pre_edit_log_count = self.eco_account.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 test_conservation_office = self.get_conservation_office_code() test_deductable_surface = self.eco_account.deductable_surface + 100 check_on_elements = { self.eco_account.title: new_title, self.eco_account.identifier: new_identifier, self.eco_account.comment: new_comment, self.eco_account.deductable_surface: test_deductable_surface, } for k, v in check_on_elements.items(): self.assertNotEqual(k, v) post_data = { "identifier": new_identifier, "title": new_title, "comment": new_comment, "geom": new_geometry.geojson, "surface": test_deductable_surface, "conservation_office": test_conservation_office.id } self.client_user.post(url, post_data) self.eco_account.refresh_from_db() check_on_elements = { self.eco_account.title: new_title, self.eco_account.identifier: new_identifier, self.eco_account.deductable_surface: test_deductable_surface, self.eco_account.deductable_rest: test_deductable_surface, self.eco_account.comment: new_comment, } for k, v in check_on_elements.items(): self.assertEqual(k, v) self.assert_equal_geometries(self.eco_account.geometry.geom, new_geometry) # Expect logs to be set self.assertEqual(pre_edit_log_count + 1, self.eco_account.log.count()) self.assertEqual(self.eco_account.log.first().action, UserAction.EDITED) def test_recordability(self): """ This tests if the recordability of the EcoAccount is triggered by the quality of it's data (e.g. not all fields filled) Returns: """ # Add proper privilege for the user self.eco_account.share_with_user(self.superuser) pre_record_log_count = self.eco_account.log.count() # Prepare url and form data record_url = reverse("compensation:acc:record", args=(self.eco_account.id,)) post_data = { "confirm": True, } self.eco_account.refresh_from_db() # Make sure the account is not recorded self.assertIsNone(self.eco_account.recorded) # Run the request --> expect fail, since the account is not valid, yet self.client_user.post(record_url, post_data) # Check that the account is still not recorded self.assertIsNone(self.eco_account.recorded) # Now fill out the data for an ecoaccount self.eco_account = self.fill_out_eco_account(self.eco_account) # Rerun the request self.client_user.post(record_url, post_data) # Expect the EcoAccount 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.eco_account.refresh_from_db() recorded = self.eco_account.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.eco_account.log.all()) self.assertEqual(pre_record_log_count + 1, self.eco_account.log.count()) def test_new_deduction(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_user(self.superuser) pre_deduction_acc_log_count = self.eco_account.log.count() pre_deduction_int_log_count = self.intervention.log.count() # Prepare data for deduction creation deduct_url = reverse("compensation:acc:new-deduction", args=(self.eco_account.id,)) test_surface = 10.50 post_data = { "surface": test_surface, "account": self.eco_account.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 since the eco account is not recorded, yet self.assertEqual(0, self.eco_account.deductions.count()) self.assertEqual(0, self.intervention.deductions.count()) self.assertEqual(pre_deduction_acc_log_count, 0) self.assertEqual(pre_deduction_int_log_count, 0) # Now mock the eco account as it would be recorded (with invalid data) # Make sure the deductible surface is valid for the request self.eco_account.set_recorded(self.superuser) self.eco_account.refresh_from_db() self.eco_account.deductable_surface = test_surface + 1.0 self.eco_account.save() self.assertIsNotNone(self.eco_account.recorded) self.assertGreater(self.eco_account.deductable_surface, test_surface) # Expect the recorded entry in the log self.assertEqual(pre_deduction_acc_log_count + 1, self.eco_account.log.count()) self.assertTrue(self.eco_account.log.first().action == UserAction.RECORDED) # Rerun the request self.client_user.post(deduct_url, post_data) # Expect that the deduction has been created self.eco_account.refresh_from_db() 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(self.eco_account.deductable_rest, self.eco_account.deductable_surface - deduction.surface) self.assertEqual(deduction.account, self.eco_account) self.assertEqual(deduction.intervention, self.intervention) # Expect entries in the log self.assertEqual(pre_deduction_acc_log_count + 2, self.eco_account.log.count()) self.assertTrue(self.eco_account.log.first().action == UserAction.EDITED) self.assertEqual(pre_deduction_int_log_count + 1, self.intervention.log.count()) self.assertTrue(self.intervention.log.first().action == UserAction.EDITED) def test_edit_deduction(self): test_surface = self.eco_account.deductable_rest self.eco_account.set_recorded(self.superuser) self.intervention.share_with_user(self.superuser) self.eco_account.refresh_from_db() self.assertTrue(self.superuser, self.intervention.is_shared_with(self.superuser)) deduction = EcoAccountDeduction.objects.create( intervention=self.intervention, account=self.eco_account, surface=1.10 ) self.assertEqual(1, self.intervention.deductions.count()) self.assertEqual(1, self.eco_account.deductions.count()) # Prepare url and form data to be posted new_url = reverse("compensation:acc:edit-deduction", args=(self.eco_account.id, deduction.id)) post_data = { "intervention": deduction.intervention.id, "account": deduction.account.id, "surface": test_surface, } pre_edit_intervention_log_count = self.intervention.log.count() pre_edit_account_log_count = self.eco_account.log.count() num_deductions_intervention = self.intervention.deductions.count() num_deductions_account = self.eco_account.deductions.count() self.client_user.post(new_url, post_data) self.intervention.refresh_from_db() self.eco_account.refresh_from_db() deduction.refresh_from_db() self.assertEqual(self.eco_account.deductable_rest, self.eco_account.deductable_surface - deduction.surface) self.assertEqual(num_deductions_intervention, self.intervention.deductions.count()) self.assertEqual(num_deductions_account, self.eco_account.deductions.count()) self.assertEqual(deduction.surface, test_surface) # Expect logs to be set self.assertEqual(pre_edit_intervention_log_count + 1, self.intervention.log.count()) self.assertEqual(pre_edit_account_log_count + 1, self.eco_account.log.count()) self.assertEqual(self.intervention.log.first().action, UserAction.EDITED) self.assertEqual(self.eco_account.log.first().action, UserAction.EDITED) def test_remove_deduction(self): intervention = self.deduction.intervention account = self.deduction.account deducted_surface = self.deduction.surface # Prepare url and form data to be posted new_url = reverse("compensation:acc:remove-deduction", args=(account.id, self.deduction.id)) post_data = { "confirm": True, } intervention.share_with_user(self.superuser) account.share_with_user(self.superuser) pre_edit_intervention_log_count = intervention.log.count() pre_edit_account_log_count = account.log.count() pre_edit_account_rest = account.deductable_rest num_deductions_intervention = intervention.deductions.count() num_deductions_account = account.deductions.count() self.client_user.post(new_url, post_data) intervention.refresh_from_db() account.refresh_from_db() self.assertEqual(num_deductions_intervention - 1, intervention.deductions.count()) self.assertEqual(num_deductions_account - 1, account.deductions.count()) self.assertEqual(account.deductable_rest, pre_edit_account_rest + deducted_surface) # Expect logs to be set self.assertEqual(pre_edit_intervention_log_count + 1, intervention.log.count()) self.assertEqual(pre_edit_account_log_count + 1, account.log.count()) self.assertEqual(intervention.log.first().action, UserAction.EDITED) self.assertEqual(account.log.first().action, UserAction.EDITED) def test_non_editable_after_recording(self): """ Tests that the eco_account can not be edited after being recorded User must be redirected to another page Returns: """ self.assertIsNotNone(self.eco_account) self.assertFalse(self.eco_account.is_recorded) edit_url = reverse("compensation:acc:edit", args=(self.eco_account.id,)) response = self.client_user.get(edit_url) has_redirect = response.status_code == 302 self.assertFalse(has_redirect) self.eco_account.set_recorded(self.user) self.assertTrue(self.eco_account.is_recorded) edit_url = reverse("compensation:acc:edit", args=(self.eco_account.id,)) response = self.client_user.get(edit_url) has_redirect = response.status_code == 302 self.assertTrue(has_redirect) self.eco_account.set_unrecorded(self.user)