diff --git a/src/open_inwoner/conf/base.py b/src/open_inwoner/conf/base.py
index 39859f122..86cb6d46f 100644
--- a/src/open_inwoner/conf/base.py
+++ b/src/open_inwoner/conf/base.py
@@ -911,15 +911,15 @@
],
},
"case_notification": {
- "name": _("Case status update notification"),
+ "name": _("Case update notification"),
"description": _(
- "This email is used to notify people a status update happened to their case"
+ "This email is used to notify people an update happened to their case"
),
- "subject_default": "Status update to your case at {{ site_name }}",
+ "subject_default": "Update to your case at {{ site_name }}",
"body_default": """
Beste
- You are receiving this email because one of your cases received a status update.
+ You are receiving this email because one of your cases received a new status update or document attachment.
@@ -962,7 +962,7 @@
},
{
"name": "case_link",
- "description": _("The link to your case details."),
+ "description": _("The link to the case details."),
},
{
"name": "site_name",
diff --git a/src/open_inwoner/openzaak/admin.py b/src/open_inwoner/openzaak/admin.py
index eb05b9ccd..a32278352 100644
--- a/src/open_inwoner/openzaak/admin.py
+++ b/src/open_inwoner/openzaak/admin.py
@@ -1,6 +1,6 @@
from django.contrib import admin, messages
from django.core.exceptions import ValidationError
-from django.db.models import Count
+from django.db.models import BooleanField, Count, Exists, ExpressionWrapper, Q
from django.forms.models import BaseInlineFormSet
from django.utils.translation import gettext_lazy as _, ngettext
@@ -9,6 +9,7 @@
from .models import (
CatalogusConfig,
OpenZaakConfig,
+ UserCaseInfoObjectNotification,
UserCaseStatusNotification,
ZaakTypeConfig,
ZaakTypeInformatieObjectTypeConfig,
@@ -41,6 +42,25 @@ class CatalogusConfigAdmin(admin.ModelAdmin):
ordering = ("domein", "rsin")
+class HasDocNotifyListFilter(admin.SimpleListFilter):
+ title = _("notify document attachment")
+ parameter_name = "doc_notify"
+
+ def lookups(self, request, model_admin):
+ return [
+ ("yes", _("Yes")),
+ ("no", _("No")),
+ ]
+
+ def queryset(self, request, queryset):
+ v = self.value()
+ if v == "yes":
+ queryset = queryset.filter(has_doc_notify=True)
+ elif v == "no":
+ queryset = queryset.filter(has_doc_notify=False)
+ return queryset
+
+
class CatalogUsedListFilter(admin.SimpleListFilter):
title = _("Catalogus")
parameter_name = "catalogus"
@@ -81,12 +101,14 @@ class ZaakTypeInformatieObjectTypeConfigInline(admin.TabularInline):
fields = [
"omschrijving",
"document_upload_enabled",
- "informatieobjecttype_url",
+ "document_notification_enabled",
+ "informatieobjecttype_uuid",
"zaaktype_uuids",
]
readonly_fields = [
"omschrijving",
"informatieobjecttype_url",
+ "informatieobjecttype_uuid",
"zaaktype_uuids",
]
ordering = ("omschrijving",)
@@ -126,6 +148,7 @@ class ZaakTypeConfigAdmin(admin.ModelAdmin):
"omschrijving",
"catalogus",
"notify_status_changes",
+ "has_doc_notify",
"document_upload_enabled",
"num_infotypes",
]
@@ -135,6 +158,7 @@ class ZaakTypeConfigAdmin(admin.ModelAdmin):
]
list_filter = [
"notify_status_changes",
+ HasDocNotifyListFilter,
CatalogUsedListFilter,
]
search_fields = [
@@ -154,6 +178,19 @@ def has_delete_permission(self, request, obj=None):
def get_queryset(self, request):
qs = super().get_queryset(request)
qs = qs.annotate(num_infotypes=Count("zaaktypeinformatieobjecttypeconfig"))
+ qs = qs.annotate(
+ num_doc_notify=Count(
+ "zaaktypeinformatieobjecttypeconfig",
+ filter=Q(
+ zaaktypeinformatieobjecttypeconfig__document_notification_enabled=True
+ ),
+ )
+ )
+ qs = qs.annotate(
+ has_doc_notify=ExpressionWrapper(
+ Q(num_doc_notify__gt=0), output_field=BooleanField()
+ )
+ )
return qs
def num_infotypes(self, obj=None):
@@ -162,6 +199,17 @@ def num_infotypes(self, obj=None):
else:
return getattr(obj, "num_infotypes", 0)
+ num_infotypes.admin_order_field = "num_infotypes"
+
+ def has_doc_notify(self, obj=None):
+ if not obj or not obj.pk:
+ return False
+ else:
+ return getattr(obj, "has_doc_notify", False)
+
+ has_doc_notify.boolean = True
+ has_doc_notify.admin_order_field = "has_doc_notify"
+
@admin.action(description="Set selected Zaaktypes to notify on status changes")
def mark_as_notify_status_changes(self, request, qs):
count = qs.update(notify_status_changes=True)
@@ -206,7 +254,29 @@ class UserCaseStatusNotificationAdmin(admin.ModelAdmin):
"user",
"case_uuid",
"status_uuid",
- "created",
+ "created_on",
+ ]
+
+ def has_change_permission(self, request, obj=None):
+ return False
+
+
+@admin.register(UserCaseInfoObjectNotification)
+class UserCaseInfoObjectNotificationAdmin(admin.ModelAdmin):
+ raw_id_fields = ["user"]
+ search_fields = [
+ "user__first_name",
+ "user__last_name",
+ "user__email",
+ "user__id",
+ "case_uuid",
+ "zaak_info_object_uuid",
+ ]
+ list_display = [
+ "user",
+ "case_uuid",
+ "zaak_info_object_uuid",
+ "created_on",
]
def has_change_permission(self, request, obj=None):
diff --git a/src/open_inwoner/openzaak/cases.py b/src/open_inwoner/openzaak/cases.py
index 5043d38d8..00497ab14 100644
--- a/src/open_inwoner/openzaak/cases.py
+++ b/src/open_inwoner/openzaak/cases.py
@@ -77,6 +77,29 @@ def fetch_single_case(case_uuid: str) -> Optional[Zaak]:
return case
+@cache_result(
+ "single_case_information_object:{url}", timeout=settings.CACHE_ZGW_ZAKEN_TIMEOUT
+)
+def fetch_single_case_information_object(url: str) -> Optional[ZaakInformatieObject]:
+ client = build_client("zaak")
+
+ if client is None:
+ return
+
+ try:
+ response = client.retrieve("zaakinformatieobject", url=url)
+ except RequestException as e:
+ logger.exception("exception while making request", exc_info=e)
+ return
+ except ClientError as e:
+ logger.exception("exception while making request", exc_info=e)
+ return
+
+ case = factory(ZaakInformatieObject, response)
+
+ return case
+
+
def fetch_case_by_url_no_cache(case_url: str) -> Optional[Zaak]:
client = build_client("zaak")
try:
diff --git a/src/open_inwoner/openzaak/documents.py b/src/open_inwoner/openzaak/documents.py
index e03915cc0..00dc7bee8 100644
--- a/src/open_inwoner/openzaak/documents.py
+++ b/src/open_inwoner/openzaak/documents.py
@@ -19,7 +19,6 @@
ZaakTypeInformatieObjectTypeConfig,
)
-from .api_models import Zaak
from .utils import cache as cache_result
logger = logging.getLogger(__name__)
@@ -117,6 +116,7 @@ def upload_document(
"inhoud": base64.b64encode(file.read()).decode("utf-8"),
"bestandsomvang": file.size,
"bestandsnaam": file.name,
+ # "status": "definitief",
"taal": "nld",
"informatieobjecttype": ZaakTypeInformatieObjectTypeConfig.objects.get(
id=user_choice
diff --git a/src/open_inwoner/openzaak/managers.py b/src/open_inwoner/openzaak/managers.py
index 430da69bb..c6ecbf5fe 100644
--- a/src/open_inwoner/openzaak/managers.py
+++ b/src/open_inwoner/openzaak/managers.py
@@ -4,24 +4,16 @@
from django.db import IntegrityError, models, transaction
from open_inwoner.accounts.models import User
+from open_inwoner.openzaak.api_models import Zaak, ZaakType
if TYPE_CHECKING:
- from open_inwoner.openzaak.models import UserCaseStatusNotification
+ from open_inwoner.openzaak.models import (
+ UserCaseInfoObjectNotification,
+ UserCaseStatusNotification,
+ )
-class UserCaseStatusNotificationQueryset(models.QuerySet):
- def get_user_case_notifications(self, user, case_uuid):
- return self.filter(user=user, case_uuid=case_uuid)
-
- def has_notification(self, user, case_uuid, status_uuid):
- return self.filter(
- user=user, case_uuid=case_uuid, status_uuid=status_uuid
- ).exists()
-
-
-class UserCaseStatusNotificationManager(
- models.Manager.from_queryset(UserCaseStatusNotificationQueryset)
-):
+class UserCaseStatusNotificationManager(models.Manager):
def record_if_unique_notification(
self,
user: User,
@@ -44,12 +36,60 @@ def record_if_unique_notification(
return None
+class UserCaseInfoObjectNotificationManager(models.Manager):
+ def record_if_unique_notification(
+ self,
+ user: User,
+ case_uuid: UUID,
+ zaak_info_object_uuid: UUID,
+ ) -> Optional["UserCaseInfoObjectNotification"]:
+ """
+ assume this is the first delivery but depend on the unique constraint
+ """
+ kwargs = {
+ "user": user,
+ "case_uuid": case_uuid,
+ "zaak_info_object_uuid": zaak_info_object_uuid,
+ }
+ try:
+ with transaction.atomic():
+ note = self.create(**kwargs)
+ return note
+ except IntegrityError:
+ return None
+
+
class ZaakTypeInformatieObjectTypeConfigQueryset(models.QuerySet):
- def get_visible_ztiot_configs_for_case(self, case):
+ def filter_catalogus(self, case_type: ZaakType):
+ if case_type.catalogus:
+ # support both url and resolved dataclass
+ catalogus_url = (
+ case_type.catalogus
+ if isinstance(case_type.catalogus, str)
+ else case_type.catalogus.url
+ )
+ return self.filter(
+ zaaktype_config__catalogus__url=catalogus_url,
+ )
+ else:
+ return self.filter(
+ zaaktype_config__catalogus__isnull=True,
+ )
+
+ def filter_case_type(self, case_type: ZaakType):
+ return self.filter_catalogus(case_type).filter(
+ zaaktype_uuids__contains=[case_type.uuid],
+ zaaktype_config__identificatie=case_type.identificatie,
+ )
+
+ def get_visible_ztiot_configs_for_case(self, case: Zaak):
"""
Returns all ZaakTypeInformatieObjectTypeConfig instances which allow
documents upload and are based on a specific case and case type.
"""
+ # TODO rename to 'filter_visible_for_case'
+ # TODO change signature to accept case_type/ZaakType
+ # TODO refactor to use self.filter_case_type(case_type)
if not case:
return self.none()
@@ -59,8 +99,36 @@ def get_visible_ztiot_configs_for_case(self, case):
document_upload_enabled=True,
)
+ def get_for_case_and_info_type(
+ self, case_type: ZaakType, info_object_type_url: str
+ ):
+ return self.filter_case_type(case_type).get(
+ informatieobjecttype_url=info_object_type_url,
+ )
+
class ZaakTypeConfigQueryset(models.QuerySet):
+ def filter_catalogus(self, case_type: ZaakType):
+ if case_type.catalogus:
+ # support both url and resolved dataclass
+ catalogus_url = (
+ case_type.catalogus
+ if isinstance(case_type.catalogus, str)
+ else case_type.catalogus.url
+ )
+ return self.filter(
+ catalogus__url=catalogus_url,
+ )
+ else:
+ return self.filter(
+ catalogus__isnull=True,
+ )
+
+ def filter_case_type(self, case_type: ZaakType):
+ return self.filter_catalogus(case_type).filter(
+ identificatie=case_type.identificatie,
+ )
+
def get_visible_zt_configs_for_case_type_identification(
self, case_type_identification
):
@@ -71,6 +139,10 @@ def get_visible_zt_configs_for_case_type_identification(
if not case_type_identification:
return self.none()
+ # TODO rename to 'filter_visible_for_case_type'
+ # TODO change signature to accept case_type
+ # TODO refactor to use self.filter_case_type(case_type)
+
return self.filter(
identificatie=case_type_identification,
document_upload_enabled=True,
diff --git a/src/open_inwoner/openzaak/migrations/0011_auto_20230207_1030.py b/src/open_inwoner/openzaak/migrations/0011_auto_20230207_1030.py
new file mode 100644
index 000000000..c73232f35
--- /dev/null
+++ b/src/open_inwoner/openzaak/migrations/0011_auto_20230207_1030.py
@@ -0,0 +1,63 @@
+# Generated by Django 3.2.15 on 2023-02-07 09:30
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+ ("openzaak", "0010_openzaakconfig_reformat_esuite_zaak_identificatie"),
+ ]
+
+ operations = [
+ migrations.AlterModelOptions(
+ name="usercasestatusnotification",
+ options={"verbose_name": "Open Zaak status notification record"},
+ ),
+ migrations.CreateModel(
+ name="UserCaseInfoObjectNotification",
+ fields=[
+ (
+ "id",
+ models.AutoField(
+ auto_created=True,
+ primary_key=True,
+ serialize=False,
+ verbose_name="ID",
+ ),
+ ),
+ ("case_uuid", models.UUIDField(verbose_name="Zaak UUID")),
+ (
+ "zaak_info_object_uuid",
+ models.UUIDField(verbose_name="InformatieObject UUID"),
+ ),
+ (
+ "created",
+ models.DateTimeField(
+ default=django.utils.timezone.now, verbose_name="Created"
+ ),
+ ),
+ (
+ "user",
+ models.ForeignKey(
+ on_delete=django.db.models.deletion.CASCADE,
+ to=settings.AUTH_USER_MODEL,
+ ),
+ ),
+ ],
+ options={
+ "verbose_name": "Open Zaak info object notification record",
+ },
+ ),
+ migrations.AddConstraint(
+ model_name="usercaseinfoobjectnotification",
+ constraint=models.UniqueConstraint(
+ fields=("user", "case_uuid", "zaak_info_object_uuid"),
+ name="unique_user_case_info_object",
+ ),
+ ),
+ ]
diff --git a/src/open_inwoner/openzaak/migrations/0012_zaaktypeinformatieobjecttypeconfig_document_notification_enabled.py b/src/open_inwoner/openzaak/migrations/0012_zaaktypeinformatieobjecttypeconfig_document_notification_enabled.py
new file mode 100644
index 000000000..9aada59ba
--- /dev/null
+++ b/src/open_inwoner/openzaak/migrations/0012_zaaktypeinformatieobjecttypeconfig_document_notification_enabled.py
@@ -0,0 +1,22 @@
+# Generated by Django 3.2.15 on 2023-02-08 09:34
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ("openzaak", "0011_auto_20230207_1030"),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name="zaaktypeinformatieobjecttypeconfig",
+ name="document_notification_enabled",
+ field=models.BooleanField(
+ default=False,
+ help_text="When enabled the user will receive a notification when a visible document is added to the case",
+ verbose_name="Enable document notifications",
+ ),
+ ),
+ ]
diff --git a/src/open_inwoner/openzaak/migrations/0013_auto_20230209_1053.py b/src/open_inwoner/openzaak/migrations/0013_auto_20230209_1053.py
new file mode 100644
index 000000000..d7d449b4f
--- /dev/null
+++ b/src/open_inwoner/openzaak/migrations/0013_auto_20230209_1053.py
@@ -0,0 +1,26 @@
+# Generated by Django 3.2.15 on 2023-02-09 09:53
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ (
+ "openzaak",
+ "0012_zaaktypeinformatieobjecttypeconfig_document_notification_enabled",
+ ),
+ ]
+
+ operations = [
+ migrations.RenameField(
+ model_name="usercaseinfoobjectnotification",
+ old_name="created",
+ new_name="created_on",
+ ),
+ migrations.RenameField(
+ model_name="usercasestatusnotification",
+ old_name="created",
+ new_name="created_on",
+ ),
+ ]
diff --git a/src/open_inwoner/openzaak/migrations/0014_auto_20230209_1055.py b/src/open_inwoner/openzaak/migrations/0014_auto_20230209_1055.py
new file mode 100644
index 000000000..978a60b45
--- /dev/null
+++ b/src/open_inwoner/openzaak/migrations/0014_auto_20230209_1055.py
@@ -0,0 +1,28 @@
+# Generated by Django 3.2.15 on 2023-02-09 09:55
+
+import django.utils.timezone
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ("openzaak", "0013_auto_20230209_1053"),
+ ]
+
+ operations = [
+ migrations.AlterField(
+ model_name="usercaseinfoobjectnotification",
+ name="created_on",
+ field=models.DateTimeField(
+ default=django.utils.timezone.now, verbose_name="Created on"
+ ),
+ ),
+ migrations.AlterField(
+ model_name="usercasestatusnotification",
+ name="created_on",
+ field=models.DateTimeField(
+ default=django.utils.timezone.now, verbose_name="Created on"
+ ),
+ ),
+ ]
diff --git a/src/open_inwoner/openzaak/models.py b/src/open_inwoner/openzaak/models.py
index a3f8e80af..274ab5e3f 100644
--- a/src/open_inwoner/openzaak/models.py
+++ b/src/open_inwoner/openzaak/models.py
@@ -4,11 +4,13 @@
from django.utils.translation import gettext_lazy as _
from django_better_admin_arrayfield.models.fields import ArrayField
+from furl import furl
from solo.models import SingletonModel
from zgw_consumers.api_models.constants import VertrouwelijkheidsAanduidingen
from zgw_consumers.constants import APITypes
from open_inwoner.openzaak.managers import (
+ UserCaseInfoObjectNotificationManager,
UserCaseStatusNotificationManager,
ZaakTypeConfigQueryset,
ZaakTypeInformatieObjectTypeConfigQueryset,
@@ -225,7 +227,13 @@ class ZaakTypeInformatieObjectTypeConfig(models.Model):
verbose_name=_("Enable document upload"),
default=False,
)
-
+ document_notification_enabled = models.BooleanField(
+ verbose_name=_("Enable document notifications"),
+ default=False,
+ help_text=_(
+ "When enabled the user will receive a notification when a visible document is added to the case"
+ ),
+ )
objects = ZaakTypeInformatieObjectTypeConfigQueryset.as_manager()
class Meta:
@@ -238,6 +246,20 @@ class Meta:
)
]
+ def informatieobjecttype_uuid(self):
+ if self.informatieobjecttype_url:
+ segments = furl(self.informatieobjecttype_url).path.segments
+ # grab uuid as last bit of url,
+ # but handle trailing slash or weird urls from factories
+ while segments:
+ s = segments.pop()
+ if s:
+ return s
+ return self.informatieobjecttype_url
+ return ""
+
+ informatieobjecttype_uuid.short_description = _("Information object UUID")
+
def __str__(self):
return self.omschrijving
@@ -253,12 +275,14 @@ class UserCaseStatusNotification(models.Model):
status_uuid = models.UUIDField(
verbose_name=_("Status UUID"),
)
- created = models.DateTimeField(verbose_name=_("Created"), default=timezone.now)
+ created_on = models.DateTimeField(
+ verbose_name=_("Created on"), default=timezone.now
+ )
objects = UserCaseStatusNotificationManager()
class Meta:
- verbose_name = _("Open Zaak notification user inform record")
+ verbose_name = _("Open Zaak status notification record")
constraints = [
UniqueConstraint(
@@ -266,3 +290,31 @@ class Meta:
fields=["user", "case_uuid", "status_uuid"],
)
]
+
+
+class UserCaseInfoObjectNotification(models.Model):
+ user = models.ForeignKey(
+ "accounts.User",
+ on_delete=models.CASCADE,
+ )
+ case_uuid = models.UUIDField(
+ verbose_name=_("Zaak UUID"),
+ )
+ zaak_info_object_uuid = models.UUIDField(
+ verbose_name=_("InformatieObject UUID"),
+ )
+ created_on = models.DateTimeField(
+ verbose_name=_("Created on"), default=timezone.now
+ )
+
+ objects = UserCaseInfoObjectNotificationManager()
+
+ class Meta:
+ verbose_name = _("Open Zaak info object notification record")
+
+ constraints = [
+ UniqueConstraint(
+ name="unique_user_case_info_object",
+ fields=["user", "case_uuid", "zaak_info_object_uuid"],
+ )
+ ]
diff --git a/src/open_inwoner/openzaak/notifications.py b/src/open_inwoner/openzaak/notifications.py
index d1db5e57b..dca729cd8 100644
--- a/src/open_inwoner/openzaak/notifications.py
+++ b/src/open_inwoner/openzaak/notifications.py
@@ -7,10 +7,17 @@
from zgw_consumers.api_models.constants import RolOmschrijving, RolTypes
from open_inwoner.accounts.models import User
-from open_inwoner.openzaak.api_models import Notification, Rol, Status, Zaak
+from open_inwoner.openzaak.api_models import (
+ Notification,
+ Rol,
+ Status,
+ Zaak,
+ ZaakInformatieObject,
+)
from open_inwoner.openzaak.cases import (
fetch_case_by_url_no_cache,
fetch_case_roles,
+ fetch_single_case_information_object,
fetch_specific_status,
fetch_status_history_no_cache,
)
@@ -18,10 +25,17 @@
fetch_single_case_type,
fetch_single_status_type,
)
-from open_inwoner.openzaak.models import OpenZaakConfig, UserCaseStatusNotification
+from open_inwoner.openzaak.documents import fetch_single_information_object_url
+from open_inwoner.openzaak.models import (
+ OpenZaakConfig,
+ UserCaseInfoObjectNotification,
+ UserCaseStatusNotification,
+)
from open_inwoner.openzaak.utils import (
format_zaak_identificatie,
get_zaak_type_config,
+ get_zaak_type_info_object_type_config,
+ is_info_object_visible,
is_zaak_visible,
)
from open_inwoner.utils.logentry import system_action as log_system_action
@@ -30,6 +44,20 @@
logger = logging.getLogger(__name__)
+def wrap_join(iter, glue="") -> str:
+ parts = list(sorted(f"'{v}'" for v in iter))
+ if not parts:
+ return ""
+ elif len(parts) == 1:
+ return parts[0]
+ elif glue:
+ end = parts.pop()
+ lead = ", ".join(parts)
+ return f"{lead} {glue} {end}"
+ else:
+ return ", ".join(parts)
+
+
def handle_zaken_notification(notification: Notification):
if notification.kanaal != "zaken":
raise Exception(
@@ -39,23 +67,22 @@ def handle_zaken_notification(notification: Notification):
# on the 'zaken' channel the hoofd_object is always the zaak
case_url = notification.hoofd_object
- oz_config = OpenZaakConfig.get_solo()
+ # we're only interested in some updates
+ resources = ("status", "zaakinformatieobject")
+ r = notification.resource # short alias for logging
- # were only interested in status updates
- if notification.resource != "status":
+ if notification.resource not in resources:
log_system_action(
- f"ignored notification: resource is not 'status' but '{notification.resource}' for case {case_url}",
+ f"ignored {r} notification: resource is not {wrap_join(resources, 'or')} but '{notification.resource}' for case {case_url}",
log_level=logging.INFO,
)
return
- status_url = notification.resource_url
-
# check if we have users that need to be informed about this case
roles = fetch_case_roles(case_url)
if not roles:
log_system_action(
- f"ignored notification: cannot retrieve rollen for case {case_url}",
+ f"ignored {r} notification: cannot retrieve rollen for case {case_url}",
log_level=logging.ERROR,
)
return
@@ -63,24 +90,163 @@ def handle_zaken_notification(notification: Notification):
inform_users = get_emailable_initiator_users_from_roles(roles)
if not inform_users:
log_system_action(
- f"ignored notification: no users with bsn and valid email as (mede)initiators in case {case_url}",
+ f"ignored {r} notification: no users with bsn and valid email as (mede)initiators in case {case_url}",
+ log_level=logging.INFO,
+ )
+ return
+
+ # check if this case is visible
+ case = fetch_case_by_url_no_cache(case_url)
+ if not case:
+ log_system_action(
+ f"ignored {r} notification: cannot retrieve case {case_url}",
+ log_level=logging.ERROR,
+ )
+ return
+
+ case_type = fetch_single_case_type(case.zaaktype)
+ if not case_type:
+ log_system_action(
+ f"ignored {r} notification: cannot retrieve case_type {case.zaaktype} for case {case_url}",
+ log_level=logging.ERROR,
+ )
+ return
+
+ case.zaaktype = case_type
+
+ if not is_zaak_visible(case):
+ log_system_action(
+ f"ignored {r} notification: case not visible after applying website visibility filter for case {case_url}",
+ log_level=logging.INFO,
+ )
+ return
+
+ if notification.resource == "status":
+ _handle_status_notification(notification, case, inform_users)
+ elif notification.resource == "zaakinformatieobject":
+ _handle_zaakinformatieobject_notification(notification, case, inform_users)
+ else:
+ raise NotImplementedError("programmer error in earlier resource filter")
+
+
+def _handle_zaakinformatieobject_notification(
+ notification: Notification, case: Zaak, inform_users
+):
+ oz_config = OpenZaakConfig.get_solo()
+ r = notification.resource # short alias for logging
+
+ """
+ {
+ "kanaal": "zaken",
+ "hoofdObject": "https://test.openzaak.nl/zaken/api/v1/zaken/af715571-a542-4b68-9a46-3821b9589045",
+ "resource": "zaakinformatieobject",
+ "resourceUrl": "https://test.openzaak.nl/zaken/api/v1/zaakinformatieobjecten/348d0669-0145-48de-859f-29dafa8a885a",
+ "actie": "create",
+ "aanmaakdatum": "2023-02-06T09:33:17.402799Z",
+ "kenmerken": {
+ "bronorganisatie": "100000009",
+ "zaaktype": "https://test.openzaak.nl/catalogi/api/v1/zaaktypen/2c1feba6-3163-4e15-9afa-fa4f01dcb4f9",
+ "vertrouwelijkheidaanduiding": "openbaar"
+ }}
+ """
+
+ # check if this is a zaakinformatieobject we want to inform on
+ ziobj_url = notification.resource_url
+
+ ziobj = fetch_single_case_information_object(ziobj_url)
+
+ if not ziobj:
+ log_system_action(
+ f"ignored {r} notification: cannot retrieve zaakinformatieobject {ziobj_url} for case {case.url}",
+ log_level=logging.ERROR,
+ )
+ return
+
+ info_object = fetch_single_information_object_url(ziobj.informatieobject)
+ if not info_object:
+ log_system_action(
+ f"ignored {r} notification: cannot retrieve informatieobject {ziobj.informatieobject} for case {case.url}",
+ log_level=logging.ERROR,
+ )
+ return
+
+ ziobj.informatieobject = info_object
+
+ if not is_info_object_visible(info_object, oz_config.document_max_confidentiality):
+ log_system_action(
+ f"ignored {r} notification: informatieobject not visible after applying website visibility filter for case {case.url}",
log_level=logging.INFO,
)
return
+ # NOTE for documents we don't check the statustype.informeren
+ ztiotc = get_zaak_type_info_object_type_config(
+ case.zaaktype, info_object.informatieobjecttype
+ )
+ if not ztiotc:
+ log_system_action(
+ f"ignored {r} notification: cannot retrieve info_type configuration {info_object.informatieobjecttype} and case {case.url}",
+ log_level=logging.INFO,
+ )
+ return
+ elif not ztiotc.document_notification_enabled:
+ log_system_action(
+ f"ignored {r} notification: info_type configuration '{ztiotc.omschrijving}' {info_object.informatieobjecttype} found but 'document_notification_enabled' is False for case {case.url}",
+ log_level=logging.INFO,
+ )
+ return
+
+ # reaching here means we're going to inform users
+ log_system_action(
+ f"accepted {r} notification: attempt informing users {wrap_join(inform_users)} for case {case.url}",
+ log_level=logging.INFO,
+ )
+ for user in inform_users:
+ # TODO run in try/except so it can't bail?
+ handle_zaakinformatieobject_update(user, case, ziobj)
+
+
+def handle_zaakinformatieobject_update(
+ user: User, case: Zaak, zaak_info_object: ZaakInformatieObject
+):
+ note = UserCaseInfoObjectNotification.objects.record_if_unique_notification(
+ user,
+ case.uuid,
+ zaak_info_object.uuid,
+ )
+ if not note:
+ log_system_action(
+ f"ignored duplicate zaakinformatieobject notification delivery for user '{user}' zaakinformatieobject {zaak_info_object.url} case {case.url}",
+ log_level=logging.INFO,
+ )
+ return
+
+ send_case_update_email(user, case)
+
+ log_system_action(
+ f"send zaakinformatieobject notification email for user '{user}' zaakinformatieobject {zaak_info_object.url} case {case.url}",
+ log_level=logging.INFO,
+ )
+
+
+def _handle_status_notification(notification: Notification, case: Zaak, inform_users):
+ oz_config = OpenZaakConfig.get_solo()
+ r = notification.resource # short alias for logging
+
# check if this is a status we want to inform on
+ status_url = notification.resource_url
- status_history = fetch_status_history_no_cache(case_url)
+ status_history = fetch_status_history_no_cache(case.url)
if not status_history:
log_system_action(
- f"ignored notification: cannot retrieve status_history for case {case_url}",
+ f"ignored {r} notification: cannot retrieve status_history for case {case.url}",
log_level=logging.ERROR,
)
return
if len(status_history) == 1:
log_system_action(
- f"ignored notification: skip initial status notification for case {case_url}",
+ f"ignored {r} notification: skip initial status notification for case {case.url}",
log_level=logging.INFO,
)
return
@@ -94,7 +260,7 @@ def handle_zaken_notification(notification: Notification):
if not status:
log_system_action(
- f"ignored notification: cannot retrieve status {status_url} for case {case_url}",
+ f"ignored {r} notification: cannot retrieve status {status_url} for case {case.url}",
log_level=logging.ERROR,
)
return
@@ -102,7 +268,7 @@ def handle_zaken_notification(notification: Notification):
status_type = fetch_single_status_type(status.statustype)
if not status_type:
log_system_action(
- f"ignored notification: cannot retrieve status_type {status.statustype} for case {case_url}",
+ f"ignored {r} notification: cannot retrieve status_type {status.statustype} for case {case.url}",
log_level=logging.ERROR,
)
return
@@ -110,58 +276,32 @@ def handle_zaken_notification(notification: Notification):
if not oz_config.skip_notification_statustype_informeren:
if not status_type.informeren:
log_system_action(
- f"ignored notification: status_type.informeren is false for status {status.url} and case {case_url}",
+ f"ignored {r} notification: status_type.informeren is false for status {status.url} and case {case.url}",
log_level=logging.INFO,
)
return
status.statustype = status_type
- # check if this case is visible
- case = fetch_case_by_url_no_cache(case_url)
- if not case:
- log_system_action(
- f"ignored notification: cannot retrieve case {case_url}",
- log_level=logging.ERROR,
- )
- return
-
- case_type = fetch_single_case_type(case.zaaktype)
- if not case_type:
- log_system_action(
- f"ignored notification: cannot retrieve case_type {case.zaaktype} for case {case_url}",
- log_level=logging.ERROR,
- )
- return
-
- case.zaaktype = case_type
-
# check the ZaakTypeConfig
if oz_config.skip_notification_statustype_informeren:
- ztc = get_zaak_type_config(case_type)
+ ztc = get_zaak_type_config(case.zaaktype)
if not ztc:
log_system_action(
- f"ignored notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{case.zaaktype.identificatie}' for case {case_url}",
+ f"ignored {r} notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{case.zaaktype.identificatie}' for case {case.url}",
log_level=logging.INFO,
)
return
elif not ztc.notify_status_changes:
log_system_action(
- f"ignored notification: case_type configuration '{case.zaaktype.identificatie}' found but 'notify_status_changes' is False for case {case_url}",
+ f"ignored {r} notification: case_type configuration '{case.zaaktype.identificatie}' found but 'notify_status_changes' is False for case {case.url}",
log_level=logging.INFO,
)
return
- if not is_zaak_visible(case):
- log_system_action(
- f"ignored notification: case not visible after applying website visibility filter for case {case_url}",
- log_level=logging.INFO,
- )
- return
-
# reaching here means we're going to inform users
log_system_action(
- f"accepted notification: attempt informing users {', '.join(sorted(map(str, inform_users)))} for case {case_url}",
+ f"accepted {r} notification: attempt informing users {wrap_join(inform_users)} for case {case.url}",
log_level=logging.INFO,
)
for user in inform_users:
@@ -171,24 +311,26 @@ def handle_zaken_notification(notification: Notification):
def handle_status_update(user: User, case: Zaak, status: Status):
note = UserCaseStatusNotification.objects.record_if_unique_notification(
- user, case.uuid, status.uuid
+ user,
+ case.uuid,
+ status.uuid,
)
if not note:
log_system_action(
- f"ignored duplicate notification delivery for user '{user}' status {status.url} case {case.url}",
+ f"ignored duplicate status notification delivery for user '{user}' status {status.url} case {case.url}",
log_level=logging.INFO,
)
return
- send_status_update_email(user, case, status)
+ send_case_update_email(user, case)
log_system_action(
- f"send notification email for user '{user}' status {status.url} case {case.url}",
+ f"send status notification email for user '{user}' status {status.url} case {case.url}",
log_level=logging.INFO,
)
-def send_status_update_email(user: User, case: Zaak, status: Status):
+def send_case_update_email(user: User, case: Zaak):
"""
send the actual mail
"""
diff --git a/src/open_inwoner/openzaak/tests/factories.py b/src/open_inwoner/openzaak/tests/factories.py
index 7faa869dc..e575b4767 100644
--- a/src/open_inwoner/openzaak/tests/factories.py
+++ b/src/open_inwoner/openzaak/tests/factories.py
@@ -3,6 +3,7 @@
from simple_certmanager.constants import CertificateTypes
from simple_certmanager.models import Certificate
from zgw_consumers.api_models.base import factory as zwg_factory
+from zgw_consumers.api_models.catalogi import InformatieObjectType
from zgw_consumers.api_models.constants import RolOmschrijving
from zgw_consumers.models import Service
from zgw_consumers.test import generate_oas_component
@@ -77,6 +78,34 @@ class ZaakTypeInformatieObjectTypeConfigFactory(factory.django.DjangoModelFactor
class Meta:
model = ZaakTypeInformatieObjectTypeConfig
+ @staticmethod
+ def from_case_type_info_object_dicts(
+ zaak_type: dict,
+ info_object: dict,
+ document_upload_enabled=False,
+ document_notification_enabled=False,
+ **extra_kwargs,
+ ):
+ kwargs = dict(
+ zaaktype_config__identificatie=zaak_type["identificatie"],
+ zaaktype_config__omschrijving=zaak_type["omschrijving"],
+ informatieobjecttype_url=info_object["informatieobjecttype"],
+ document_upload_enabled=document_upload_enabled,
+ document_notification_enabled=document_notification_enabled,
+ zaaktype_uuids=[zaak_type["uuid"]],
+ )
+ if zaak_type["catalogus"]:
+ kwargs.update(
+ zaaktype_config__catalogus__url=zaak_type["catalogus"],
+ )
+ else:
+ kwargs.update(
+ zaaktype_config__catalogus=None,
+ )
+ if extra_kwargs:
+ kwargs.update(extra_kwargs)
+ return ZaakTypeInformatieObjectTypeConfigFactory(**kwargs)
+
class NotificationFactory(factory.Factory):
kanaal = "zaken"
diff --git a/src/open_inwoner/openzaak/tests/test_models.py b/src/open_inwoner/openzaak/tests/test_models.py
new file mode 100644
index 000000000..7365e4cd1
--- /dev/null
+++ b/src/open_inwoner/openzaak/tests/test_models.py
@@ -0,0 +1,152 @@
+from django.test import TestCase
+
+from zgw_consumers.api_models.base import factory
+from zgw_consumers.test import generate_oas_component
+
+from open_inwoner.openzaak.api_models import ZaakType
+from open_inwoner.openzaak.models import (
+ ZaakTypeConfig,
+ ZaakTypeInformatieObjectTypeConfig,
+)
+from open_inwoner.openzaak.tests.factories import (
+ CatalogusConfigFactory,
+ ZaakTypeConfigFactory,
+ ZaakTypeInformatieObjectTypeConfigFactory,
+)
+from open_inwoner.openzaak.tests.shared import CATALOGI_ROOT
+
+
+class ZaakTypeConfigModelTestCase(TestCase):
+ def test_queryset_filter_case_type_with_catalog(self):
+ catalog = CatalogusConfigFactory(
+ url=f"{CATALOGI_ROOT}catalogussen/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ )
+ zaak_type_config = ZaakTypeConfigFactory(
+ catalogus=catalog,
+ identificatie="AAAA",
+ )
+ case_type = factory(
+ ZaakType,
+ generate_oas_component(
+ "ztc",
+ "schemas/ZaakType",
+ uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ url=f"{CATALOGI_ROOT}zaaktype/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ catalogus=catalog.url,
+ identificatie="AAAA",
+ ),
+ )
+
+ actual = list(ZaakTypeConfig.objects.filter_case_type(case_type))
+ self.assertEqual(actual, [zaak_type_config])
+
+ def test_queryset_filter_case_type_without_catalog(self):
+ zaak_type_config = ZaakTypeConfigFactory(
+ catalogus=None,
+ identificatie="AAAA",
+ )
+ case_type = factory(
+ ZaakType,
+ generate_oas_component(
+ "ztc",
+ "schemas/ZaakType",
+ uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ url=f"{CATALOGI_ROOT}zaaktype/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ catalogus=None,
+ identificatie="AAAA",
+ ),
+ )
+
+ actual = list(ZaakTypeConfig.objects.filter_case_type(case_type))
+ self.assertEqual(actual, [zaak_type_config])
+
+
+class ZaakTypeInformatieObjectTypeConfigFactoryModelTestCase(TestCase):
+ def test_queryset_filter_case_type_with_catalog(self):
+ catalog = CatalogusConfigFactory(
+ url=f"{CATALOGI_ROOT}catalogussen/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ )
+ zaak_type_config = ZaakTypeConfigFactory(
+ catalogus=catalog,
+ identificatie="AAAA",
+ )
+ a1 = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config=zaak_type_config,
+ informatieobjecttype_url="https://example.com/v1/infoobject/a1",
+ zaaktype_uuids=["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"],
+ )
+ a2 = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config=zaak_type_config,
+ informatieobjecttype_url="https://example.com/v1/infoobject/a2",
+ zaaktype_uuids=[],
+ )
+ b = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config=zaak_type_config,
+ informatieobjecttype_url="https://example.com/v1/infoobject/bbb",
+ zaaktype_uuids=["aaaaaaaa-aaaa-bbbb-aaaa-aaaaaaaaaaaa"],
+ )
+ c = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config__catalogus=catalog,
+ zaaktype_config__identificatie="CCC",
+ informatieobjecttype_url="https://example.com/v1/infoobject/bbb",
+ zaaktype_uuids=["aaaaaaaa-aaaa-bbbb-aaaa-aaaaaaaaaaaa"],
+ )
+ case_type = factory(
+ ZaakType,
+ generate_oas_component(
+ "ztc",
+ "schemas/ZaakType",
+ uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ url=f"{CATALOGI_ROOT}zaaktype/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ catalogus=catalog.url,
+ identificatie="AAAA",
+ ),
+ )
+
+ actual = list(
+ ZaakTypeInformatieObjectTypeConfig.objects.filter_case_type(case_type)
+ )
+ self.assertEqual(actual, [a1])
+
+ def test_queryset_filter_case_type_without_catalog(self):
+ zaak_type_config = ZaakTypeConfigFactory(
+ catalogus=None,
+ identificatie="AAAA",
+ )
+ a1 = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config=zaak_type_config,
+ informatieobjecttype_url="https://example.com/v1/infoobject/a1",
+ zaaktype_uuids=["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"],
+ )
+ a2 = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config=zaak_type_config,
+ informatieobjecttype_url="https://example.com/v1/infoobject/a2",
+ zaaktype_uuids=[],
+ )
+ b = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config=zaak_type_config,
+ informatieobjecttype_url="https://example.com/v1/infoobject/bbb",
+ zaaktype_uuids=["aaaaaaaa-aaaa-bbbb-aaaa-aaaaaaaaaaaa"],
+ )
+ c = ZaakTypeInformatieObjectTypeConfigFactory(
+ zaaktype_config__catalogus=None,
+ zaaktype_config__identificatie="CCC",
+ informatieobjecttype_url="https://example.com/v1/infoobject/bbb",
+ zaaktype_uuids=["aaaaaaaa-aaaa-bbbb-aaaa-aaaaaaaaaaaa"],
+ )
+ case_type = factory(
+ ZaakType,
+ generate_oas_component(
+ "ztc",
+ "schemas/ZaakType",
+ uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ url=f"{CATALOGI_ROOT}zaaktype/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ catalogus=None,
+ identificatie="AAAA",
+ ),
+ )
+
+ actual = list(
+ ZaakTypeInformatieObjectTypeConfig.objects.filter_case_type(case_type)
+ )
+ self.assertEqual(actual, [a1])
diff --git a/src/open_inwoner/openzaak/tests/test_notification_data.py b/src/open_inwoner/openzaak/tests/test_notification_data.py
new file mode 100644
index 000000000..0afa8e645
--- /dev/null
+++ b/src/open_inwoner/openzaak/tests/test_notification_data.py
@@ -0,0 +1,214 @@
+from typing import List, Optional
+
+from zgw_consumers.api_models.constants import (
+ RolOmschrijving,
+ RolTypes,
+ VertrouwelijkheidsAanduidingen,
+)
+from zgw_consumers.constants import APITypes
+from zgw_consumers.test import generate_oas_component, mock_service_oas_get
+
+from open_inwoner.accounts.tests.factories import DigidUserFactory
+from open_inwoner.openzaak.tests.factories import (
+ NotificationFactory,
+ ServiceFactory,
+ ZaakTypeConfigFactory,
+ ZaakTypeInformatieObjectTypeConfigFactory,
+)
+from open_inwoner.utils.test import paginated_response
+
+from ..models import OpenZaakConfig
+from .shared import CATALOGI_ROOT, DOCUMENTEN_ROOT, ZAKEN_ROOT
+
+
+class MockAPIData:
+ """
+ initializes isolated and valid data for a complete mock-request API flow,
+ allows to manipulate data per test to break it,
+ and still get dry/readable access to the data for assertions
+
+ # usage:
+
+ data = MockAPIData()
+
+ # change some resources
+ data.zaak["some_field"] = "a different value"
+
+ # install to your @requests_mock.Mocker()
+ data.install_mocks(m)
+
+ # install but return 404 for some resource
+ data.install_mocks(m, res404=["zaak"])
+
+ # also:
+
+ @classmethod
+ def setUpTestData(cls):
+ super().setUpTestData()
+ MockAPIData.setUpServices()
+
+ """
+
+ def __init__(self):
+ self.user_initiator = DigidUserFactory(
+ bsn="100000001",
+ email="initiator@example.com",
+ )
+ self.zaak_type = generate_oas_component(
+ "ztc",
+ "schemas/ZaakType",
+ uuid="aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ url=f"{CATALOGI_ROOT}zaaktype/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ catalogus=f"{CATALOGI_ROOT}catalogussen/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ identificatie="My Zaaktype",
+ indicatieInternOfExtern="extern",
+ omschrijving="My Zaaktype omschrijving",
+ )
+ self.status_type_initial = generate_oas_component(
+ "ztc",
+ "schemas/StatusType",
+ url=f"{CATALOGI_ROOT}statustypen/aaaaaaaa-aaaa-aaaa-aaaa-111111111111",
+ zaaktype=self.zaak_type["url"],
+ informeren=True,
+ volgnummer=1,
+ omschrijving="initial",
+ isEindStatus=False,
+ )
+ self.status_type_final = generate_oas_component(
+ "ztc",
+ "schemas/StatusType",
+ url=f"{CATALOGI_ROOT}statustypen/aaaaaaaa-aaaa-aaaa-aaaa-222222222222",
+ zaaktype=self.zaak_type["url"],
+ informeren=True,
+ volgnummer=2,
+ omschrijving="final",
+ isEindStatus=True,
+ )
+ self.zaak = generate_oas_component(
+ "zrc",
+ "schemas/Zaak",
+ url=f"{ZAKEN_ROOT}zaken/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ zaaktype=self.zaak_type["url"],
+ status=f"{ZAKEN_ROOT}statussen/aaaaaaaa-aaaa-aaaa-aaaa-222222222222",
+ resultaat=f"{ZAKEN_ROOT}resultaten/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ vertrouwelijkheidaanduiding=VertrouwelijkheidsAanduidingen.openbaar,
+ )
+ self.status_initial = generate_oas_component(
+ "zrc",
+ "schemas/Status",
+ url=f"{ZAKEN_ROOT}statussen/aaaaaaaa-aaaa-aaaa-aaaa-111111111111",
+ zaak=self.zaak["url"],
+ statustype=self.status_type_initial["url"],
+ )
+ self.status_final = generate_oas_component(
+ "zrc",
+ "schemas/Status",
+ url=f"{ZAKEN_ROOT}statussen/aaaaaaaa-aaaa-aaaa-aaaa-222222222222",
+ zaak=self.zaak["url"],
+ statustype=self.status_type_final["url"],
+ )
+
+ self.informatie_object = generate_oas_component(
+ "drc",
+ "schemas/EnkelvoudigInformatieObject",
+ url=f"{DOCUMENTEN_ROOT}enkelvoudiginformatieobjecten/aaaaaaaa-0001-bbbb-aaaa-aaaaaaaaaaaa",
+ informatieobjecttype=f"{CATALOGI_ROOT}informatieobjecttypen/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+ status="definitief",
+ vertrouwelijkheidaanduiding=VertrouwelijkheidsAanduidingen.openbaar,
+ )
+ self.zaak_informatie_object = generate_oas_component(
+ "zrc",
+ "schemas/ZaakInformatieObject",
+ url=f"{ZAKEN_ROOT}zaakinformatieobjecten/aaaaaaaa-0001-aaaa-aaaa-aaaaaaaaaaaa",
+ informatieobject=self.informatie_object["url"],
+ zaak=self.zaak["url"],
+ )
+
+ self.role_initiator = generate_oas_component(
+ "zrc",
+ "schemas/Rol",
+ url=f"{ZAKEN_ROOT}rollen/aaaaaaaa-0001-aaaa-aaaa-aaaaaaaaaaaa",
+ omschrijvingGeneriek=RolOmschrijving.initiator,
+ betrokkeneType=RolTypes.natuurlijk_persoon,
+ betrokkeneIdentificatie={
+ "inpBsn": self.user_initiator.bsn,
+ },
+ )
+
+ self.case_roles = [self.role_initiator]
+ self.status_history = [self.status_initial, self.status_final]
+
+ self.status_notification = NotificationFactory(
+ resource="status",
+ actie="update",
+ resource_url=self.status_final["url"],
+ hoofd_object=self.zaak["url"],
+ )
+ self.zio_notification = NotificationFactory(
+ resource="zaakinformatieobject",
+ actie="create",
+ resource_url=self.zaak_informatie_object["url"],
+ hoofd_object=self.zaak["url"],
+ )
+
+ def setUpOASMocks(self, m):
+ mock_service_oas_get(m, ZAKEN_ROOT, "zrc")
+ mock_service_oas_get(m, CATALOGI_ROOT, "ztc")
+ mock_service_oas_get(m, DOCUMENTEN_ROOT, "drc")
+
+ def install_mocks(self, m, *, res404: Optional[List[str]] = None) -> "MockData":
+ self.setUpOASMocks(m)
+ if res404 is None:
+ res404 = []
+ for attr in res404:
+ if not hasattr(self, attr):
+ raise Exception("configuration error")
+
+ if "case_roles" in res404:
+ m.get(f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}", status_code=404)
+ else:
+ m.get(
+ f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}",
+ json=paginated_response(self.case_roles),
+ )
+
+ if "status_history" in res404:
+ m.get(f"{ZAKEN_ROOT}statussen?zaak={self.zaak['url']}", status_code=404)
+ else:
+ m.get(
+ f"{ZAKEN_ROOT}statussen?zaak={self.zaak['url']}",
+ json=paginated_response(self.status_history),
+ )
+
+ for resource_attr in [
+ "zaak",
+ "zaak_type",
+ "status_initial",
+ "status_final",
+ "status_type_initial",
+ "status_type_final",
+ "status_type_final",
+ "informatie_object",
+ "zaak_informatie_object",
+ ]:
+ resource = getattr(self, resource_attr)
+ if resource_attr in res404:
+ m.get(resource["url"], status_code=404)
+ else:
+ m.get(resource["url"], json=resource)
+
+ return self
+
+ @classmethod
+ def setUpServices(cls):
+ # openzaak config
+ config = OpenZaakConfig.get_solo()
+ config.zaak_service = ServiceFactory(api_root=ZAKEN_ROOT, api_type=APITypes.zrc)
+ config.catalogi_service = ServiceFactory(
+ api_root=CATALOGI_ROOT, api_type=APITypes.ztc
+ )
+ config.document_service = ServiceFactory(
+ api_root=DOCUMENTEN_ROOT, api_type=APITypes.drc
+ )
+ config.zaak_max_confidentiality = VertrouwelijkheidsAanduidingen.openbaar
+ config.save()
diff --git a/src/open_inwoner/openzaak/tests/test_notification_handler.py b/src/open_inwoner/openzaak/tests/test_notification_handler.py
deleted file mode 100644
index d466717dd..000000000
--- a/src/open_inwoner/openzaak/tests/test_notification_handler.py
+++ /dev/null
@@ -1,774 +0,0 @@
-import logging
-from typing import List, Optional
-from unittest.mock import Mock, patch
-
-from django.core import mail
-from django.test import TestCase
-from django.urls import reverse
-from django.utils.formats import date_format
-
-import requests_mock
-from notifications_api_common.models import NotificationsConfig
-from zgw_consumers.api_models.base import factory
-from zgw_consumers.api_models.constants import (
- RolOmschrijving,
- RolTypes,
- VertrouwelijkheidsAanduidingen,
-)
-from zgw_consumers.constants import APITypes
-from zgw_consumers.test import generate_oas_component, mock_service_oas_get
-
-from open_inwoner.accounts.tests.factories import DigidUserFactory, UserFactory
-from open_inwoner.openzaak.notifications import (
- get_emailable_initiator_users_from_roles,
- get_np_initiator_bsns_from_roles,
- handle_status_update,
- handle_zaken_notification,
- send_status_update_email,
-)
-from open_inwoner.openzaak.tests.factories import (
- NotificationFactory,
- ServiceFactory,
- ZaakTypeConfigFactory,
- generate_rol,
-)
-
-from ...configurations.models import SiteConfiguration
-from ...utils.test import ClearCachesMixin, paginated_response
-from ...utils.tests.helpers import AssertTimelineLogMixin, Lookups
-from ..api_models import Status, StatusType, Zaak, ZaakType
-from ..models import OpenZaakConfig
-from ..utils import format_zaak_identificatie
-from .shared import CATALOGI_ROOT, DOCUMENTEN_ROOT, ZAKEN_ROOT
-
-
-class MockAPIData:
- """
- initializes isolated and valid data for a complete mock-request API flow,
- allows to manipulate data per test to break it,
- and still get dry/readable access to the data for assertions
-
- usage:
-
- data = MockAPIData()
-
- # change some resources
- data.zaak["some_field"] = "a different value"
-
- # install to your @requests_mock.Mocker()
- data.install_mocks(m)
-
- # install but return 404 for some resource
- data.install_mocks(m, res404=["zaak"])
-
- """
-
- def __init__(self):
- self.user_initiator = DigidUserFactory(
- bsn="100000001",
- email="initiator@example.com",
- )
- self.zaak_type = generate_oas_component(
- "ztc",
- "schemas/ZaakType",
- url=f"{CATALOGI_ROOT}zaaktype/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
- catalogus=f"{CATALOGI_ROOT}catalogussen/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
- identificatie="My Zaaktype",
- indicatieInternOfExtern="extern",
- )
- self.status_type_initial = generate_oas_component(
- "ztc",
- "schemas/StatusType",
- url=f"{CATALOGI_ROOT}statustypen/aaaaaaaa-aaaa-aaaa-aaaa-111111111111",
- zaaktype=self.zaak_type["url"],
- informeren=True,
- volgnummer=1,
- omschrijving="initial",
- isEindStatus=False,
- )
- self.status_type_final = generate_oas_component(
- "ztc",
- "schemas/StatusType",
- url=f"{CATALOGI_ROOT}statustypen/aaaaaaaa-aaaa-aaaa-aaaa-222222222222",
- zaaktype=self.zaak_type["url"],
- informeren=True,
- volgnummer=2,
- omschrijving="final",
- isEindStatus=True,
- )
- self.zaak = generate_oas_component(
- "zrc",
- "schemas/Zaak",
- url=f"{ZAKEN_ROOT}zaken/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
- zaaktype=self.zaak_type["url"],
- status=f"{ZAKEN_ROOT}statussen/aaaaaaaa-aaaa-aaaa-aaaa-222222222222",
- resultaat=f"{ZAKEN_ROOT}resultaten/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
- vertrouwelijkheidaanduiding=VertrouwelijkheidsAanduidingen.openbaar,
- )
- self.status_initial = generate_oas_component(
- "zrc",
- "schemas/Status",
- url=f"{ZAKEN_ROOT}statussen/aaaaaaaa-aaaa-aaaa-aaaa-111111111111",
- zaak=self.zaak["url"],
- statustype=self.status_type_initial["url"],
- )
- self.status_final = generate_oas_component(
- "zrc",
- "schemas/Status",
- url=f"{ZAKEN_ROOT}statussen/aaaaaaaa-aaaa-aaaa-aaaa-222222222222",
- zaak=self.zaak["url"],
- statustype=self.status_type_final["url"],
- )
-
- self.role_initiator = generate_oas_component(
- "zrc",
- "schemas/Rol",
- url=f"{ZAKEN_ROOT}rollen/aaaaaaaa-0001-aaaa-aaaa-aaaaaaaaaaaa",
- omschrijvingGeneriek=RolOmschrijving.initiator,
- betrokkeneType=RolTypes.natuurlijk_persoon,
- betrokkeneIdentificatie={
- "inpBsn": self.user_initiator.bsn,
- },
- )
-
- self.case_roles = [self.role_initiator]
- self.status_history = [self.status_initial, self.status_final]
-
- self.notification = NotificationFactory(
- resource="status",
- actie="update",
- resource_url=self.status_final["url"],
- hoofd_object=self.zaak["url"],
- )
-
- def setUpOASMocks(self, m):
- mock_service_oas_get(m, ZAKEN_ROOT, "zrc")
- mock_service_oas_get(m, CATALOGI_ROOT, "ztc")
-
- def install_mocks(self, m, *, res404: Optional[List[str]] = None) -> "MockData":
- self.setUpOASMocks(m)
- if res404 is None:
- res404 = []
- for attr in res404:
- if not hasattr(self, attr):
- raise Exception("configuration error")
-
- if "case_roles" in res404:
- m.get(f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}", status_code=404)
- else:
- m.get(
- f"{ZAKEN_ROOT}rollen?zaak={self.zaak['url']}",
- json=paginated_response(self.case_roles),
- )
-
- if "status_history" in res404:
- m.get(f"{ZAKEN_ROOT}statussen?zaak={self.zaak['url']}", status_code=404)
- else:
- m.get(
- f"{ZAKEN_ROOT}statussen?zaak={self.zaak['url']}",
- json=paginated_response(self.status_history),
- )
-
- for resource_attr in [
- "zaak",
- "zaak_type",
- "status_initial",
- "status_final",
- "status_type_initial",
- "status_type_final",
- ]:
- resource = getattr(self, resource_attr)
- if resource_attr in res404:
- m.get(resource["url"], status_code=404)
- else:
- m.get(resource["url"], json=resource)
-
- return self
-
-
-@requests_mock.Mocker()
-@patch("open_inwoner.openzaak.notifications.handle_status_update")
-class NotificationHandlerTestCase(AssertTimelineLogMixin, ClearCachesMixin, TestCase):
- maxDiff = None
- config: OpenZaakConfig
- note_config: NotificationsConfig
-
- @classmethod
- def setUpTestData(cls):
- super().setUpTestData()
-
- # services
- cls.zaak_service = ServiceFactory(api_root=ZAKEN_ROOT, api_type=APITypes.zrc)
- cls.catalogi_service = ServiceFactory(
- api_root=CATALOGI_ROOT, api_type=APITypes.ztc
- )
- cls.document_service = ServiceFactory(
- api_root=DOCUMENTEN_ROOT, api_type=APITypes.drc
- )
- # openzaak config
- cls.config = OpenZaakConfig.get_solo()
- cls.config.zaak_service = cls.zaak_service
- cls.config.catalogi_service = cls.catalogi_service
- cls.config.document_service = cls.document_service
- cls.config.zaak_max_confidentiality = VertrouwelijkheidsAanduidingen.openbaar
- cls.config.save()
-
- def test_handle_zaken_notification(self, m, mock_handle: Mock):
- """
- happy-flow from valid data calls the (mocked) handle_status_update()
- """
- data = MockAPIData().install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- mock_handle.assert_called_once()
-
- # check call arguments
- args = mock_handle.call_args.args
- self.assertEqual(args[0], data.user_initiator)
- self.assertEqual(args[1].url, data.zaak["url"])
- self.assertEqual(args[2].url, data.status_final["url"])
-
- self.assertTimelineLog(
- "accepted notification: attempt informing users ",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
-
- def test_bails_when_bad_notification_channel(self, m, mock_handle: Mock):
- notification = NotificationFactory(kanaal="not_zaken")
- with self.assertRaisesRegex(
- Exception, r"^handler expects kanaal 'zaken' but received 'not_zaken'"
- ):
- handle_zaken_notification(notification)
-
- mock_handle.assert_not_called()
-
- def test_bails_when_bad_notification_resource(self, m, mock_handle: Mock):
- notification = NotificationFactory(resource="not_status")
-
- handle_zaken_notification(notification)
-
- self.assertTimelineLog(
- "ignored notification: resource is not 'status' but 'not_status' for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_no_roles_found_for_case(self, m, mock_handle: Mock):
- data = MockAPIData()
- data.install_mocks(m, res404=["case_roles"])
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- "ignored notification: cannot retrieve rollen for case https://",
- lookup=Lookups.startswith,
- level=logging.ERROR,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_no_emailable_users_are_found_for_roles(
- self, m, mock_handle: Mock
- ):
- data = MockAPIData()
- data.user_initiator.delete()
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- "ignored notification: no users with bsn and valid email as (mede)initiators in case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_cannot_fetch_status_history(self, m, mock_handle: Mock):
- data = MockAPIData()
- data.install_mocks(m, res404=["status_history"])
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: cannot retrieve status_history for case https://",
- lookup=Lookups.startswith,
- level=logging.ERROR,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_status_history_is_single_initial_item(
- self, m, mock_handle: Mock
- ):
- data = MockAPIData()
- data.status_history = [data.status_initial]
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: skip initial status notification for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_cannot_fetch_status_type(self, m, mock_handle: Mock):
- data = MockAPIData()
- data.install_mocks(m, res404=["status_type_final"])
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: cannot retrieve status_type {data.status_type_final['url']} for case https://",
- lookup=Lookups.startswith,
- level=logging.ERROR,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_status_type_not_marked_as_informeren(
- self, m, mock_handle: Mock
- ):
- data = MockAPIData()
- data.status_type_final["informeren"] = False
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: status_type.informeren is false for status {data.status_final['url']} and case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_cannot_fetch_case(self, m, mock_handle: Mock):
- data = MockAPIData()
- data.install_mocks(m, res404=["zaak"])
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: cannot retrieve case https://",
- lookup=Lookups.startswith,
- level=logging.ERROR,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_cannot_fetch_case_type(self, m, mock_handle: Mock):
- data = MockAPIData()
- data.install_mocks(m, res404=["zaak_type"])
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: cannot retrieve case_type https://",
- lookup=Lookups.startswith,
- level=logging.ERROR,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_case_not_visible_because_confidentiality(
- self, m, mock_handle: Mock
- ):
- data = MockAPIData()
- data.zaak["vertrouwelijkheidaanduiding"] = VertrouwelijkheidsAanduidingen.geheim
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: case not visible after applying website visibility filter for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_case_not_visible_because_internal_case(
- self, m, mock_handle: Mock
- ):
- data = MockAPIData()
- data.zaak_type["indicatieInternOfExtern"] = "intern"
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: case not visible after applying website visibility filter for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_skip_informeren_is_set_and_no_zaaktypeconfig_is_found(
- self, m, mock_handle: Mock
- ):
- oz_config = OpenZaakConfig.get_solo()
- oz_config.skip_notification_statustype_informeren = True
- oz_config.save()
-
- data = MockAPIData()
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{data.zaak_type['identificatie']}' for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_skip_informeren_is_set_and_no_zaaktypeconfig_is_found_from_zaaktype_none_catalog(
- self, m, mock_handle: Mock
- ):
- oz_config = OpenZaakConfig.get_solo()
- oz_config.skip_notification_statustype_informeren = True
- oz_config.save()
-
- data = MockAPIData()
- data.zaak_type["catalogus"] = None
- data.install_mocks(m)
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{data.zaak_type['identificatie']}' for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_handle_notification_when_skip_informeren_is_set_and_zaaktypeconfig_is_found(
- self, m, mock_handle: Mock
- ):
- oz_config = OpenZaakConfig.get_solo()
- oz_config.skip_notification_statustype_informeren = True
- oz_config.save()
-
- data = MockAPIData().install_mocks(m)
-
- ZaakTypeConfigFactory.create(
- catalogus__url=data.zaak_type["catalogus"],
- identificatie=data.zaak_type["identificatie"],
- # set this to notify
- notify_status_changes=True,
- )
-
- handle_zaken_notification(data.notification)
-
- mock_handle.assert_called_once()
-
- # check call arguments
- args = mock_handle.call_args.args
- self.assertEqual(args[0], data.user_initiator)
- self.assertEqual(args[1].url, data.zaak["url"])
- self.assertEqual(args[2].url, data.status_final["url"])
-
- self.assertTimelineLog(
- "accepted notification: attempt informing users ",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
-
- def test_handle_notification_when_skip_informeren_is_set_and_zaaktypeconfig_is_found_from_zaaktype_none_catalog(
- self, m, mock_handle: Mock
- ):
- oz_config = OpenZaakConfig.get_solo()
- oz_config.skip_notification_statustype_informeren = True
- oz_config.save()
-
- data = MockAPIData()
- data.zaak_type["catalogus"] = None
- data.install_mocks(m)
-
- ZaakTypeConfigFactory.create(
- catalogus=None,
- identificatie=data.zaak_type["identificatie"],
- # set this to notify
- notify_status_changes=True,
- )
-
- handle_zaken_notification(data.notification)
-
- mock_handle.assert_called_once()
-
- # check call arguments
- args = mock_handle.call_args.args
- self.assertEqual(args[0], data.user_initiator)
- self.assertEqual(args[1].url, data.zaak["url"])
- self.assertEqual(args[2].url, data.status_final["url"])
-
- self.assertTimelineLog(
- "accepted notification: attempt informing users ",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
-
- def test_bails_when_skip_informeren_is_set_and_zaaktypeconfig_is_found_but_not_set(
- self, m, mock_handle: Mock
- ):
- oz_config = OpenZaakConfig.get_solo()
- oz_config.skip_notification_statustype_informeren = True
- oz_config.save()
-
- data = MockAPIData()
- data.install_mocks(m)
-
- ZaakTypeConfigFactory.create(
- catalogus__url=data.zaak_type["catalogus"],
- identificatie=data.zaak_type["identificatie"],
- notify_status_changes=False,
- )
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: case_type configuration '{data.zaak_type['identificatie']}' found but 'notify_status_changes' is False for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
- def test_bails_when_skip_informeren_is_set_and_zaaktypeconfig_is_not_found_because_different_catalog(
- self, m, mock_handle: Mock
- ):
- oz_config = OpenZaakConfig.get_solo()
- oz_config.skip_notification_statustype_informeren = True
- oz_config.save()
-
- data = MockAPIData()
- data.install_mocks(m)
-
- ZaakTypeConfigFactory.create(
- catalogus__url="http://not-the-catalogus.xyz",
- identificatie=data.zaak_type["identificatie"],
- notify_status_changes=False,
- )
-
- handle_zaken_notification(data.notification)
-
- self.assertTimelineLog(
- f"ignored notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{data.zaak_type['identificatie']}' for case https://",
- lookup=Lookups.startswith,
- level=logging.INFO,
- )
- mock_handle.assert_not_called()
-
-
-class NotificationHandlerEmailTestCase(TestCase):
- @patch("open_inwoner.openzaak.notifications.send_status_update_email")
- def test_handle_status_update(self, mock_send: Mock):
- data = MockAPIData()
- user = data.user_initiator
-
- case = factory(Zaak, data.zaak)
- case.zaaktype = factory(ZaakType, data.zaak_type)
-
- status = factory(Status, data.status_final)
- status.statustype = factory(StatusType, data.status_type_final)
-
- # first call
- handle_status_update(user, case, status)
-
- mock_send.assert_called_once()
-
- # check call arguments
- args = mock_send.call_args.args
- self.assertEqual(args[0], user)
- self.assertEqual(args[1].url, case.url)
- self.assertEqual(args[2].url, status.url)
-
- mock_send.reset_mock()
-
- # second call with same case/status
- handle_status_update(user, case, status)
-
- # no duplicate mail for this user/case/status
- mock_send.assert_not_called()
-
- # other user is fine
- other_user = UserFactory.create()
- handle_status_update(other_user, case, status)
-
- mock_send.assert_called_once()
-
- args = mock_send.call_args.args
- self.assertEqual(args[0], other_user)
-
- @patch(
- "open_inwoner.openzaak.notifications.format_zaak_identificatie",
- wraps=format_zaak_identificatie,
- )
- def test_send_status_update_email(self, spy_format):
- config = SiteConfiguration.get_solo()
- data = MockAPIData()
-
- user = data.user_initiator
-
- case = factory(Zaak, data.zaak)
- case.zaaktype = factory(ZaakType, data.zaak_type)
-
- status = factory(Status, data.status_final)
- status.statustype = factory(StatusType, data.status_type_final)
-
- case_url = reverse("accounts:case_status", kwargs={"object_id": str(case.uuid)})
-
- send_status_update_email(user, case, status)
-
- spy_format.assert_called_once()
-
- self.assertEqual(len(mail.outbox), 1)
- email = mail.outbox[0]
- self.assertEqual(email.to, [user.email])
- self.assertIn(config.name, email.subject)
-
- body_html = email.alternatives[0][0]
- self.assertIn(case.identificatie, body_html)
- self.assertIn(case.zaaktype.omschrijving, body_html)
- self.assertIn(date_format(case.startdatum), body_html)
- self.assertIn(case_url, body_html)
- self.assertIn(config.name, body_html)
-
-
-class NotificationHandlerUtilsTestCase(TestCase):
- def test_get_np_initiator_bsns_from_roles(self):
- # roles we're interested in
- find_rol_1 = generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": "100000001"},
- RolOmschrijving.initiator,
- )
- find_rol_2 = generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": "100000002"},
- RolOmschrijving.medeinitiator,
- )
- roles = [
- find_rol_1,
- find_rol_2,
- # add some roles we're not interested in
- #
- # interested but duplicate BSN
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": "100000001"},
- RolOmschrijving.medeinitiator,
- ),
- # bad type
- generate_rol(
- RolTypes.vestiging,
- {"inpBsn": "404000001"},
- RolOmschrijving.initiator,
- ),
- # bad description
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": "404000002"},
- RolOmschrijving.behandelaar,
- ),
- # bad identification
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"not_the_expected_field": 123},
- RolOmschrijving.initiator,
- ),
- ]
- # filtered and de-duplicated
- expected = {
- "100000001",
- "100000002",
- }
- actual = get_np_initiator_bsns_from_roles(roles)
- self.assertEqual(set(actual), expected)
-
- def test_get_emailable_initiator_users_from_roles(self):
- # users we're interested in
- user_1 = DigidUserFactory(bsn="100000001", email="user_1@example.com")
- user_2 = DigidUserFactory(bsn="100000002", email="user_2@example.com")
-
- # not active
- user_not_active = DigidUserFactory(
- bsn="404000003", is_active=False, email="user_not_active@example.com"
- )
-
- # no email
- user_no_email = DigidUserFactory(bsn="404000004", email="")
-
- # placeholder email
- user_placeholder_email = DigidUserFactory(
- bsn="404000005", email="user_placeholder_email@example.org"
- )
- # bad role
- user_bad_role = DigidUserFactory(
- bsn="404000006", email="user_bad_role@example.com"
- )
-
- # not part of roles
- user_not_a_role = DigidUserFactory(
- bsn="404000007", email="user_not_a_role@example.com"
- )
-
- # not a digid user
- user_no_bsn = UserFactory(bsn="", email="user_no_bsn@example.com")
-
- # good roles
- role_1 = generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_1.bsn},
- RolOmschrijving.initiator,
- )
- role_2 = generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_2.bsn},
- RolOmschrijving.medeinitiator,
- )
- roles = [
- role_1,
- role_2,
- # add some bad roles
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_not_active.bsn},
- RolOmschrijving.initiator,
- ),
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_no_email.bsn},
- RolOmschrijving.initiator,
- ),
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_placeholder_email.bsn},
- RolOmschrijving.initiator,
- ),
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_bad_role.bsn},
- RolOmschrijving.behandelaar,
- ),
- # duplicate with different role
- generate_rol(
- RolTypes.natuurlijk_persoon,
- {"inpBsn": user_1.bsn},
- RolOmschrijving.medeinitiator,
- ),
- ]
-
- # verify we have a lot of Roles with initiators & bsn's
- check_roles = get_np_initiator_bsns_from_roles(roles)
- expected_roles = {
- user_1.bsn,
- user_2.bsn,
- user_not_active.bsn,
- user_no_email.bsn,
- user_placeholder_email.bsn,
- }
- self.assertEqual(set(check_roles), expected_roles)
-
- # of all the Users with Roles only these match all conditions to actually get notified
- expected = {user_1, user_2}
- actual = get_emailable_initiator_users_from_roles(roles)
-
- self.assertEqual(set(actual), expected)
diff --git a/src/open_inwoner/openzaak/tests/test_notification_utils.py b/src/open_inwoner/openzaak/tests/test_notification_utils.py
new file mode 100644
index 000000000..261e8b2d5
--- /dev/null
+++ b/src/open_inwoner/openzaak/tests/test_notification_utils.py
@@ -0,0 +1,195 @@
+from unittest.mock import patch
+
+from django.core import mail
+from django.test import TestCase
+from django.urls import reverse
+from django.utils.formats import date_format
+
+from zgw_consumers.api_models.base import factory
+from zgw_consumers.api_models.constants import RolOmschrijving, RolTypes
+
+from open_inwoner.accounts.tests.factories import DigidUserFactory, UserFactory
+from open_inwoner.configurations.models import SiteConfiguration
+from open_inwoner.openzaak.notifications import (
+ get_emailable_initiator_users_from_roles,
+ get_np_initiator_bsns_from_roles,
+ send_case_update_email,
+)
+from open_inwoner.openzaak.tests.factories import generate_rol
+
+from ..api_models import Zaak, ZaakType
+from ..utils import format_zaak_identificatie
+from .test_notification_data import MockAPIData
+
+
+class NotificationHandlerUtilsTestCase(TestCase):
+ @patch(
+ "open_inwoner.openzaak.notifications.format_zaak_identificatie",
+ wraps=format_zaak_identificatie,
+ )
+ def test_send_case_update_email(self, spy_format):
+ config = SiteConfiguration.get_solo()
+ data = MockAPIData()
+
+ user = data.user_initiator
+
+ case = factory(Zaak, data.zaak)
+ case.zaaktype = factory(ZaakType, data.zaak_type)
+
+ case_url = reverse("accounts:case_status", kwargs={"object_id": str(case.uuid)})
+
+ send_case_update_email(user, case)
+
+ spy_format.assert_called_once()
+
+ self.assertEqual(len(mail.outbox), 1)
+ email = mail.outbox[0]
+ self.assertEqual(email.to, [user.email])
+ self.assertIn(config.name, email.subject)
+
+ body_html = email.alternatives[0][0]
+ self.assertIn(case.identificatie, body_html)
+ self.assertIn(case.zaaktype.omschrijving, body_html)
+ self.assertIn(date_format(case.startdatum), body_html)
+ self.assertIn(case_url, body_html)
+ self.assertIn(config.name, body_html)
+
+ def test_get_np_initiator_bsns_from_roles(self):
+ # roles we're interested in
+ find_rol_1 = generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": "100000001"},
+ RolOmschrijving.initiator,
+ )
+ find_rol_2 = generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": "100000002"},
+ RolOmschrijving.medeinitiator,
+ )
+ roles = [
+ find_rol_1,
+ find_rol_2,
+ # add some roles we're not interested in
+ #
+ # interested but duplicate BSN
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": "100000001"},
+ RolOmschrijving.medeinitiator,
+ ),
+ # bad type
+ generate_rol(
+ RolTypes.vestiging,
+ {"inpBsn": "404000001"},
+ RolOmschrijving.initiator,
+ ),
+ # bad description
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": "404000002"},
+ RolOmschrijving.behandelaar,
+ ),
+ # bad identification
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"not_the_expected_field": 123},
+ RolOmschrijving.initiator,
+ ),
+ ]
+ # filtered and de-duplicated
+ expected = {
+ "100000001",
+ "100000002",
+ }
+ actual = get_np_initiator_bsns_from_roles(roles)
+ self.assertEqual(set(actual), expected)
+
+ def test_get_emailable_initiator_users_from_roles(self):
+ # users we're interested in
+ user_1 = DigidUserFactory(bsn="100000001", email="user_1@example.com")
+ user_2 = DigidUserFactory(bsn="100000002", email="user_2@example.com")
+
+ # not active
+ user_not_active = DigidUserFactory(
+ bsn="404000003", is_active=False, email="user_not_active@example.com"
+ )
+
+ # no email
+ user_no_email = DigidUserFactory(bsn="404000004", email="")
+
+ # placeholder email
+ user_placeholder_email = DigidUserFactory(
+ bsn="404000005", email="user_placeholder_email@example.org"
+ )
+ # bad role
+ user_bad_role = DigidUserFactory(
+ bsn="404000006", email="user_bad_role@example.com"
+ )
+
+ # not part of roles
+ user_not_a_role = DigidUserFactory(
+ bsn="404000007", email="user_not_a_role@example.com"
+ )
+
+ # not a digid user
+ user_no_bsn = UserFactory(bsn="", email="user_no_bsn@example.com")
+
+ # good roles
+ role_1 = generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_1.bsn},
+ RolOmschrijving.initiator,
+ )
+ role_2 = generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_2.bsn},
+ RolOmschrijving.medeinitiator,
+ )
+ roles = [
+ role_1,
+ role_2,
+ # add some bad roles
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_not_active.bsn},
+ RolOmschrijving.initiator,
+ ),
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_no_email.bsn},
+ RolOmschrijving.initiator,
+ ),
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_placeholder_email.bsn},
+ RolOmschrijving.initiator,
+ ),
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_bad_role.bsn},
+ RolOmschrijving.behandelaar,
+ ),
+ # duplicate with different role
+ generate_rol(
+ RolTypes.natuurlijk_persoon,
+ {"inpBsn": user_1.bsn},
+ RolOmschrijving.medeinitiator,
+ ),
+ ]
+
+ # verify we have a lot of Roles with initiators & bsn's
+ check_roles = get_np_initiator_bsns_from_roles(roles)
+ expected_roles = {
+ user_1.bsn,
+ user_2.bsn,
+ user_not_active.bsn,
+ user_no_email.bsn,
+ user_placeholder_email.bsn,
+ }
+ self.assertEqual(set(check_roles), expected_roles)
+
+ # of all the Users with Roles only these match all conditions to actually get notified
+ expected = {user_1, user_2}
+ actual = get_emailable_initiator_users_from_roles(roles)
+
+ self.assertEqual(set(actual), expected)
diff --git a/src/open_inwoner/openzaak/tests/test_notification_zaak_infoobject.py b/src/open_inwoner/openzaak/tests/test_notification_zaak_infoobject.py
new file mode 100644
index 000000000..4f0283dea
--- /dev/null
+++ b/src/open_inwoner/openzaak/tests/test_notification_zaak_infoobject.py
@@ -0,0 +1,318 @@
+import logging
+from unittest.mock import Mock, patch
+
+from django.test import TestCase
+
+import requests_mock
+from notifications_api_common.models import NotificationsConfig
+from zgw_consumers.api_models.base import factory
+from zgw_consumers.api_models.constants import VertrouwelijkheidsAanduidingen
+
+from open_inwoner.accounts.tests.factories import UserFactory
+from open_inwoner.openzaak.notifications import (
+ handle_zaakinformatieobject_update,
+ handle_zaken_notification,
+)
+from open_inwoner.openzaak.tests.factories import (
+ NotificationFactory,
+ ZaakTypeInformatieObjectTypeConfigFactory,
+)
+from open_inwoner.utils.test import ClearCachesMixin
+from open_inwoner.utils.tests.helpers import AssertTimelineLogMixin, Lookups
+
+from ..api_models import InformatieObject, Zaak, ZaakInformatieObject, ZaakType
+from ..models import OpenZaakConfig, ZaakTypeInformatieObjectTypeConfig
+from .test_notification_data import MockAPIData
+
+
+@requests_mock.Mocker()
+@patch("open_inwoner.openzaak.notifications.handle_zaakinformatieobject_update")
+class ZaakInformatieObjectNotificationHandlerTestCase(
+ AssertTimelineLogMixin, ClearCachesMixin, TestCase
+):
+ maxDiff = None
+ config: OpenZaakConfig
+ note_config: NotificationsConfig
+
+ @classmethod
+ def setUpTestData(cls):
+ super().setUpTestData()
+ MockAPIData.setUpServices()
+
+ def test_zio_handle_zaken_notification(self, m, mock_handle: Mock):
+ """
+ happy-flow from valid data calls the (mocked) handle_zaakinformatieobject()
+ """
+ data = MockAPIData().install_mocks(m)
+
+ ZaakTypeInformatieObjectTypeConfigFactory.from_case_type_info_object_dicts(
+ data.zaak_type, data.informatie_object, document_notification_enabled=True
+ )
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_called_once()
+
+ # check call arguments
+ args = mock_handle.call_args.args
+ self.assertEqual(args[0], data.user_initiator)
+ self.assertEqual(args[1].url, data.zaak["url"])
+ self.assertEqual(args[2].url, data.zaak_informatie_object["url"])
+
+ self.assertTimelineLog(
+ "accepted zaakinformatieobject notification: attempt informing users ",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ # start of generic checks
+
+ def test_zio_bails_when_bad_notification_channel(self, m, mock_handle: Mock):
+ notification = NotificationFactory(kanaal="not_zaken")
+ with self.assertRaisesRegex(
+ Exception, r"^handler expects kanaal 'zaken' but received 'not_zaken'"
+ ):
+ handle_zaken_notification(notification)
+
+ mock_handle.assert_not_called()
+
+ def test_zio_bails_when_bad_notification_resource(self, m, mock_handle: Mock):
+ notification = NotificationFactory(resource="not_status")
+
+ handle_zaken_notification(notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ "ignored not_status notification: resource is not 'status' or 'zaakinformatieobject' but 'not_status' for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_zio_bails_when_no_roles_found_for_case(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["case_roles"])
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ "ignored zaakinformatieobject notification: cannot retrieve rollen for case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_zio_bails_when_no_emailable_users_are_found_for_roles(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.user_initiator.delete()
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ "ignored zaakinformatieobject notification: no users with bsn and valid email as (mede)initiators in case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_zio_bails_when_cannot_fetch_case(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["zaak"])
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: cannot retrieve case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_zio_bails_when_cannot_fetch_case_type(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["zaak_type"])
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: cannot retrieve case_type https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_zio_bails_when_case_not_visible_because_confidentiality(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.zaak["vertrouwelijkheidaanduiding"] = VertrouwelijkheidsAanduidingen.geheim
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: case not visible after applying website visibility filter for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_zio_bails_when_case_not_visible_because_internal_case(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.zaak_type["indicatieInternOfExtern"] = "intern"
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: case not visible after applying website visibility filter for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ # end of generic checks
+
+ # start of status specific checks
+ def test_zio_bails_when_cannot_fetch_zaak_informatie_object(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["zaak_informatie_object"])
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: cannot retrieve zaakinformatieobject {data.zaak_informatie_object['url']} for case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_zio_bails_when_cannot_fetch_informatie_object(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["informatie_object"])
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: cannot retrieve informatieobject {data.informatie_object['url']} for case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_zio_bails_when_info_object_not_visible_because_confidentiality(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.informatie_object[
+ "vertrouwelijkheidaanduiding"
+ ] = VertrouwelijkheidsAanduidingen.geheim
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: informatieobject not visible after applying website visibility filter for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_zio_bails_when_info_object_not_visible_because_not_definitive(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.informatie_object["status"] = "concept"
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: informatieobject not visible after applying website visibility filter for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_zio_bails_when_zaak_type_info_object_type_config_is_not_found(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData().install_mocks(m)
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: cannot retrieve info_type configuration {data.informatie_object['informatieobjecttype']} and case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_zio_bails_when_zaak_type_info_object_type_config_is_found_not_marked_for_notifications(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData().install_mocks(m)
+
+ ZaakTypeInformatieObjectTypeConfigFactory.from_case_type_info_object_dicts(
+ data.zaak_type,
+ data.informatie_object,
+ document_notification_enabled=False,
+ omschrijving="important document",
+ )
+
+ handle_zaken_notification(data.zio_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored zaakinformatieobject notification: info_type configuration 'important document' {data.informatie_object['informatieobjecttype']} found but 'document_notification_enabled' is False for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ # TODO add some no-catalog variations
+
+
+class NotificationHandlerEmailTestCase(TestCase):
+ @patch("open_inwoner.openzaak.notifications.send_case_update_email")
+ def test_handle_zaak_info_object_update(self, mock_send: Mock):
+ data = MockAPIData()
+ user = data.user_initiator
+
+ case = factory(Zaak, data.zaak)
+ case.zaaktype = factory(ZaakType, data.zaak_type)
+
+ zio = factory(ZaakInformatieObject, data.zaak_informatie_object)
+ zio.informatieobject = factory(InformatieObject, data.informatie_object)
+
+ # first call
+ handle_zaakinformatieobject_update(user, case, zio)
+
+ mock_send.assert_called_once()
+
+ # check call arguments
+ args = mock_send.call_args.args
+ self.assertEqual(args[0], user)
+ self.assertEqual(args[1].url, case.url)
+
+ mock_send.reset_mock()
+
+ # second call with same case/status
+ handle_zaakinformatieobject_update(user, case, zio)
+
+ # no duplicate mail for this user/case/status
+ mock_send.assert_not_called()
+
+ # other user is fine
+ other_user = UserFactory.create()
+ handle_zaakinformatieobject_update(other_user, case, zio)
+
+ mock_send.assert_called_once()
+
+ args = mock_send.call_args.args
+ self.assertEqual(args[0], other_user)
diff --git a/src/open_inwoner/openzaak/tests/test_notification_zaak_status.py b/src/open_inwoner/openzaak/tests/test_notification_zaak_status.py
new file mode 100644
index 000000000..5f882eee5
--- /dev/null
+++ b/src/open_inwoner/openzaak/tests/test_notification_zaak_status.py
@@ -0,0 +1,429 @@
+import logging
+from unittest.mock import Mock, patch
+
+from django.test import TestCase
+
+import requests_mock
+from notifications_api_common.models import NotificationsConfig
+from zgw_consumers.api_models.base import factory
+from zgw_consumers.api_models.constants import VertrouwelijkheidsAanduidingen
+
+from open_inwoner.accounts.tests.factories import UserFactory
+from open_inwoner.openzaak.notifications import (
+ handle_status_update,
+ handle_zaken_notification,
+)
+from open_inwoner.openzaak.tests.factories import (
+ NotificationFactory,
+ ZaakTypeConfigFactory,
+)
+from open_inwoner.utils.test import ClearCachesMixin
+from open_inwoner.utils.tests.helpers import AssertTimelineLogMixin, Lookups
+
+from ..api_models import Status, StatusType, Zaak, ZaakType
+from ..models import OpenZaakConfig
+from .test_notification_data import MockAPIData
+
+
+@requests_mock.Mocker()
+@patch("open_inwoner.openzaak.notifications.handle_status_update")
+class StatusNotificationHandlerTestCase(
+ AssertTimelineLogMixin, ClearCachesMixin, TestCase
+):
+ maxDiff = None
+ config: OpenZaakConfig
+ note_config: NotificationsConfig
+
+ @classmethod
+ def setUpTestData(cls):
+ super().setUpTestData()
+ MockAPIData.setUpServices()
+
+ def test_status_handle_zaken_notification(self, m, mock_handle: Mock):
+ """
+ happy-flow from valid data calls the (mocked) handle_status_update()
+ """
+ data = MockAPIData().install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_called_once()
+
+ # check call arguments
+ args = mock_handle.call_args.args
+ self.assertEqual(args[0], data.user_initiator)
+ self.assertEqual(args[1].url, data.zaak["url"])
+ self.assertEqual(args[2].url, data.status_final["url"])
+
+ self.assertTimelineLog(
+ "accepted status notification: attempt informing users ",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ # start of generic checks
+
+ def test_status_bails_when_bad_notification_channel(self, m, mock_handle: Mock):
+ notification = NotificationFactory(kanaal="not_zaken")
+ with self.assertRaisesRegex(
+ Exception, r"^handler expects kanaal 'zaken' but received 'not_zaken'"
+ ):
+ handle_zaken_notification(notification)
+
+ mock_handle.assert_not_called()
+
+ def test_status_bails_when_bad_notification_resource(self, m, mock_handle: Mock):
+ notification = NotificationFactory(resource="not_status")
+
+ handle_zaken_notification(notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ "ignored not_status notification: resource is not 'status' or 'zaakinformatieobject' but 'not_status' for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_no_roles_found_for_case(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["case_roles"])
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ "ignored status notification: cannot retrieve rollen for case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_status_bails_when_no_emailable_users_are_found_for_roles(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.user_initiator.delete()
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ "ignored status notification: no users with bsn and valid email as (mede)initiators in case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_cannot_fetch_case(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["zaak"])
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: cannot retrieve case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_status_bails_when_cannot_fetch_case_type(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["zaak_type"])
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: cannot retrieve case_type https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_status_bails_when_case_not_visible_because_confidentiality(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.zaak["vertrouwelijkheidaanduiding"] = VertrouwelijkheidsAanduidingen.geheim
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: case not visible after applying website visibility filter for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_case_not_visible_because_internal_case(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.zaak_type["indicatieInternOfExtern"] = "intern"
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: case not visible after applying website visibility filter for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ # end of generic checks
+
+ # start of status specific checks
+
+ def test_status_bails_when_cannot_fetch_status_history(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["status_history"])
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: cannot retrieve status_history for case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_status_bails_when_status_history_is_single_initial_item(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.status_history = [data.status_initial]
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: skip initial status notification for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_cannot_fetch_status_type(self, m, mock_handle: Mock):
+ data = MockAPIData()
+ data.install_mocks(m, res404=["status_type_final"])
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: cannot retrieve status_type {data.status_type_final['url']} for case https://",
+ lookup=Lookups.startswith,
+ level=logging.ERROR,
+ )
+
+ def test_status_bails_when_status_type_not_marked_as_informeren(
+ self, m, mock_handle: Mock
+ ):
+ data = MockAPIData()
+ data.status_type_final["informeren"] = False
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: status_type.informeren is false for status {data.status_final['url']} and case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_skip_informeren_is_set_and_no_zaaktypeconfig_is_found(
+ self, m, mock_handle: Mock
+ ):
+ oz_config = OpenZaakConfig.get_solo()
+ oz_config.skip_notification_statustype_informeren = True
+ oz_config.save()
+
+ data = MockAPIData()
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{data.zaak_type['identificatie']}' for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_skip_informeren_is_set_and_no_zaaktypeconfig_is_found_from_zaaktype_none_catalog(
+ self, m, mock_handle: Mock
+ ):
+ oz_config = OpenZaakConfig.get_solo()
+ oz_config.skip_notification_statustype_informeren = True
+ oz_config.save()
+
+ data = MockAPIData()
+ data.zaak_type["catalogus"] = None
+ data.install_mocks(m)
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{data.zaak_type['identificatie']}' for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_handle_notification_when_skip_informeren_is_set_and_zaaktypeconfig_is_found(
+ self, m, mock_handle: Mock
+ ):
+ oz_config = OpenZaakConfig.get_solo()
+ oz_config.skip_notification_statustype_informeren = True
+ oz_config.save()
+
+ data = MockAPIData().install_mocks(m)
+
+ ZaakTypeConfigFactory.create(
+ catalogus__url=data.zaak_type["catalogus"],
+ identificatie=data.zaak_type["identificatie"],
+ # set this to notify
+ notify_status_changes=True,
+ )
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_called_once()
+
+ # check call arguments
+ args = mock_handle.call_args.args
+ self.assertEqual(args[0], data.user_initiator)
+ self.assertEqual(args[1].url, data.zaak["url"])
+ self.assertEqual(args[2].url, data.status_final["url"])
+
+ self.assertTimelineLog(
+ "accepted status notification: attempt informing users ",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_handle_notification_when_skip_informeren_is_set_and_zaaktypeconfig_is_found_from_zaaktype_none_catalog(
+ self, m, mock_handle: Mock
+ ):
+ oz_config = OpenZaakConfig.get_solo()
+ oz_config.skip_notification_statustype_informeren = True
+ oz_config.save()
+
+ data = MockAPIData()
+ data.zaak_type["catalogus"] = None
+ data.install_mocks(m)
+
+ ZaakTypeConfigFactory.create(
+ catalogus=None,
+ identificatie=data.zaak_type["identificatie"],
+ # set this to notify
+ notify_status_changes=True,
+ )
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_called_once()
+
+ # check call arguments
+ args = mock_handle.call_args.args
+ self.assertEqual(args[0], data.user_initiator)
+ self.assertEqual(args[1].url, data.zaak["url"])
+ self.assertEqual(args[2].url, data.status_final["url"])
+
+ self.assertTimelineLog(
+ "accepted status notification: attempt informing users ",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_skip_informeren_is_set_and_zaaktypeconfig_is_found_but_not_set(
+ self, m, mock_handle: Mock
+ ):
+ oz_config = OpenZaakConfig.get_solo()
+ oz_config.skip_notification_statustype_informeren = True
+ oz_config.save()
+
+ data = MockAPIData()
+ data.install_mocks(m)
+
+ ZaakTypeConfigFactory.create(
+ catalogus__url=data.zaak_type["catalogus"],
+ identificatie=data.zaak_type["identificatie"],
+ notify_status_changes=False,
+ )
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: case_type configuration '{data.zaak_type['identificatie']}' found but 'notify_status_changes' is False for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+ def test_status_bails_when_skip_informeren_is_set_and_zaaktypeconfig_is_not_found_because_different_catalog(
+ self, m, mock_handle: Mock
+ ):
+ oz_config = OpenZaakConfig.get_solo()
+ oz_config.skip_notification_statustype_informeren = True
+ oz_config.save()
+
+ data = MockAPIData()
+ data.install_mocks(m)
+
+ ZaakTypeConfigFactory.create(
+ catalogus__url="http://not-the-catalogus.xyz",
+ identificatie=data.zaak_type["identificatie"],
+ notify_status_changes=False,
+ )
+
+ handle_zaken_notification(data.status_notification)
+
+ mock_handle.assert_not_called()
+ self.assertTimelineLog(
+ f"ignored status notification: 'skip_notification_statustype_informeren' is True but cannot retrieve case_type configuration '{data.zaak_type['identificatie']}' for case https://",
+ lookup=Lookups.startswith,
+ level=logging.INFO,
+ )
+
+
+class NotificationHandlerEmailTestCase(TestCase):
+ @patch("open_inwoner.openzaak.notifications.send_case_update_email")
+ def test_handle_status_update(self, mock_send: Mock):
+ data = MockAPIData()
+ user = data.user_initiator
+
+ case = factory(Zaak, data.zaak)
+ case.zaaktype = factory(ZaakType, data.zaak_type)
+
+ status = factory(Status, data.status_final)
+ status.statustype = factory(StatusType, data.status_type_final)
+
+ # first call
+ handle_status_update(user, case, status)
+
+ mock_send.assert_called_once()
+
+ # check call arguments
+ args = mock_send.call_args.args
+ self.assertEqual(args[0], user)
+ self.assertEqual(args[1].url, case.url)
+
+ mock_send.reset_mock()
+
+ # second call with same case/status
+ handle_status_update(user, case, status)
+
+ # no duplicate mail for this user/case/status
+ mock_send.assert_not_called()
+
+ # other user is fine
+ other_user = UserFactory.create()
+ handle_status_update(other_user, case, status)
+
+ mock_send.assert_called_once()
+
+ args = mock_send.call_args.args
+ self.assertEqual(args[0], other_user)
diff --git a/src/open_inwoner/openzaak/utils.py b/src/open_inwoner/openzaak/utils.py
index d42ea67d5..9bd3c4f83 100644
--- a/src/open_inwoner/openzaak/utils.py
+++ b/src/open_inwoner/openzaak/utils.py
@@ -12,7 +12,7 @@
from open_inwoner.openzaak.api_models import InformatieObject, Rol, Zaak, ZaakType
-from .models import OpenZaakConfig, ZaakTypeConfig
+from .models import OpenZaakConfig, ZaakTypeConfig, ZaakTypeInformatieObjectTypeConfig
logger = logging.getLogger(__name__)
@@ -191,19 +191,24 @@ def get_retrieve_resource_by_uuid_url(
def get_zaak_type_config(case_type: ZaakType) -> Optional[ZaakTypeConfig]:
try:
- if case_type.catalogus:
- return ZaakTypeConfig.objects.get(
- catalogus__url=case_type.catalogus,
- identificatie=case_type.identificatie,
- )
- else:
- return ZaakTypeConfig.objects.get(
- catalogus=None, identificatie=case_type.identificatie
- )
+ return ZaakTypeConfig.objects.filter_case_type(case_type).get()
except ZaakTypeConfig.DoesNotExist:
return None
+def get_zaak_type_info_object_type_config(
+ case_type: ZaakType,
+ info_object_type_url: str,
+) -> Optional[ZaakTypeInformatieObjectTypeConfig]:
+ assert isinstance(info_object_type_url, str)
+ try:
+ return ZaakTypeInformatieObjectTypeConfig.objects.get_for_case_and_info_type(
+ case_type, info_object_type_url
+ )
+ except ZaakTypeInformatieObjectTypeConfig.DoesNotExist:
+ return None
+
+
def format_zaak_identificatie(
identificatie: str, config: Optional[OpenZaakConfig] = None
):