IT-Naturschutz/konova
7
0
Fork 0
Code Issues 7 Pull Requests Actions Projects Releases 47 Wiki Activity

Merge pull request 'master' (#113) from master into Docker

Browse Source 
Reviewed-on: SGD-Nord/konova#113
This commit is contained in:
mpeltriaux
2022-02-11 16:10:47 +01:00
parent b4e75fa2cd 68b8ff07e9
commit 2fcc41bf4a
86 changed files with 3700 additions and 1381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
konova/autocompletes.py
View File

@@ -35,8 +35,9 @@ class EcoAccountAutocomplete(Select2QuerySetView):
)
if self.q:
qs = qs.filter(
identifier__icontains=self.q
)
Q(identifier__icontains=self.q) |
Q(title__icontains=self.q)
).distinct()
return qs
@@ -57,8 +58,9 @@ class InterventionAutocomplete(Select2QuerySetView):
)
if self.q:
qs = qs.filter(
identifier__icontains=self.q
)
Q(identifier__icontains=self.q) |
Q(title__icontains=self.q)
).distinct()
return qs
@@ -81,8 +83,9 @@ class ShareUserAutocomplete(Select2QuerySetView):
if self.q:
# Due to privacy concerns only a full username match will return the proper user entry
qs = qs.filter(
username=self.q
)
Q(username=self.q) |
Q(email=self.q)
).distinct()
return qs

79
konova/forms.py
View File

