feat: [#2886] add BRP config check

Author: OlhaZahoruiko

src/open_inwoner/haalcentraal/admin.py

@@ -3,10 +3,14 @@
 
 from solo.admin import SingletonModelAdmin
 
+from maykin_config_checks.api.api import with_config_checks
+
+from .config_checks.fetch_brp import FetchBRPCheck
 from .models import HaalCentraalConfig
 
 
 @admin.register(HaalCentraalConfig)
+@with_config_checks(FetchBRPCheck)
 class HaalCentraalConfigAdmin(SingletonModelAdmin):
     fieldsets = (
         (
@@ -24,4 +28,11 @@ class HaalCentraalConfigAdmin(SingletonModelAdmin):
                 "fields": ("headers",),
             },
         ),
+        (
+            _("Checks"),
+            {
+                "fields": ("config_check_links",),
+            },
+        ),
     )
+    readonly_fields = ("config_check_links",)

src/open_inwoner/haalcentraal/apps.py

@@ -5,4 +5,4 @@ class HaalCentraalConfig(AppConfig):
     name = "open_inwoner.haalcentraal"
 
     def ready(self):
-        pass
+        from .config_checks import fetch_brp  # noqa

src/open_inwoner/haalcentraal/config_checks/__init__.py

@@ -0,0 +1,141 @@
+from typing import Any, Optional, TypedDict
+
+from django import forms
+from django.core.exceptions import ImproperlyConfigured
+from django.http import HttpRequest
+from django.utils.translation import gettext as _
+
+from log_outgoing_requests.models import OutgoingRequestsLogConfig
+from pydantic import ValidationError
+
+from maykin_config_checks import GenericConfigCheckResult
+from maykin_config_checks.permissions import IsSuperUser
+from maykin_config_checks.protocols import InteractiveConfigCheck
+from maykin_config_checks.registry import registry
+from open_inwoner.haalcentraal.clients import BRPClient
+from open_inwoner.haalcentraal.exceptions import BRPAPIError
+from open_inwoner.haalcentraal.models import HaalCentraalConfig
+from open_inwoner.utils.logentry import system_action, user_action
+
+
+class FetchBRPForm(forms.Form):
+    bsn = forms.CharField(max_length=9, label="BSN")
+
+    def clean_bsn(self):
+        bsn = self.cleaned_data["bsn"]
+        if not bsn.isdigit() or len(bsn) != 9:
+            raise forms.ValidationError(_("Invalid BSN format (must be 9 digits)"))
+        return bsn
+
+
+class FetchBRPCheckParams(TypedDict):
+    bsn: str
+
+
+class FetchBRPCheck(
+    InteractiveConfigCheck[
+        FetchBRPCheckParams,
+        HaalCentraalConfig,
+    ]
+):
+    identifier = "fetch_brp"
+    label = _("Fetch BRP data for BSN")
+    form_class = FetchBRPForm
+
+    required_permissions = (IsSuperUser(),)
+
+    @classmethod
+    def get_form_kwargs(
+        cls,
+        instance: Optional[HaalCentraalConfig] = None,
+    ) -> dict[str, Any]:
+        return {}
+
+    def run(
+        self,
+        data: FetchBRPCheckParams,
+        instance: Optional[HaalCentraalConfig] = None,
+        request: Optional[HttpRequest] = None,
+    ) -> GenericConfigCheckResult:
+        bsn = data["bsn"]
+        masked_bsn = f"{bsn[:3]}******"
+        log_message = f"fetch BRP check run for bsn {masked_bsn}"
+
+        if request:
+            user_action(request, request.user, log_message)
+        else:
+            system_action(log_message)
+
+        if OutgoingRequestsLogConfig.get_solo().save_body_enabled:
+            return GenericConfigCheckResult(
+                success=False,
+                identifier=self.identifier,
+                verbose_name=self.label,
+                message=_(
+                    "This check cannot be run while outgoing request log bodies "
+                    "are being saved, since it exposes personal data (BSN and "
+                    "BRP data) in the outgoing request logs. Disable 'Save "
+                    "request + response body' in the outgoing request log "
+                    "configuration before running this check."
+                ),
+                extra={},
+            )
+
+        try:
+            client = BRPClient.from_config()
+        except ImproperlyConfigured as exc:
+            return GenericConfigCheckResult(
+                success=False,
+                identifier=self.identifier,
+                verbose_name=self.label,
+                message=_("Haal Centraal is not configured"),
+                extra={"error": str(exc)},
+            )
+
+        try:
+            persoon = client.fetch_brp_data_for_bsn(bsn)
+        except ValidationError as exc:
+            return GenericConfigCheckResult(
+                success=False,
+                identifier=self.identifier,
+                verbose_name=self.label,
+                message=_("Connected, but the response data failed validation"),
+                extra={"error": str(exc), "version": client.version},
+            )
+        except BRPAPIError as exc:
+            return GenericConfigCheckResult(
+                success=False,
+                identifier=self.identifier,
+                verbose_name=self.label,
+                message=_("Failed to connect to the BRP API"),
+                extra={"error": str(exc), "type": type(exc).__name__},
+            )
+        except Exception as exc:
+            return GenericConfigCheckResult(
+                success=False,
+                identifier=self.identifier,
+                verbose_name=self.label,
+                message=_("Unexpected error while fetching BRP data"),
+                extra={"exception": str(exc), "type": type(exc).__name__},
+            )
+
+        if persoon is None:
+            return GenericConfigCheckResult(
+                success=False,
+                identifier=self.identifier,
+                verbose_name=self.label,
+                message=_("No person found for this BSN"),
+                extra={"version": client.version},
+            )
+
+        return GenericConfigCheckResult(
+            success=True,
+            identifier=self.identifier,
+            verbose_name=self.label,
+            message=_("BRP data retrieved and validated (version %(version)s)")
+            % {"version": client.version},
+            extra={"version": client.version},
+        )
+
+
+registry.register(FetchBRPCheck)

