diff --git a/src/nrc/datamodel/admin.py b/src/nrc/datamodel/admin.py
index f3e37cf..6bcad83 100644
--- a/src/nrc/datamodel/admin.py
+++ b/src/nrc/datamodel/admin.py
@@ -1,12 +1,16 @@
from django.contrib import admin, messages
from django.db.models import Count, OuterRef, Q, Subquery
-from django.urls import reverse
+from django.http import HttpResponseRedirect
+from django.urls import path, reverse
from django.utils.safestring import mark_safe
from django.utils.translation import gettext_lazy as _
+from requests.exceptions import RequestException
+from rest_framework.exceptions import ValidationError
from rest_framework.fields import DateTimeField
from nrc.api.utils import send_notification
+from nrc.api.validators import CallbackURLValidator
from .admin_filters import ActionFilter, ResourceFilter, ResultFilter
from .models import (
@@ -45,11 +49,112 @@ def get_object_actions(self, obj):
)
+@admin.action(
+ description=_("Check the status of the callback URLs of selected Subscriptions")
+)
+def check_callback_url_status(modeladmin, request, queryset):
+ """
+ Make a request to the callback URLs of all selected Abonnementen and store the results
+ in the session
+
+ Any subsequent executions of this action will remove the previous results from the session
+ """
+ validator = CallbackURLValidator("callback_url", "auth")
+ callback_statuses = {}
+ for obj in queryset.iterator():
+ # Any other Abonnement that has the same callback URL and auth will be skipped
+ # and gets the same status
+ key = str((obj.callback_url, obj.auth))
+ if key in callback_statuses:
+ continue
+
+ try:
+ validator({"callback_url": obj.callback_url, "auth": obj.auth}, None)
+ callback_statuses[key] = True
+ except (ValidationError, RequestException):
+ callback_statuses[key] = False
+
+ messages.add_message(
+ request,
+ messages.SUCCESS,
+ _(
+ "Retrieve status for selected subscriptions. "
+ "All previous results have been reset."
+ ),
+ )
+ request.session["callback_statuses"] = callback_statuses
+
+
+class StatusCodeFilter(admin.SimpleListFilter):
+ title = "callback URL reachable?"
+ parameter_name = "callback_url_reachable"
+
+ def lookups(self, request, model_admin):
+ return [("true", "Yes"), ("false", "No"), ("unknown", "Unknown")]
+
+ def queryset(self, request, queryset):
+ if not self.value():
+ return queryset
+
+ filtered_ids = []
+ if callback_statuses := request.session.get("callback_statuses"):
+ for obj in queryset.iterator():
+ callback_status = callback_statuses.get(
+ str((obj.callback_url, obj.auth)), None
+ )
+ if self.value() == "true" and callback_status:
+ filtered_ids.append(obj.id)
+ elif self.value() == "false" and callback_status == False: # noqa
+ filtered_ids.append(obj.id)
+ elif self.value() == "unknown" and callback_status is None:
+ filtered_ids.append(obj.id)
+ return queryset.filter(id__in=filtered_ids)
+ return queryset
+
+
@admin.register(Abonnement)
class AbonnementAdmin(admin.ModelAdmin):
- list_display = ("uuid", "client_id", "callback_url", "get_kanalen_display")
+ list_display = (
+ "uuid",
+ "client_id",
+ "callback_url",
+ "get_callback_url_reachable",
+ "get_kanalen_display",
+ )
readonly_fields = ("uuid",)
+ list_filter = (StatusCodeFilter,)
inlines = (FilterGroupInline,)
+ actions = [check_callback_url_status]
+
+ def changelist_view(self, request, extra_context=None):
+ # Store the request object to ensure the custom admin field has access to it
+ self._request = request
+ return super().changelist_view(request, extra_context=extra_context)
+
+ def check_all_callback_urls(self, request):
+ queryset = Abonnement.objects.all()
+ check_callback_url_status(self, request, queryset)
+ self.message_user(request, _("Checked status of all callback URLs"))
+ return HttpResponseRedirect(request.META.get("HTTP_REFERER"))
+
+ def get_urls(self):
+ urls = super().get_urls()
+ custom_urls = [
+ path(
+ "check-all-callback-urls/",
+ self.admin_site.admin_view(self.check_all_callback_urls),
+ name="check_all_callback_urls",
+ ),
+ ]
+ return custom_urls + urls
+
+ @admin.display(description=_("callback URL reachable?"))
+ def get_callback_url_reachable(self, obj):
+ if callback_statuses := self._request.session.get("callback_statuses"):
+ return callback_statuses.get(str((obj.callback_url, obj.auth)), None)
+ return None
+
+ get_callback_url_reachable.boolean = True
@admin.display(description=_("kanalen"))
def get_kanalen_display(self, obj):
diff --git a/src/nrc/fixtures/default_admin_index.json b/src/nrc/fixtures/default_admin_index.json
index ae61171..5f44c14 100644
--- a/src/nrc/fixtures/default_admin_index.json
+++ b/src/nrc/fixtures/default_admin_index.json
@@ -95,6 +95,10 @@
[
"notifications_api_common",
"subscription"
+ ],
+ [
+ "log_outgoing_requests",
+ "outgoingrequestslogconfig"
]
]
}
@@ -113,6 +117,10 @@
[
"axes",
"accesslog"
+ ],
+ [
+ "log_outgoing_requests",
+ "outgoingrequestslog"
]
]
}
diff --git a/src/nrc/templates/admin/datamodel/abonnement/change_list.html b/src/nrc/templates/admin/datamodel/abonnement/change_list.html
new file mode 100644
index 0000000..ea31043
--- /dev/null
+++ b/src/nrc/templates/admin/datamodel/abonnement/change_list.html
@@ -0,0 +1,9 @@
+{% extends "admin/change_list.html" %}
+{% load i18n %}
+
+{% block object-tools-items %}
+
+ {% trans "Check status of callback URLs" %}
+
+ {{ block.super }}
+{% endblock %}
diff --git a/src/nrc/tests/admin/test_abonnement.py b/src/nrc/tests/admin/test_abonnement.py
index b99c8cd..7568f85 100644
--- a/src/nrc/tests/admin/test_abonnement.py
+++ b/src/nrc/tests/admin/test_abonnement.py
@@ -1,11 +1,13 @@
-from django.test import override_settings
-from django.urls import reverse
+from django.test import override_settings, tag
+from django.urls import reverse, reverse_lazy
+import requests_mock
from django_webtest import WebTest
from freezegun import freeze_time
from maykin_2fa.test import disable_admin_mfa
+from requests.exceptions import RequestException
-from nrc.accounts.tests.factories import SuperUserFactory
+from nrc.accounts.tests.factories import SuperUserFactory, UserFactory
from nrc.datamodel.tests.factories import (
AbonnementFactory,
FilterGroupFactory,
@@ -21,6 +23,8 @@
)
class AbonnementAdminWebTest(WebTest):
maxdiff = None
+ changelist_url = reverse_lazy("admin:datamodel_abonnement_changelist")
+ action_url = reverse_lazy("admin:check_all_callback_urls")
def setUp(self):
self.user = SuperUserFactory.create()
@@ -31,6 +35,7 @@ def setUp(self):
100, abonnement=self.abonnement, notificatie__kanaal=self.kanaal
)
+ @tag("gh-157")
def test_delete_abonnement_hide_notificatie_responses(self):
"""
Regression test for https://github.com/open-zaak/open-notificaties/issues/157
@@ -50,14 +55,12 @@ def test_delete_abonnement_hide_notificatie_responses(self):
self.assertNotIn("Notificatie response", deleted_objects)
+ @tag("gh-157")
def test_bulk_delete_abonnement_hide_notificatie_responses(self):
"""
Regression test for https://github.com/open-zaak/open-notificaties/issues/157
"""
- response = self.app.get(
- reverse("admin:datamodel_abonnement_changelist"),
- user=self.user,
- )
+ response = self.app.get(self.changelist_url, user=self.user)
form = response.forms["changelist-form"]
form["action"] = "delete_selected"
@@ -75,3 +78,279 @@ def test_bulk_delete_abonnement_hide_notificatie_responses(self):
)
self.assertNotIn("Notificatie response", deleted_objects)
+
+ @tag("gh-108")
+ @requests_mock.Mocker()
+ def test_abonnement_list_check_if_callback_urls_are_reachable(self, m):
+ """
+ Test that the callback URL statuses can be checked by using the admin action
+ """
+ self.abonnement.delete()
+
+ abonnement_url_reachable = AbonnementFactory.create(
+ callback_url="http://reachable.local/foo",
+ auth="Token 1234",
+ )
+ abonnement_url_unreachable = AbonnementFactory.create(
+ callback_url="http://unreachable.local/foo",
+ auth="Token 1234",
+ )
+ abonnement_url_unreachable_duplicate = AbonnementFactory.create(
+ callback_url="http://unreachable.local/foo",
+ auth="Token 1234",
+ )
+ abonnement_url_incorrect_auth = AbonnementFactory.create(
+ callback_url="http://incorrect.auth.local/foo",
+ auth="Token 4321",
+ )
+
+ m.post("http://reachable.local/foo", status_code=204)
+ m.post("http://unreachable.local/foo", status_code=403)
+ m.post("http://incorrect.auth.local/foo", exc=RequestException)
+
+ response = self.app.get(self.changelist_url, user=self.user)
+
+ row_incorrect_auth, row_unreachable, row_unreachable2, row_reachable = (
+ response.html.find("tbody").find_all("tr")
+ )
+
+ with self.subTest("initial callback statuses are unknown"):
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[2].text,
+ abonnement_url_incorrect_auth.callback_url,
+ )
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ self.assertEqual(
+ row_unreachable.find_all("td")[2].text,
+ abonnement_url_unreachable.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ self.assertEqual(
+ row_unreachable2.find_all("td")[2].text,
+ abonnement_url_unreachable_duplicate.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable2.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ self.assertEqual(
+ row_reachable.find_all("td")[2].text,
+ abonnement_url_reachable.callback_url,
+ )
+ self.assertEqual(
+ row_reachable.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ form = response.forms[0]
+ form["action"] = "check_callback_url_status"
+ form["_selected_action"] = [
+ abonnement_url_reachable.pk,
+ abonnement_url_unreachable.pk,
+ abonnement_url_incorrect_auth.pk,
+ ]
+
+ response = form.submit().follow()
+
+ row_incorrect_auth, row_unreachable, row_unreachable2, row_reachable = (
+ response.html.find("tbody").find_all("tr")
+ )
+ with self.subTest("callback statuses are known"):
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[2].text,
+ abonnement_url_incorrect_auth.callback_url,
+ )
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ self.assertEqual(
+ row_unreachable.find_all("td")[2].text,
+ abonnement_url_unreachable.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ # This row is marked as false despite not being selected for the action,
+ # because it has the same callback URL and auth as another selected Subscription
+ self.assertEqual(
+ row_unreachable2.find_all("td")[2].text,
+ abonnement_url_unreachable_duplicate.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable2.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ self.assertEqual(
+ row_reachable.find_all("td")[2].text,
+ abonnement_url_reachable.callback_url,
+ )
+ self.assertEqual(
+ row_reachable.find_all("td")[3].find("img").attrs["alt"], "True"
+ )
+
+ filtered_response = self.app.get(
+ f"{self.changelist_url}?callback_url_reachable=false", user=self.user
+ )
+
+ row_incorrect_auth, row_unreachable, row_unreachable2 = (
+ filtered_response.html.find("tbody").find_all("tr")
+ )
+ with self.subTest("callback statuses are known"):
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[2].text,
+ abonnement_url_incorrect_auth.callback_url,
+ )
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ self.assertEqual(
+ row_unreachable.find_all("td")[2].text,
+ abonnement_url_unreachable.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ self.assertEqual(
+ row_unreachable2.find_all("td")[2].text,
+ abonnement_url_unreachable_duplicate.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable2.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ @tag("gh-108")
+ @requests_mock.Mocker()
+ def test_abonnement_list_check_if_callback_urls_are_reachable_with_custom_button(
+ self, m
+ ):
+ """
+ Test that the callback URL statuses can be checked by using the button that
+ triggers the admin action with all Abonnementen
+ """
+ self.abonnement.delete()
+
+ abonnement_url_reachable = AbonnementFactory.create(
+ callback_url="http://reachable.local/foo",
+ auth="Token 1234",
+ )
+ abonnement_url_unreachable = AbonnementFactory.create(
+ callback_url="http://unreachable.local/foo",
+ auth="Token 1234",
+ )
+ abonnement_url_unreachable_duplicate = AbonnementFactory.create(
+ callback_url="http://unreachable.local/foo",
+ auth="Token 1234",
+ )
+ abonnement_url_incorrect_auth = AbonnementFactory.create(
+ callback_url="http://incorrect.auth.local/foo",
+ auth="Token 4321",
+ )
+
+ m.post("http://reachable.local/foo", status_code=204)
+ m.post("http://unreachable.local/foo", status_code=403)
+ m.post("http://incorrect.auth.local/foo", exc=RequestException)
+
+ response = self.app.get(self.changelist_url, user=self.user)
+
+ row_incorrect_auth, row_unreachable, row_unreachable2, row_reachable = (
+ response.html.find("tbody").find_all("tr")
+ )
+
+ with self.subTest("initial callback statuses are unknown"):
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[2].text,
+ abonnement_url_incorrect_auth.callback_url,
+ )
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ self.assertEqual(
+ row_unreachable.find_all("td")[2].text,
+ abonnement_url_unreachable.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ self.assertEqual(
+ row_unreachable2.find_all("td")[2].text,
+ abonnement_url_unreachable_duplicate.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable2.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ self.assertEqual(
+ row_reachable.find_all("td")[2].text,
+ abonnement_url_reachable.callback_url,
+ )
+ self.assertEqual(
+ row_reachable.find_all("td")[3].find("img").attrs["alt"], "None"
+ )
+
+ response = self.app.get(
+ self.action_url,
+ user=self.user,
+ headers={"Referer": str(self.changelist_url)},
+ ).follow()
+
+ row_incorrect_auth, row_unreachable, row_unreachable2, row_reachable = (
+ response.html.find("tbody").find_all("tr")
+ )
+ with self.subTest("callback statuses are known"):
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[2].text,
+ abonnement_url_incorrect_auth.callback_url,
+ )
+ self.assertEqual(
+ row_incorrect_auth.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ self.assertEqual(
+ row_unreachable.find_all("td")[2].text,
+ abonnement_url_unreachable.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ # This row is marked as false despite not being selected for the action,
+ # because it has the same callback URL and auth as another selected Subscription
+ self.assertEqual(
+ row_unreachable2.find_all("td")[2].text,
+ abonnement_url_unreachable_duplicate.callback_url,
+ )
+ self.assertEqual(
+ row_unreachable2.find_all("td")[3].find("img").attrs["alt"], "False"
+ )
+
+ self.assertEqual(
+ row_reachable.find_all("td")[2].text,
+ abonnement_url_reachable.callback_url,
+ )
+ self.assertEqual(
+ row_reachable.find_all("td")[3].find("img").attrs["alt"], "True"
+ )
+
+ def test_check_all_callback_urls_not_allowed_for_non_staff_user(self):
+ non_staff_user = UserFactory(is_staff=False)
+ response = self.app.get(
+ self.action_url,
+ user=non_staff_user,
+ headers={"Referer": str(self.changelist_url)},
+ )
+
+ login_url = reverse("admin:login")
+
+ self.assertEqual(response.status_code, 302)
+ self.assertEqual(response.location, f"{login_url}?next={self.action_url}")