You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
450 lines
18 KiB
Python
450 lines
18 KiB
Python
"""
|
|
Author: Michel Peltriaux
|
|
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
|
Contact: michel.peltriaux@sgdnord.rlp.de
|
|
Created on: 10.11.21
|
|
|
|
"""
|
|
import datetime
|
|
|
|
from django.core.exceptions import ObjectDoesNotExist
|
|
from django.urls import reverse
|
|
|
|
from compensation.models import Payment, EcoAccountDeduction
|
|
from intervention.models import Intervention
|
|
from konova.settings import ETS_GROUP, ZB_GROUP
|
|
from konova.tests.test_views import BaseWorkflowTestCase
|
|
from user.models import UserActionLogEntry, UserAction
|
|
|
|
|
|
class InterventionWorkflowTestCase(BaseWorkflowTestCase):
|
|
""" This test case adds workflow tests
|
|
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpTestData(cls):
|
|
super().setUpTestData()
|
|
|
|
def setUp(self) -> None:
|
|
super().setUp()
|
|
# Recreate a new (bare minimum) intervention before each test
|
|
self.intervention = self.create_dummy_intervention()
|
|
self.intervention.share_with(self.superuser)
|
|
|
|
def test_new(self):
|
|
"""
|
|
Checks a 'normal' case of creating a new intervention.
|
|
We expect the user to be redirected as expected right away to the detail page of the new intervention.
|
|
We expect the user to be directly added to the shared user of the intervention
|
|
We expect that a minimum of data (identifier, title, (empty) geometry) can be used to create an intervention
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Define the intervention identifier for easier handling on the next lines
|
|
test_id = self.create_dummy_string()
|
|
test_title = self.create_dummy_string()
|
|
test_geom = self.create_dummy_geometry()
|
|
|
|
new_url = reverse("intervention:new", args=())
|
|
|
|
# Expect the new intervention does not exist yet
|
|
obj_exists = Intervention.objects.filter(
|
|
identifier=test_id
|
|
).exists()
|
|
self.assertFalse(obj_exists)
|
|
|
|
# User creates a new intervention with bare minimum content, using the proper url and post data
|
|
post_data = {
|
|
"identifier": test_id,
|
|
"title": test_title,
|
|
"geom": test_geom.geojson,
|
|
}
|
|
response = self.client_user.post(
|
|
new_url,
|
|
post_data
|
|
)
|
|
|
|
# Now expect the new intervention to exist in the db
|
|
try:
|
|
obj = Intervention.objects.get(
|
|
identifier=test_id
|
|
)
|
|
self.assertEqual(obj.identifier, test_id)
|
|
self.assertEqual(obj.title, test_title)
|
|
self.assert_equal_geometries(obj.geometry.geom, test_geom)
|
|
except ObjectDoesNotExist:
|
|
# Fail if there is no such object
|
|
self.fail()
|
|
|
|
expected_redirect = reverse("intervention:detail", args=(obj.id,))
|
|
# Expect redirect to the detail view of the new intervention
|
|
self.assertRedirects(response, expected_redirect)
|
|
|
|
# Expect user to be first and only user with shared access
|
|
self.assertIn(self.superuser, obj.users.all())
|
|
self.assertEqual(1, obj.users.count())
|
|
|
|
def test_checkability(self):
|
|
""" Tests that the intervention can only be checked if all required data has been added
|
|
|
|
Returns:
|
|
|
|
"""
|
|
check_url = reverse("intervention:check", args=(self.intervention.id,))
|
|
post_data = {
|
|
"checked_intervention": True,
|
|
"checked_comps": True,
|
|
}
|
|
|
|
# First of all, the intervention should not be checked, yet
|
|
if self.intervention.checked:
|
|
self.intervention.checked.delete()
|
|
self.intervention.refresh_from_db()
|
|
|
|
# Make sure the dummy compensation is currently not linked to the intervention,
|
|
# since the system would check on it's quality as well (and it would fail)
|
|
self.intervention.compensations.set([])
|
|
|
|
# Run request with an incomplete intervention and missing user privileges --> expect to fail
|
|
self.client_user.post(check_url, post_data)
|
|
|
|
# We expect that the intervention is still not checked now
|
|
self.intervention.refresh_from_db()
|
|
self.assertIsNone(self.intervention.checked)
|
|
|
|
# Now give the user the required privileges by adding to the registration office group
|
|
group = self.groups.get(name=ZB_GROUP)
|
|
self.superuser.groups.add(group)
|
|
|
|
# Now fill in the missing data, so the intervention is 'valid' for checking
|
|
self.intervention = self.fill_out_intervention(self.intervention)
|
|
|
|
# Then add a dummy payment, so we pass the quality check (Checks whether any kind of valid compensation exists)
|
|
payment = Payment.objects.create(amount=10.00, due_on=None, comment="No due date because test")
|
|
self.intervention.payments.add(payment)
|
|
|
|
# Run request again
|
|
self.client_user.post(check_url, post_data)
|
|
|
|
# Update intervention from db
|
|
self.intervention.refresh_from_db()
|
|
|
|
# We expect the intervention to be checked now and contains the proper data
|
|
# Attention: We check the timestamp only on the date, not the time, since the microseconds delay would result
|
|
# in an unwanted assertion error
|
|
checked = self.intervention.checked
|
|
self.assertIsNotNone(checked)
|
|
self.assertEqual(self.superuser, checked.user)
|
|
self.assertEqual(datetime.date.today(), checked.timestamp.date())
|
|
self.assertEqual(UserAction.CHECKED, checked.action)
|
|
|
|
# Expect the user action now to live in the log
|
|
self.assertIn(checked, self.intervention.log.all())
|
|
|
|
def test_recordability(self):
|
|
""" Tests that the intervention can only be recorded if all required data has been added
|
|
|
|
Returns:
|
|
|
|
"""
|
|
record_url = reverse("intervention:record", args=(self.intervention.id,))
|
|
post_data = {
|
|
"confirm": True,
|
|
}
|
|
|
|
# Make sure the dummy compensation is currently not linked to the intervention,
|
|
# since we would check on it's quality as well then
|
|
self.intervention.compensations.set([])
|
|
|
|
# First of all, the intervention should not be recorded, yet
|
|
if self.intervention.recorded:
|
|
self.intervention.recorded.delete()
|
|
self.intervention.refresh_from_db()
|
|
|
|
# Run request with an incomplete intervention and missing user privileges --> expect to fail
|
|
self.client_user.post(record_url, post_data)
|
|
|
|
# We expect that the intervention is still not recorded now
|
|
self.intervention.refresh_from_db()
|
|
self.assertIsNone(self.intervention.recorded)
|
|
|
|
# Now give the user the required privileges by adding to the ETS group
|
|
group = self.groups.get(name=ETS_GROUP)
|
|
self.superuser.groups.add(group)
|
|
|
|
# Now fill in the missing data, so the intervention is 'valid' for recording
|
|
self.intervention = self.fill_out_intervention(self.intervention)
|
|
|
|
# Then add a dummy payment, so we pass the quality check (Checks whether any kind of valid compensation exists)
|
|
payment = Payment.objects.create(amount=10.00, due_on=None, comment="No due date because test")
|
|
self.intervention.payments.add(payment)
|
|
|
|
# Run request again
|
|
self.client_user.post(record_url, post_data)
|
|
|
|
# Update intervention from db
|
|
self.intervention.refresh_from_db()
|
|
|
|
# We expect the intervention to be recorded now and contains the proper data
|
|
# Attention: We check the timestamp only on the date, not the time, since the microseconds delay would result
|
|
# in an unwanted assertion error
|
|
self.assertIsNotNone(self.intervention.recorded)
|
|
self.assertEqual(self.superuser, self.intervention.recorded.user)
|
|
self.assertEqual(datetime.date.today(), self.intervention.recorded.timestamp.date())
|
|
self.assertEqual(UserAction.RECORDED, self.intervention.recorded.action)
|
|
|
|
# Expect the user action now to live in the log
|
|
self.assertIn(self.intervention.recorded, self.intervention.log.all())
|
|
|
|
def subtest_add_payment(self):
|
|
""" Subroutine for 'normal' payment tests
|
|
|
|
Checks a 'normal' case of adding a payment.
|
|
We expect a new payment to be addable to an existing intervention
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Attention: Despite the fact, this url refers to a compensation app route, we test it here for the interventions.
|
|
# Reason: A payment is some kind of compensation for an intervention. Therefore it lives inside the compensation app.
|
|
# BUT: Payments are added on the intervention detail page. Therefore it's part of a regular intervention workflow.
|
|
new_payment_url = reverse("compensation:pay:new", args=(self.intervention.id,))
|
|
|
|
# Make sure there are no payments on the intervention, yet
|
|
self.assertEqual(0, self.intervention.payments.count())
|
|
|
|
# Create form data to be sent to the url
|
|
test_amount = 10.00
|
|
test_due = "2021-01-01"
|
|
test_comment = self.create_dummy_string()
|
|
post_data = {
|
|
"amount": test_amount,
|
|
"due": test_due,
|
|
"comment": test_comment
|
|
}
|
|
self.client_user.post(
|
|
new_payment_url,
|
|
post_data,
|
|
)
|
|
# We do not test for any redirects in here, since the new payment url is realized using a modal, which does not
|
|
# perform any direct redirects but instead reloads the page after finisihing.
|
|
|
|
# Make sure there is a new payment on the intervention now
|
|
self.assertEqual(1, self.intervention.payments.count())
|
|
|
|
# Make sure the payment contains our data
|
|
payment = self.intervention.payments.all()[0]
|
|
self.assertEqual(payment.amount, test_amount)
|
|
self.assertEqual(payment.due_on, datetime.date.fromisoformat(test_due))
|
|
self.assertEqual(payment.comment, test_comment)
|
|
return payment
|
|
|
|
def subtest_delete_payment(self, payment: Payment):
|
|
""" Subroutine for 'normal' payment tests
|
|
|
|
Checks a 'normal' case of adding a payment.
|
|
We expect a payment to be deletable to an existing intervention
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Create removing url for the payment
|
|
remove_url = reverse("compensation:pay:remove", args=(payment.id,))
|
|
post_data = {
|
|
"confirm": True,
|
|
}
|
|
self.client_user.post(
|
|
remove_url,
|
|
post_data
|
|
)
|
|
|
|
# Expect the payment to be gone from the db and therefore from the intervention as well
|
|
self.assert_object_is_deleted(payment)
|
|
|
|
# Now make sure the intervention has no payments anymore
|
|
self.assertEqual(0, self.intervention.payments.count())
|
|
|
|
def test_payments(self):
|
|
"""
|
|
Checks a 'normal' case of adding a payment.
|
|
We expect a new payment to be addable to an existing intervention
|
|
We expect a payment to be deletable from an existing intervention
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Create new payment for the default intervention
|
|
payment = self.subtest_add_payment()
|
|
|
|
# Now remove the payment again
|
|
self.subtest_delete_payment(payment)
|
|
|
|
def subtest_add_deduction_fail_positive(self, new_url: str, post_data: dict, test_surface: float):
|
|
""" Holds tests for postivie fails of new deduction creation
|
|
|
|
Reasons for failing are:
|
|
* EcoAccount does not provide enough 'deductable_surface'
|
|
* EcoAccount is not recorded (not "approved"), yet
|
|
* EcoAccount is not shared with performing user
|
|
|
|
Args:
|
|
new_url (str): The url to send the post data to
|
|
post_data (dict): The form post data to be sent
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Before running fail positive tests, we need to have an account in a (normally) fine working state
|
|
self.assertIsNotNone(self.eco_account.recorded) # -> is recorded
|
|
self.assertGreater(self.eco_account.deductable_surface, test_surface) # -> has more deductable surface than we need
|
|
self.assertIn(self.superuser, self.eco_account.users.all()) # -> is shared with the performing user
|
|
|
|
# Count the number of already existing deductions in total and for the account for later comparison
|
|
num_deductions = self.eco_account.deductions.count()
|
|
num_deductions_total = EcoAccountDeduction.objects.count()
|
|
|
|
# First test that a deduction can not be created, if the account does not provide
|
|
# enough surface for the deduction. So we modify the deductable surface of the account
|
|
self.eco_account.deductable_surface = 0
|
|
self.eco_account.save()
|
|
|
|
# Now perform the (expected) failing request
|
|
self.client_user.post(new_url, post_data)
|
|
|
|
# Expect no changes at all, since the deduction should not have been created
|
|
self.assertEqual(num_deductions, self.eco_account.deductions.count())
|
|
self.assertEqual(num_deductions_total, EcoAccountDeduction.objects.count())
|
|
|
|
# Now restore the deductable surface to a valid size back again but remove the user from the shared list
|
|
self.eco_account.deductable_surface = test_surface + 100.00
|
|
self.eco_account.share_with_list([])
|
|
self.eco_account.save()
|
|
|
|
# Now perform the (expected) failing request (again)
|
|
self.client_user.post(new_url, post_data)
|
|
|
|
# Expect no changes at all, since the account is not shared
|
|
self.assertEqual(num_deductions, self.eco_account.deductions.count())
|
|
self.assertEqual(num_deductions_total, EcoAccountDeduction.objects.count())
|
|
|
|
# Restore the sharing but remove the recording state
|
|
self.eco_account.share_with_list([self.superuser])
|
|
self.eco_account.recorded.delete()
|
|
self.eco_account.refresh_from_db()
|
|
self.eco_account.save()
|
|
|
|
# Now perform the (expected) failing request (again)
|
|
self.client_user.post(new_url, post_data)
|
|
|
|
# Expect no changes at all, since the account is no shared with the user, yet
|
|
self.assertEqual(num_deductions, self.eco_account.deductions.count())
|
|
self.assertEqual(num_deductions_total, EcoAccountDeduction.objects.count())
|
|
|
|
def subtest_add_deduction_normal(self, new_url: str, post_data: dict, test_surface: float):
|
|
""" Holds tests on working ("normal") deduction creation
|
|
|
|
Args:
|
|
new_url (str): The url to send the post data to
|
|
post_data (dict): The form post data to be sent
|
|
test_surface (float): The expected surface of the deduction
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Prepare the account for a working situation (enough deductable surface, recorded and shared)
|
|
self.eco_account.deductable_surface = 10000.00
|
|
if self.eco_account.recorded is None:
|
|
rec_action = UserActionLogEntry.get_recorded_action(self.superuser)
|
|
self.eco_account.recorded = rec_action
|
|
self.eco_account.share_with_list([self.superuser])
|
|
self.eco_account.save()
|
|
num_all_deducs = EcoAccountDeduction.objects.count()
|
|
|
|
# Run the request
|
|
self.client_user.post(new_url, post_data)
|
|
|
|
# Expect the deduction to be created, since all constraints are fulfilled
|
|
self.assertEqual(1, self.eco_account.deductions.count())
|
|
self.assertEqual(num_all_deducs + 1, EcoAccountDeduction.objects.count())
|
|
|
|
# Make sure the deduction contains the expected data
|
|
deduction = EcoAccountDeduction.objects.get(
|
|
account=self.eco_account,
|
|
intervention=self.intervention
|
|
)
|
|
self.assertEqual(deduction.surface, test_surface)
|
|
|
|
# Return deduction for further usage in tests
|
|
return deduction
|
|
|
|
def subtest_add_deduction(self):
|
|
""" Holds test for adding a new deduction
|
|
|
|
Contains tests for
|
|
* positive fails (as expected)
|
|
* normal cases
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Create the url for creating a new deduction
|
|
new_url = reverse("compensation:acc:new-deduction", args=(self.eco_account.id,))
|
|
|
|
# Prepare the form data
|
|
test_surface = 100.00
|
|
post_data = {
|
|
"surface": test_surface,
|
|
"account": self.eco_account.id,
|
|
"intervention": self.intervention.id,
|
|
}
|
|
# Run some tests for regular, working cases
|
|
deduction = self.subtest_add_deduction_normal(new_url, post_data, test_surface)
|
|
|
|
# Run some tests where we expect the creation of a deduction to fail (as expected)
|
|
self.subtest_add_deduction_fail_positive(new_url, post_data, test_surface)
|
|
|
|
# Return deduction for further usage in tests
|
|
return deduction
|
|
|
|
def subtest_delete_deduction(self, deduction: EcoAccountDeduction):
|
|
""" Holds test for deleting a deduction
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Prepare url for deleting of this deduction
|
|
delete_url = reverse("compensation:acc:remove-deduction", args=(self.eco_account.id, deduction.id,))
|
|
post_data = {
|
|
"confirm": True
|
|
}
|
|
# Save number of current deductions for later comparison
|
|
num_deductions = self.eco_account.deductions.count()
|
|
num_deductions_total = EcoAccountDeduction.objects.count()
|
|
|
|
# Run request
|
|
self.client_user.post(delete_url, post_data)
|
|
|
|
# Expect the deduction to be gone from the db and relations
|
|
self.assertEqual(num_deductions - 1, self.eco_account.deductions.count())
|
|
self.assertEqual(num_deductions_total - 1, EcoAccountDeduction.objects.count())
|
|
|
|
# Expect the deduction to be totally gone
|
|
self.assert_object_is_deleted(deduction)
|
|
|
|
def test_deduction(self):
|
|
"""
|
|
Checks a 'normal case of adding a deduction.
|
|
We expect a new deduction to be addable to an existing intervention
|
|
We expect a deduction to be deletable
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Create a new deduction for the default intervention
|
|
deduction = self.subtest_add_deduction()
|
|
|
|
# Now remove the deduction again
|
|
self.subtest_delete_deduction(deduction)
|