src/open_inwoner/haalcentraal/config_checks/fetch_brp.py

@@ -0,0 +1,213 @@
+from unittest.mock import MagicMock, patch
+
+from django.contrib.auth.models import Permission
+from django.test import Client, TestCase
+from django.urls import reverse
+
+from log_outgoing_requests.constants import SaveLogsChoice
+from log_outgoing_requests.models import OutgoingRequestsLogConfig
+from maykin_2fa.test import disable_admin_mfa
+from pydantic import ValidationError
+
+from open_inwoner.accounts.tests.factories import UserFactory
+from open_inwoner.haalcentraal.api_models import BRP2xPersoon
+from open_inwoner.haalcentraal.clients import BRPClient
+from open_inwoner.haalcentraal.config_checks.fetch_brp import (
+    FetchBRPCheck,
+    FetchBRPForm,
+)
+from open_inwoner.haalcentraal.exceptions import BRPAPINetworkError
+from open_inwoner.haalcentraal.models import HaalCentraalConfig
+
+
+def make_validation_error() -> ValidationError:
+    try:
+        BRP2xPersoon.model_validate(
+            {"verblijfplaats": {"verblijfadres": {"huisnummer": "not-a-number"}}}
+        )
+    except ValidationError as exc:
+        return exc
+    raise AssertionError("expected a ValidationError")  # pragma: no cover
+
+
+class FetchBRPCheckTests(TestCase):
+    def setUp(self):
+        self.check = FetchBRPCheck()
+
+    def test_refuses_to_run_while_save_body_enabled(self):
+        config = OutgoingRequestsLogConfig.get_solo()
+        config.save_body = SaveLogsChoice.yes
+        config.save()
+
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        result = self.check.run(form.cleaned_data)
+
+        self.assertFalse(result.success)
+        self.assertIn("cannot be run", result.message)
+
+    def test_invalid_bsn(self):
+        form = FetchBRPForm(data={"bsn": "123"})
+
+        self.assertFalse(form.is_valid())
+        self.assertIn("bsn", form.errors)
+        self.assertIn("Ongeldig BSN", form.errors["bsn"][0])
+
+    def test_not_configured(self):
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        result = self.check.run(form.cleaned_data)
+
+        self.assertFalse(result.success)
+        self.assertIn("not configured", result.message)
+
+    def test_network_error(self):
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        mock_client = MagicMock()
+        mock_client.fetch_brp_data_for_bsn.side_effect = BRPAPINetworkError(
+            "Connection failed"
+        )
+
+        with patch.object(BRPClient, "from_config", return_value=mock_client):
+            result = self.check.run(form.cleaned_data)
+
+        self.assertFalse(result.success)
+        self.assertIn("Failed to connect", result.message)
+
+    def test_validation_error(self):
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        mock_client = MagicMock()
+        mock_client.fetch_brp_data_for_bsn.side_effect = make_validation_error()
+
+        with patch.object(BRPClient, "from_config", return_value=mock_client):
+            result = self.check.run(form.cleaned_data)
+
+        self.assertFalse(result.success)
+        self.assertIn("failed validation", result.message)
+
+    def test_unexpected_error(self):
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        mock_client = MagicMock()
+        mock_client.fetch_brp_data_for_bsn.side_effect = RuntimeError("boom")
+
+        with patch.object(BRPClient, "from_config", return_value=mock_client):
+            result = self.check.run(form.cleaned_data)
+
+        self.assertFalse(result.success)
+        self.assertIn("Unexpected error", result.message)
+
+    def test_no_person_found(self):
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        mock_client = MagicMock()
+        mock_client.fetch_brp_data_for_bsn.return_value = None
+        mock_client.version = "2.1"
+
+        with patch.object(BRPClient, "from_config", return_value=mock_client):
+            result = self.check.run(form.cleaned_data)
+
+        self.assertFalse(result.success)
+        self.assertIn("No person found", result.message)
+
+    def test_person_found(self):
+        form = FetchBRPForm(data={"bsn": "123456789"})
+        self.assertTrue(form.is_valid())
+
+        mock_client = MagicMock()
+        mock_client.fetch_brp_data_for_bsn.return_value = BRP2xPersoon()
+        mock_client.version = "2.1"
+
+        with patch.object(BRPClient, "from_config", return_value=mock_client):
+            result = self.check.run(form.cleaned_data)
+
+        self.assertTrue(result.success)
+        self.assertIn("BRP data retrieved", result.message)
+        self.assertEqual(result.extra["version"], "2.1")
+
+
+class FetchBRPViewTests(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.superuser = UserFactory(is_superuser=True, is_staff=True)
+        self.user = UserFactory()
+        self.config = HaalCentraalConfig.get_solo()
+
+    def get_url(self):
+        return reverse(
+            "run_config_check",
+            args=["haalcentraal", "haalcentraalconfig", self.config.pk, "fetch_brp"],
+        )
+
+    def test_permission_denied_for_normal_user(self):
+        self.client.force_login(self.user)
+
+        response = self.client.get(self.get_url())
+
+        self.assertEqual(response.status_code, 403)
+
+    def test_permission_denied_for_staff_user_with_model_access(self):
+        staff_user = UserFactory(is_staff=True, is_superuser=False)
+        staff_user.user_permissions.add(
+            *Permission.objects.filter(
+                content_type__app_label="haalcentraal",
+                content_type__model="haalcentraalconfig",
+            )
+        )
+        self.client.force_login(staff_user)
+
+        response = self.client.get(self.get_url())
+
+        self.assertEqual(response.status_code, 403)
+
+    def test_superuser_can_access(self):
+        self.client.force_login(self.superuser)
+
+        response = self.client.get(self.get_url())
+
+        self.assertEqual(response.status_code, 200)
+
+    def test_post_runs_check(self):
+        self.client.force_login(self.superuser)
+
+        response = self.client.post(self.get_url(), {"bsn": "123456789"})
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "not configured")
+
+    @disable_admin_mfa()
+    def test_config_check_button_on_admin_change_page(self):
+        self.client.force_login(self.superuser)
+
+        response = self.client.get(
+            reverse(
+                "admin:haalcentraal_haalcentraalconfig_change", args=[self.config.pk]
+            )
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, self.get_url())
+
+
+class FetchBRPStandaloneTests(TestCase):
+    def setUp(self):
+        self.client = Client()
+        self.superuser = UserFactory(is_superuser=True, is_staff=True)
+
+    def test_standalone_runs_check(self):
+        self.client.force_login(self.superuser)
+
+        url = reverse("run_config_check_standalone", args=["fetch_brp"])
+
+        response = self.client.post(url, {"bsn": "123456789"})
+
+        self.assertEqual(response.status_code, 200)
+        self.assertContains(response, "not configured")