@@ -12,6 +12,8 @@ from bootstrap_modal_forms.forms import BSModalForm
from bootstrap_modal_forms.utils import is_ajax
from django import forms
from django.contrib import messages
from django.db.models.fields.files import FieldFile
from user.models import User
from django.contrib.gis.forms import OSMWidget, MultiPolygonField
from django.contrib.gis.geos import MultiPolygon
@@ -21,10 +23,10 @@ from django.shortcuts import render
from django.utils.translation import gettext_lazy as _
from konova.contexts import BaseContext
from konova.models import BaseObject, Geometry, RecordableObjectMixin
from konova.models import BaseObject, Geometry, RecordableObjectMixin, AbstractDocument
from konova.settings import DEFAULT_SRID
from konova.tasks import celery_update_parcels
from konova.utils.message_templates import FORM_INVALID
from konova.utils.message_templates import FORM_INVALID, FILE_TYPE_UNSUPPORTED, FILE_SIZE_TOO_LARGE, DOCUMENT_EDITED
from user.models import UserActionLogEntry
@@ -87,7 +89,7 @@ class BaseForm(forms.Form):
"""
self.fields[field].widget.attrs["placeholder"] = val
def load_initial_data(self, form_data: dict, disabled_fields: list):
def load_initial_data(self, form_data: dict, disabled_fields: list = None):
""" Initializes form data from instance
Inserts instance data into form and disables form fields
@@ -99,8 +101,9 @@ class BaseForm(forms.Form):
return
for k, v in form_data.items():
self.initialize_form_field(k, v)
for field in disabled_fields:
self.disable_form_field(field)
if disabled_fields:
for field in disabled_fields:
self.disable_form_field(field)
def add_widget_html_class(self, field: str, cls: str):
""" Adds a HTML class string to the widget of a field
@@ -327,10 +330,27 @@ class RemoveModalForm(BaseModalForm):
self.instance.mark_as_deleted(self.user)
else:
# If the class does not provide restorable delete functionality, we must delete the entry finally
self.instance.delete(self.user)
self.instance.delete()
class NewDocumentForm(BaseModalForm):
class RemoveDeadlineModalForm(RemoveModalForm):
""" Removing modal form for deadlines
Can be used for anything, where removing shall be confirmed by the user a second time.
"""
deadline = None
def __init__(self, *args, **kwargs):
deadline = kwargs.pop("deadline", None)
self.deadline = deadline
super().__init__(*args, **kwargs)
def save(self):
self.instance.remove_deadline(self)
class NewDocumentModalForm(BaseModalForm):
""" Modal form for new documents
"""
@@ -402,18 +422,22 @@ class NewDocumentForm(BaseModalForm):
_file = self.cleaned_data.get("file", None)
if _file is None or isinstance(_file, FieldFile):
# FieldFile declares that no new file has been uploaded and we do not need to check on the file again
return super_valid
mime_type_valid = self.document_model.is_mime_type_valid(_file)
if not mime_type_valid:
self.add_error(
"file",
_("Unsupported file type")
FILE_TYPE_UNSUPPORTED
)
file_size_valid = self.document_model.is_file_size_valid(_file)
if not file_size_valid:
self.add_error(
"file",
_("File too large")
FILE_SIZE_TOO_LARGE
)
file_valid = mime_type_valid and file_size_valid
@@ -440,6 +464,39 @@ class NewDocumentForm(BaseModalForm):
return doc
class EditDocumentModalForm(NewDocumentModalForm):
document = None
document_model = AbstractDocument
def __init__(self, *args, **kwargs):
self.document = kwargs.pop("document", None)
super().__init__(*args, **kwargs)
form_data = {
"title": self.document.title,
"comment": self.document.comment,
"creation_date": str(self.document.date_of_creation),
"file": self.document.file,
}
self.load_initial_data(form_data)
def save(self):
with transaction.atomic():
document = self.document
file = self.cleaned_data.get("file", None)
document.title = self.cleaned_data.get("title", None)
document.comment = self.cleaned_data.get("comment", None)
document.date_of_creation = self.cleaned_data.get("creation_date", None)
if not isinstance(file, FieldFile):
document.replace_file(file)
document.save()
self.instance.mark_as_edited(self.user, self.request, edit_comment=DOCUMENT_EDITED)
return document
class RecordModalForm(BaseModalForm):
""" Modal form for recording data
@@ -515,7 +572,9 @@ class RecordModalForm(BaseModalForm):
Returns:
"""
comps = self.instance.compensations.all()
comps = self.instance.compensations.filter(
deleted=None,
)
comps_valid = True
for comp in comps:
checker = comp.quality_check()

54
konova/migrations/0003_auto_20220208_1801.py Normal file
View File

@@ -0,0 +1,54 @@
# Generated by Django 3.1.3 on 2022-02-08 17:01
from django.db import migrations, models
import django.db.models.deletion
import uuid
def migrate_parcels(apps, schema_editor):
Geometry = apps.get_model('konova', 'Geometry')
SpatialIntersection = apps.get_model('konova', 'SpatialIntersection')
all_geoms = Geometry.objects.all()
for geom in all_geoms:
SpatialIntersection.objects.bulk_create([
SpatialIntersection(geometry=geom, parcel=parcel)
for parcel in geom.parcels.all()
])
class Migration(migrations.Migration):
dependencies = [
('konova', '0002_auto_20220114_0936'),
]
operations = [
migrations.CreateModel(
name='SpatialIntersection',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('calculated_on', models.DateTimeField(auto_now_add=True, null=True)),
('geometry', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='konova.geometry')),
('parcel', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='konova.parcel')),
],
options={
'abstract': False,
},
),
migrations.RunPython(migrate_parcels),
migrations.AddField(
model_name='parcel',
name='geometries_tmp',
field=models.ManyToManyField(blank=True, related_name='parcels', through='konova.SpatialIntersection', to='konova.Geometry'),
),
migrations.RemoveField(
model_name='parcel',
name='geometries',
),
migrations.RenameField(
model_name='parcel',
old_name='geometries_tmp',
new_name='geometries',
),
]

17
konova/migrations/0004_auto_20220209_0839.py Normal file
View File

@@ -0,0 +1,17 @@
# Generated by Django 3.1.3 on 2022-02-09 07:39
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('konova', '0003_auto_20220208_1801'),
]
operations = [
migrations.RenameModel(
old_name='SpatialIntersection',
new_name='ParcelIntersection',
),
]

16
konova/models/document.py
View File

@@ -101,3 +101,19 @@ class AbstractDocument(BaseResource):
def is_file_size_valid(cls, _file):
max_size = cls._maximum_file_size * pow(1000, 2)
return _file.size <= max_size
def replace_file(self, new_file):
""" Replaces the old file on the hard drive with the new one
Args:
new_file (File): The new file
Returns:
"""
try:
os.remove(self.file.file.name)
except FileNotFoundError:
pass
self.file = new_file
self.save()

23
konova/models/geometry.py
View File

@@ -99,7 +99,7 @@ class Geometry(BaseResource):
Returns:
"""
from konova.models import Parcel, District
from konova.models import Parcel, District, ParcelIntersection
parcel_fetcher = ParcelWFSFetcher(
geometry_id=self.id,
)
@@ -107,6 +107,7 @@ class Geometry(BaseResource):
fetched_parcels = parcel_fetcher.get_features(
typename
)
_now = timezone.now()
underlying_parcels = []
for result in fetched_parcels:
fetched_parcel = result[typename]
@@ -125,19 +126,35 @@ class Geometry(BaseResource):
krs=fetched_parcel["ave:kreis"],
)[0]
parcel_obj.district = district
parcel_obj.updated_on = timezone.now()
parcel_obj.updated_on = _now
parcel_obj.save()
underlying_parcels.append(parcel_obj)
# Update the linked parcels
self.parcels.set(underlying_parcels)
# Set the calculated_on intermediate field, so this related data will be found on lookups
intersections_without_ts = self.parcelintersection_set.filter(
parcel__in=self.parcels.all(),
calculated_on__isnull=True,
)
for entry in intersections_without_ts:
entry.calculated_on = _now
ParcelIntersection.objects.bulk_update(
intersections_without_ts,
["calculated_on"]
)
def get_underlying_parcels(self):
""" Getter for related parcels and their districts
Returns:
parcels (QuerySet): The related parcels as queryset
"""
parcels = self.parcels.all().prefetch_related(
parcels = self.parcels.filter(
parcelintersection__calculated_on__isnull=False,
).prefetch_related(
"district"
).order_by(
"gmrkng",

79
konova/models/object.py
View File

@@ -12,6 +12,7 @@ from abc import abstractmethod
from django.contrib import messages
from django.db.models import QuerySet
from konova.sub_settings.lanis_settings import DEFAULT_SRID_RLP, LANIS_ZOOM_LUT, LANIS_LINK_TEMPLATE
from konova.tasks import celery_send_mail_shared_access_removed, celery_send_mail_shared_access_given, \
celery_send_mail_shared_data_recorded, celery_send_mail_shared_data_unrecorded, \
celery_send_mail_shared_data_deleted, celery_send_mail_shared_data_checked
@@ -128,10 +129,27 @@ class BaseObject(BaseResource):
# Send mail
shared_users = self.shared_users.values_list("id", flat=True)
for user_id in shared_users:
celery_send_mail_shared_data_deleted.delay(self.identifier, user_id)
celery_send_mail_shared_data_deleted.delay(self.identifier, self.title, user_id)
self.save()
def mark_as_edited(self, performing_user: User, request: HttpRequest = None, edit_comment: str = None):
""" In case the object or a related object changed the log history needs to be updated
Args:
performing_user (User): The user which performed the editing action
request (HttpRequest): The used request for this action
edit_comment (str): Additional comment for the log entry
Returns:
"""
edit_action = UserActionLogEntry.get_edited_action(performing_user, edit_comment)
self.modified = edit_action
self.log.add(edit_action)
self.save()
return edit_action
def add_log_entry(self, action: UserAction, user: User, comment: str):
""" Wraps adding of UserActionLogEntry to log
@@ -200,6 +218,10 @@ class BaseObject(BaseResource):
_str = "{}{}-{}".format(curr_month, curr_year, rand_str)
return definitions[self.__class__]["template"].format(_str)
@abstractmethod
def get_detail_url(self):
raise NotImplementedError()
class RecordableObjectMixin(models.Model):
""" Wraps record related fields and functionality
@@ -236,7 +258,7 @@ class RecordableObjectMixin(models.Model):
shared_users = self.users.all().values_list("id", flat=True)
for user_id in shared_users:
celery_send_mail_shared_data_unrecorded.delay(self.identifier, user_id)
celery_send_mail_shared_data_unrecorded.delay(self.identifier, self.title, user_id)
return action
@@ -258,29 +280,22 @@ class RecordableObjectMixin(models.Model):
shared_users = self.users.all().values_list("id", flat=True)
for user_id in shared_users:
celery_send_mail_shared_data_recorded.delay(self.identifier, user_id)
celery_send_mail_shared_data_recorded.delay(self.identifier, self.title, user_id)
return action
def mark_as_edited(self, performing_user: User, request: HttpRequest = None, edit_comment: str = None, reset_recorded: bool = True):
""" In case the object or a related object changed, internal processes need to be started, such as
unrecord and uncheck
def unrecord(self, performing_user: User, request: HttpRequest = None):
""" Unrecords a dataset
Args:
performing_user (User): The user which performed the editing action
request (HttpRequest): The used request for this action
edit_comment (str): Additional comment for the log entry
reset_recorded (bool): Whether the record-state of the object should be reset
Returns:
"""
edit_action = UserActionLogEntry.get_edited_action(performing_user, edit_comment)
self.modified = edit_action
self.log.add(edit_action)
self.save()
if self.recorded and reset_recorded:
action = None
if self.recorded:
action = self.set_unrecorded(performing_user)
self.log.add(action)
if request:
@@ -288,7 +303,7 @@ class RecordableObjectMixin(models.Model):
request,
CHECKED_RECORDED_RESET
)
return edit_action
return action
@abstractmethod
def is_ready_for_publish(self) -> bool:
@@ -350,7 +365,7 @@ class CheckableObjectMixin(models.Model):
# Send mail
shared_users = self.users.all().values_list("id", flat=True)
for user_id in shared_users:
celery_send_mail_shared_data_checked.delay(self.identifier, user_id)
celery_send_mail_shared_data_checked.delay(self.identifier, self.title, user_id)
self.log.add(action)
return action
@@ -464,9 +479,9 @@ class ShareableObjectMixin(models.Model):
# Send mails
for user in removed_users:
celery_send_mail_shared_access_removed.delay(self.identifier, user["id"])
celery_send_mail_shared_access_removed.delay(self.identifier, self.title, user["id"])
for user in new_accessing_users:
celery_send_mail_shared_access_given.delay(self.identifier, user)
celery_send_mail_shared_access_given.delay(self.identifier, self.title, user)
# Set new shared users
self.share_with_list(users)
@@ -530,3 +545,31 @@ class GeoReferencedMixin(models.Model):
message_str = GEOMETRY_CONFLICT_WITH_TEMPLATE.format(instance_identifiers)
messages.info(request, message_str)
return request
def get_LANIS_link(self) -> str:
""" Generates a link for LANIS depending on the geometry
Returns:
"""
try:
geom = self.geometry.geom.transform(DEFAULT_SRID_RLP, clone=True)
x = geom.centroid.x
y = geom.centroid.y
area = int(geom.envelope.area)
z_l = 16
for k_area, v_zoom in LANIS_ZOOM_LUT.items():
if k_area < area:
z_l = v_zoom
break
zoom_lvl = z_l
except (AttributeError, IndexError) as e:
# If no geometry has been added, yet.
x = 1
y = 1
zoom_lvl = 6
return LANIS_LINK_TEMPLATE.format(
zoom_lvl,
x,
y,
)

21
konova/models/parcel.py
View File

@@ -22,7 +22,7 @@ class Parcel(UuidModel):
To avoid conflicts due to german Umlaute, the field names are shortened and vocals are dropped.
"""
geometries = models.ManyToManyField("konova.Geometry", related_name="parcels", blank=True)
geometries = models.ManyToManyField("konova.Geometry", blank=True, related_name="parcels", through='ParcelIntersection')
district = models.ForeignKey("konova.District", on_delete=models.SET_NULL, null=True, blank=True, related_name="parcels")
gmrkng = models.CharField(
max_length=1000,
@@ -77,3 +77,22 @@ class District(UuidModel):
def __str__(self):
return f"{self.gmnd} | {self.krs}"
class ParcelIntersection(UuidModel):
""" ParcelIntersection is an intermediary model, which is used to configure the
M2M relation between Parcel and Geometry.
Based on uuids, we will not have (practically) any problems on outrunning primary keys
and extending the model with calculated_on timestamp, we can 'hide' entries while they
are being recalculated and keep track on the last time they have been calculated this
way.
Please note: The calculated_on describes when the relation between the Parcel and the Geometry
has been established. The updated_on field of Parcel describes when this Parcel has been
changed the last time.
"""
parcel = models.ForeignKey(Parcel, on_delete=models.CASCADE)
geometry = models.ForeignKey("konova.Geometry", on_delete=models.CASCADE)
calculated_on = models.DateTimeField(auto_now_add=True, null=True, blank=True)

7
konova/static/css/konova.css
View File

@@ -219,6 +219,13 @@ Overwrites bootstrap .btn:focus box shadow color
overflow: auto;
}
.w-20{
width: 20%;
}
.w-10{
width: 20%;
}
/*
Extends css for django autocomplete light (dal)
No other approach worked to get the autocomplete fields to full width of parent containers

36
konova/tasks.py
View File

@@ -4,13 +4,19 @@ from celery import shared_task
from django.core.exceptions import ObjectDoesNotExist
@shared_task
def celery_update_parcels(geometry_id: str, recheck: bool = True):
from konova.models import Geometry
from konova.models import Geometry, ParcelIntersection
try:
geom = Geometry.objects.get(id=geometry_id)
geom.parcels.clear()
objs = geom.parcelintersection_set.all()
for obj in objs:
obj.calculated_on = None
ParcelIntersection.objects.bulk_update(
objs,
["calculated_on"]
)
geom.update_parcels()
except ObjectDoesNotExist:
if recheck:
@@ -19,42 +25,42 @@ def celery_update_parcels(geometry_id: str, recheck: bool = True):
@shared_task
def celery_send_mail_shared_access_removed(obj_identifier, user_id):
def celery_send_mail_shared_access_removed(obj_identifier, obj_title=None, user_id=None):
from user.models import User
user = User.objects.get(id=user_id)
user.send_mail_shared_access_removed(obj_identifier)
user.send_mail_shared_access_removed(obj_identifier, obj_title)
@shared_task
def celery_send_mail_shared_access_given(obj_identifier, user_id):
def celery_send_mail_shared_access_given(obj_identifier, obj_title=None, user_id=None):
from user.models import User
user = User.objects.get(id=user_id)
user.send_mail_shared_access_given(obj_identifier)
user.send_mail_shared_access_given(obj_identifier, obj_title)
@shared_task
def celery_send_mail_shared_data_recorded(obj_identifier, user_id):
def celery_send_mail_shared_data_recorded(obj_identifier, obj_title=None, user_id=None):
from user.models import User
user = User.objects.get(id=user_id)
user.send_mail_shared_data_recorded(obj_identifier)
user.send_mail_shared_data_recorded(obj_identifier, obj_title)
@shared_task
def celery_send_mail_shared_data_unrecorded(obj_identifier, user_id):
def celery_send_mail_shared_data_unrecorded(obj_identifier, obj_title=None, user_id=None):
from user.models import User
user = User.objects.get(id=user_id)
user.send_mail_shared_data_unrecorded(obj_identifier)
user.send_mail_shared_data_unrecorded(obj_identifier, obj_title)
@shared_task
def celery_send_mail_shared_data_deleted(obj_identifier, user_id):
def celery_send_mail_shared_data_deleted(obj_identifier, obj_title=None, user_id=None):
from user.models import User
user = User.objects.get(id=user_id)
user.send_mail_shared_data_deleted(obj_identifier)
user.send_mail_shared_data_deleted(obj_identifier, obj_title)
@shared_task
def celery_send_mail_shared_data_checked(obj_identifier, user_id):
def celery_send_mail_shared_data_checked(obj_identifier, obj_title=None, user_id=None):
from user.models import User
user = User.objects.get(id=user_id)
user.send_mail_shared_data_checked(obj_identifier)
user.send_mail_shared_data_checked(obj_identifier, obj_title)

205
konova/tests/test_views.py
View File

@@ -7,6 +7,7 @@ Created on: 26.10.21
"""
import datetime
from codelist.settings import CODELIST_CONSERVATION_OFFICE_ID
from ema.models import Ema
from user.models import User
from django.contrib.auth.models import Group
@@ -15,7 +16,7 @@ from django.core.exceptions import ObjectDoesNotExist
from django.test import TestCase, Client
from django.urls import reverse
from codelist.models import KonovaCode
from codelist.models import KonovaCode, KonovaCodeList
from compensation.models import Compensation, CompensationState, CompensationAction, EcoAccount, EcoAccountDeduction
from intervention.models import Legal, Responsibility, Intervention
from konova.management.commands.setup_data import GROUPS_DATA
@@ -46,43 +47,58 @@ class BaseTestCase(TestCase):
class Meta:
abstract = True
@classmethod
def setUpTestData(cls):
cls.create_users()
cls.create_groups()
cls.intervention = cls.create_dummy_intervention()
cls.compensation = cls.create_dummy_compensation()
cls.eco_account = cls.create_dummy_eco_account()
cls.ema = cls.create_dummy_ema()
cls.deduction = cls.create_dummy_deduction()
cls.create_dummy_states()
cls.create_dummy_action()
cls.codes = cls.create_dummy_codes()
def setUp(self) -> None:
""" Setup data before each test run
@classmethod
def create_users(cls):
Returns:
"""
super().setUp()
self.create_users()
self.create_groups()
self.intervention = self.create_dummy_intervention()
self.compensation = self.create_dummy_compensation()
self.eco_account = self.create_dummy_eco_account()
self.ema = self.create_dummy_ema()
self.deduction = self.create_dummy_deduction()
self.create_dummy_states()
self.create_dummy_action()
self.codes = self.create_dummy_codes()
# Set the default group as only group for the user
default_group = self.groups.get(name=DEFAULT_GROUP)
self.superuser.groups.set([default_group])
# Create fresh logged in client and a non-logged in client (anon) for each test
self.client_user = Client()
self.client_user.login(username=self.superuser.username, password=self.superuser_pw)
self.client_anon = Client()
def create_users(self):
# Create superuser and regular user
cls.superuser = User.objects.create_superuser(
self.superuser = User.objects.create_superuser(
username="root",
email="root@root.com",
password=cls.superuser_pw,
password=self.superuser_pw,
)
cls.user = User.objects.create_user(
self.user = User.objects.create_user(
username="user1",
email="user@root.com",
password=cls.user_pw
password=self.user_pw
)
cls.users = User.objects.all()
self.users = User.objects.all()
@classmethod
def create_groups(cls):
def create_groups(self):
# Create groups
for group_data in GROUPS_DATA:
name = group_data.get("name")
Group.objects.get_or_create(
name=name,
)
cls.groups = Group.objects.all()
self.groups = Group.objects.all()
@staticmethod
def create_dummy_string(prefix: str = ""):
@@ -93,8 +109,7 @@ class BaseTestCase(TestCase):
"""
return f"{prefix}{generate_random_string(3, True)}"
@classmethod
def create_dummy_intervention(cls):
def create_dummy_intervention(self):
""" Creates an intervention which can be used for tests
Returns:
@@ -102,7 +117,7 @@ class BaseTestCase(TestCase):
"""
# Create dummy data
# Create log entry
action = UserActionLogEntry.get_created_action(cls.superuser)
action = UserActionLogEntry.get_created_action(self.superuser)
# Create legal data object (without M2M laws first)
legal_data = Legal.objects.create()
# Create responsible data object
@@ -121,32 +136,30 @@ class BaseTestCase(TestCase):
intervention.generate_access_token(make_unique=True)
return intervention
@classmethod
def create_dummy_compensation(cls):
def create_dummy_compensation(self):
""" Creates a compensation which can be used for tests
Returns:
"""
if cls.intervention is None:
cls.intervention = cls.create_dummy_intervention()
if self.intervention is None:
self.intervention = self.create_dummy_intervention()
# Create dummy data
# Create log entry
action = UserActionLogEntry.get_created_action(cls.superuser)
action = UserActionLogEntry.get_created_action(self.superuser)
geometry = Geometry.objects.create()
# Finally create main object, holding the other objects
compensation = Compensation.objects.create(
identifier="TEST",
title="Test_title",
intervention=cls.intervention,
intervention=self.intervention,
created=action,
geometry=geometry,
comment="Test",
)
return compensation
@classmethod
def create_dummy_eco_account(cls):
def create_dummy_eco_account(self):
""" Creates an eco account which can be used for tests
Returns:
@@ -154,7 +167,7 @@ class BaseTestCase(TestCase):
"""
# Create dummy data
# Create log entry
action = UserActionLogEntry.get_created_action(cls.superuser)
action = UserActionLogEntry.get_created_action(self.superuser)
geometry = Geometry.objects.create()
# Create responsible data object
lega_data = Legal.objects.create()
@@ -171,8 +184,7 @@ class BaseTestCase(TestCase):
)
return eco_account
@classmethod
def create_dummy_ema(cls):
def create_dummy_ema(self):
""" Creates an ema which can be used for tests
Returns:
@@ -180,7 +192,7 @@ class BaseTestCase(TestCase):
"""
# Create dummy data
# Create log entry
action = UserActionLogEntry.get_created_action(cls.superuser)
action = UserActionLogEntry.get_created_action(self.superuser)
geometry = Geometry.objects.create()
# Create responsible data object
responsible_data = Responsibility.objects.create()
@@ -195,51 +207,47 @@ class BaseTestCase(TestCase):
)
return ema
@classmethod
def create_dummy_deduction(cls):
def create_dummy_deduction(self):
return EcoAccountDeduction.objects.create(
account=cls.create_dummy_eco_account(),
intervention=cls.create_dummy_intervention(),
account=self.create_dummy_eco_account(),
intervention=self.create_dummy_intervention(),
surface=100,
)
@classmethod
def create_dummy_states(cls):
def create_dummy_states(self):
""" Creates an intervention which can be used for tests
Returns:
"""
cls.comp_state = CompensationState.objects.create(
self.comp_state = CompensationState.objects.create(
surface=10.00,
biotope_type=None,
)
return cls.comp_state
return self.comp_state
@classmethod
def create_dummy_action(cls):
def create_dummy_action(self):
""" Creates an intervention which can be used for tests
Returns:
"""
cls.comp_action = CompensationAction.objects.create(
self.comp_action = CompensationAction.objects.create(
amount=10
)
return cls.comp_action
return self.comp_action
@classmethod
def create_dummy_codes(cls):
def create_dummy_codes(self):
""" Creates some dummy KonovaCodes which can be used for testing
Returns:
"""
codes = KonovaCode.objects.bulk_create([
KonovaCode(id=1, is_selectable=True, long_name="Test1"),
KonovaCode(id=2, is_selectable=True, long_name="Test2"),
KonovaCode(id=3, is_selectable=True, long_name="Test3"),
KonovaCode(id=4, is_selectable=True, long_name="Test4"),
KonovaCode(id=1, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test1"),
KonovaCode(id=2, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test2"),
KonovaCode(id=3, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test3"),
KonovaCode(id=4, is_selectable=True, is_archived=False, is_leaf=True, long_name="Test4"),
])
return codes
@@ -255,8 +263,7 @@ class BaseTestCase(TestCase):
polygon = polygon.transform(3857, clone=True)
return MultiPolygon(polygon, srid=3857) # 3857 is the default srid used for MultiPolygonField in the form
@classmethod
def fill_out_intervention(cls, intervention: Intervention) -> Intervention:
def fill_out_intervention(self, intervention: Intervention) -> Intervention:
""" Adds all required (dummy) data to an intervention
Args:
@@ -276,13 +283,12 @@ 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 = cls.create_dummy_geometry()
intervention.geometry.geom = self.create_dummy_geometry()
intervention.geometry.save()
intervention.save()
return intervention
@classmethod
def fill_out_compensation(cls, compensation: Compensation) -> Compensation:
def fill_out_compensation(self, compensation: Compensation) -> Compensation:
""" Adds all required (dummy) data to a compensation
Args:
@@ -291,13 +297,62 @@ class BaseTestCase(TestCase):
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.after_states.add(self.comp_state)
compensation.before_states.add(self.comp_state)
compensation.actions.add(self.comp_action)
compensation.geometry.geom = self.create_dummy_geometry()
compensation.geometry.save()
return compensation
def get_conservation_office_code(self):
""" Returns a dummy KonovaCode as conservation office code
Returns:
"""
codelist = KonovaCodeList.objects.get_or_create(
id=CODELIST_CONSERVATION_OFFICE_ID
)[0]
code = KonovaCode.objects.get(id=2)
codelist.codes.add(code)
return code
def fill_out_ema(self, ema):
""" Adds all required (dummy) data to an Ema
Returns:
"""
ema.responsible.conservation_office = self.get_conservation_office_code()
ema.responsible.conservation_file_number = "test"
ema.responsible.handler = "handler"
ema.responsible.save()
ema.after_states.add(self.comp_state)
ema.before_states.add(self.comp_state)
ema.actions.add(self.comp_action)
ema.geometry.geom = self.create_dummy_geometry()
ema.geometry.save()
return ema
def fill_out_eco_account(self, eco_account):
""" Adds all required (dummy) data to an EcoAccount
Returns:
"""
eco_account.legal.registration_date = "2022-01-01"
eco_account.legal.save()
eco_account.responsible.conservation_office = self.get_conservation_office_code()
eco_account.responsible.conservation_file_number = "test"
eco_account.responsible.handler = "handler"
eco_account.responsible.save()
eco_account.after_states.add(self.comp_state)
eco_account.before_states.add(self.comp_state)
eco_account.actions.add(self.comp_action)
eco_account.geometry.geom = self.create_dummy_geometry()
eco_account.geometry.save()
eco_account.deductable_surface = eco_account.get_state_after_surface_sum()
eco_account.save()
return eco_account
def assert_equal_geometries(self, geom1: MultiPolygon, geom2: MultiPolygon):
""" Assert for geometries to be equal
@@ -337,7 +392,10 @@ class BaseViewTestCase(BaseTestCase):
@classmethod
def setUpTestData(cls) -> None:
super().setUpTestData()
cls.login_url = reverse("simple-sso-login")
def setUp(self) -> None:
super().setUp()
self.login_url = reverse("simple-sso-login")
def assert_url_success(self, client: Client, urls: list):
""" Assert for all given urls a direct 200 response
@@ -496,21 +554,6 @@ class BaseWorkflowTestCase(BaseTestCase):
def setUpTestData(cls):
super().setUpTestData()
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])
# Create fresh logged in client and a non-logged in client (anon) for each test
self.client_user = Client()
self.client_user.login(username=self.superuser.username, password=self.superuser_pw)
self.client_anon = Client()
def assert_object_is_deleted(self, obj):
""" Provides a quick check whether an object has been removed from the database or not

2
konova/utils/documents.py
View File

@@ -50,5 +50,5 @@ def remove_document(request: HttpRequest, doc: AbstractDocument):
form = RemoveModalForm(request.POST or None, instance=doc, request=request)
return form.process_request(
request=request,
msg_success=DOCUMENT_REMOVED_TEMPLATE.format(title)
msg_success=DOCUMENT_REMOVED_TEMPLATE.format(title),
)

19
konova/utils/mailer.py
View File

@@ -45,11 +45,12 @@ class Mailer:
auth_password=self.auth_password
)
def send_mail_shared_access_removed(self, obj_identifier, user):
def send_mail_shared_access_removed(self, obj_identifier, obj_title, user):
""" Send a mail if user has no access to the object anymore
Args:
obj_identifier (str): The object identifier
obj_title (str): The object title
Returns:
@@ -57,6 +58,7 @@ class Mailer:
context = {
"user": user,
"obj_identifier": obj_identifier,
"obj_title": obj_title,
"EMAIL_REPLY_TO": EMAIL_REPLY_TO,
}
msg = render_to_string("email/sharing/shared_access_removed.html", context)
@@ -67,7 +69,7 @@ class Mailer:
msg
)
def send_mail_shared_access_given(self, obj_identifier, user):
def send_mail_shared_access_given(self, obj_identifier, obj_title, user):
""" Send a mail if user just got access to the object
Args:
@@ -79,6 +81,7 @@ class Mailer:
context = {
"user": user,
"obj_identifier": obj_identifier,
"obj_title": obj_title,
"EMAIL_REPLY_TO": EMAIL_REPLY_TO,
}
msg = render_to_string("email/sharing/shared_access_given.html", context)
@@ -89,7 +92,7 @@ class Mailer:
msg
)
def send_mail_shared_data_recorded(self, obj_identifier, user):
def send_mail_shared_data_recorded(self, obj_identifier, obj_title, user):
""" Send a mail if the user's shared data has just been unrecorded
Args:
@@ -101,6 +104,7 @@ class Mailer:
context = {
"user": user,
"obj_identifier": obj_identifier,
"obj_title": obj_title,
"EMAIL_REPLY_TO": EMAIL_REPLY_TO,
}
msg = render_to_string("email/recording/shared_data_recorded.html", context)
@@ -111,7 +115,7 @@ class Mailer:
msg
)
def send_mail_shared_data_unrecorded(self, obj_identifier, user):
def send_mail_shared_data_unrecorded(self, obj_identifier, obj_title, user):
""" Send a mail if the user's shared data has just been unrecorded
Args:
@@ -123,6 +127,7 @@ class Mailer:
context = {
"user": user,
"obj_identifier": obj_identifier,
"obj_title": obj_title,
"EMAIL_REPLY_TO": EMAIL_REPLY_TO,
}
msg = render_to_string("email/recording/shared_data_unrecorded.html", context)
@@ -133,7 +138,7 @@ class Mailer:
msg
)
def send_mail_shared_data_deleted(self, obj_identifier, user):
def send_mail_shared_data_deleted(self, obj_identifier, obj_title, user):
""" Send a mail if shared data has just been deleted
Args:
@@ -145,6 +150,7 @@ class Mailer:
context = {
"user": user,
"obj_identifier": obj_identifier,
"obj_title": obj_title,
"EMAIL_REPLY_TO": EMAIL_REPLY_TO,
}
msg = render_to_string("email/deleting/shared_data_deleted.html", context)
@@ -155,7 +161,7 @@ class Mailer:
msg
)
def send_mail_shared_data_checked(self, obj_identifier, user):
def send_mail_shared_data_checked(self, obj_identifier, obj_title, user):
""" Send a mail if shared data just has been checked
Args:
@@ -167,6 +173,7 @@ class Mailer:
context = {
"user": user,
"obj_identifier": obj_identifier,
"obj_title": obj_title,
"EMAIL_REPLY_TO": EMAIL_REPLY_TO,
}
msg = render_to_string("email/checking/shared_data_checked.html", context)

31
konova/utils/message_templates.py
View File

@@ -18,34 +18,61 @@ MISSING_GROUP_PERMISSION = _("You need to be part of another user group.")
CHECKED_RECORDED_RESET = _("Status of Checked and Recorded reseted")
# FILES
FILE_TYPE_UNSUPPORTED = _("Unsupported file type")
FILE_SIZE_TOO_LARGE = _("File too large")
# ECO ACCOUNT
CANCEL_ACC_RECORDED_OR_DEDUCTED = _("Action canceled. Eco account is recorded or deductions exist. Only conservation office member can perform this action.")
# COMPENSATION
COMPENSATION_ADDED_TEMPLATE = _("Compensation {} added")
COMPENSATION_REMOVED_TEMPLATE = _("Compensation {} removed")
COMPENSATION_EDITED_TEMPLATE = _("Compensation {} edited")
ADDED_COMPENSATION_ACTION = _("Added compensation action")
ADDED_COMPENSATION_STATE = _("Added compensation state")
# COMPENSATION STATE
COMPENSATION_STATE_REMOVED = _("State removed")
COMPENSATION_STATE_EDITED = _("State edited")
COMPENSATION_STATE_ADDED = _("State added")
# COMPENSATION ACTION
COMPENSATION_ACTION_ADDED = _("Action added")
COMPENSATION_ACTION_EDITED = _("Action edited")
COMPENSATION_ACTION_REMOVED = _("Action removed")
# DEDUCTIONS
DEDUCTION_ADDED = _("Deduction added")
DEDUCTION_EDITED = _("Deduction edited")
DEDUCTION_REMOVED = _("Deduction removed")
# DEADLINE
DEADLINE_ADDED = _("Deadline added")
DEADLINE_EDITED = _("Deadline edited")
DEADLINE_REMOVED = _("Deadline removed")
# PAYMENTS
PAYMENT_ADDED = _("Payment added")
PAYMENT_EDITED = _("Payment edited")
PAYMENT_REMOVED = _("Payment removed")
# REVOCATIONS
REVOCATION_ADDED = _("Revocation added")
REVOCATION_EDITED = _("Revocation edited")
REVOCATION_REMOVED = _("Revocation removed")
# DOCUMENTS
DOCUMENT_REMOVED_TEMPLATE = _("Document '{}' deleted")
DOCUMENT_ADDED = _("Document added")
DOCUMENT_EDITED = _("Document edited")
# Edited
EDITED_GENERAL_DATA = _("Edited general data")
ADDED_COMPENSATION_STATE = _("Added compensation state")
ADDED_DEADLINE = _("Added deadline")
ADDED_COMPENSATION_ACTION = _("Added compensation action")
# Geometry conflicts
GEOMETRY_CONFLICT_WITH_TEMPLATE = _("Geometry conflict detected with {}")
# INTERVENTION
INTERVENTION_HAS_REVOCATIONS_TEMPLATE = _("This intervention has {} revocations")