From 60d0608968f1d155ac26ad5199b917f8f5dd066e Mon Sep 17 00:00:00 2001 From: vasileios Date: Wed, 8 May 2024 16:42:04 +0200 Subject: [PATCH] Moved configuration check into its own class --- src/openforms/config/checks.py | 134 +++++++++++++++++++++++++++++++++ src/openforms/config/views.py | 118 ++--------------------------- 2 files changed, 139 insertions(+), 113 deletions(-) create mode 100644 src/openforms/config/checks.py diff --git a/src/openforms/config/checks.py b/src/openforms/config/checks.py new file mode 100644 index 0000000000..dfd4bc8d35 --- /dev/null +++ b/src/openforms/config/checks.py @@ -0,0 +1,134 @@ +from typing import Any, Generator, Protocol, TypeGuard + +from django.utils.encoding import force_str +from django.utils.translation import gettext as _ + +from openforms.appointments.registry import register as appointments_register +from openforms.contrib.brk.checks import BRKValidatorCheck +from openforms.contrib.kadaster.config_check import BAGCheck, LocatieServerCheck +from openforms.contrib.kvk.checks import KVKRemoteValidatorCheck +from openforms.dmn.registry import register as dmn_register +from openforms.payments.registry import register as payments_register +from openforms.plugins.plugin import AbstractBasePlugin +from openforms.prefill.registry import register as prefill_register +from openforms.registrations.registry import register as registrations_register + +from .data import Action, Entry + + +class ConfigCheckable(Protocol): + verbose_name: str + + def check_config(self) -> None: ... + + def get_config_actions(self) -> list[Action]: ... + + +def _subset_match(requested: str | None, checking: str) -> bool: + if not requested: + return True + return requested == checking + + +def is_plugin(plugin: Any) -> TypeGuard[AbstractBasePlugin]: + if hasattr(plugin, "identifier"): + return True + return False + + +class ConfigurationCheck: + def __init__(self, requested_plugin: AbstractBasePlugin = None) -> None: + self.requested_plugin = requested_plugin + + def get_configuration_results( + self, module: str | None = None + ) -> list[dict[str, Any]]: + sections = [] + + # add custom non-generic + if _subset_match(module, "address_lookup"): + sections += [ + { + "name": _("Address lookup plugins"), + "entries": [ + self.get_plugin_entry( + BAGCheck, + ), # Location client + self.get_plugin_entry(LocatieServerCheck), # Kadaster search + ], + }, + ] + + if _subset_match(module, "validators"): + sections += [ + { + "name": _("Validator plugins"), + "entries": [ + # uses KVK 'zoeken' client + self.get_plugin_entry(KVKRemoteValidatorCheck), + self.get_plugin_entry(BRKValidatorCheck), + ], + }, + ] + + # Iterate over all plugin registries. + plugin_registries = [ + ("appointments", _("Appointment plugins"), appointments_register), + ("registrations", _("Registration plugins"), registrations_register), + ("prefill", _("Prefill plugins"), prefill_register), + ("payments", _("Payment plugins"), payments_register), + ("dmn", _("DMN plugins"), dmn_register), + ] + + for registry_module, name, register in plugin_registries: + if not _subset_match(module, registry_module): + continue + sections.append( + { + "name": name, + "entries": list(self.get_register_entries(register)), + } + ) + + return sections + + def get_register_entries(self, register) -> Generator[Entry, None, None]: + for plugin in register.iter_enabled_plugins(): + if hasattr(plugin, "iter_config_checks"): + yield from plugin.iter_config_checks() + else: + yield self.get_plugin_entry(plugin) + + def get_plugin_entry(self, plugin: AbstractBasePlugin | ConfigCheckable) -> Entry: + # undocumented query string support - helps for developers ;) + status, error = True, "" + if is_plugin(plugin) and not _subset_match( + self.requested_plugin, plugin.identifier + ): + return Entry( + name=force_str(plugin.verbose_name), + status=None, + actions=[], + ) + + try: + plugin.check_config() + except Exception as e: + status, error = False, str(e) + + try: + actions = plugin.get_config_actions() + except Exception as e: + actions = [ + ( + _("Internal error: {exception}").format(exception=e), + "", + ) + ] + + return Entry( + name=force_str(plugin.verbose_name), + status=status, + actions=actions, + error=error, + ) diff --git a/src/openforms/config/views.py b/src/openforms/config/views.py index 341eaec3fd..baa2498a31 100644 --- a/src/openforms/config/views.py +++ b/src/openforms/config/views.py @@ -1,47 +1,18 @@ -from typing import Any, Generator, Protocol, TypeGuard +from typing import Any from django.contrib.auth.mixins import PermissionRequiredMixin from django.urls import reverse -from django.utils.encoding import force_str from django.utils.translation import gettext as _ from django.views.generic import TemplateView -from openforms.appointments.registry import register as appointments_register -from openforms.contrib.brk.checks import BRKValidatorCheck -from openforms.contrib.kadaster.config_check import BAGCheck, LocatieServerCheck -from openforms.contrib.kvk.checks import KVKRemoteValidatorCheck -from openforms.dmn.registry import register as dmn_register -from openforms.payments.registry import register as payments_register -from openforms.plugins.plugin import AbstractBasePlugin -from openforms.prefill.registry import register as prefill_register -from openforms.registrations.registry import register as registrations_register from openforms.utils.mixins import UserIsStaffMixin -from .data import Action, Entry +from .checks import ConfigurationCheck +from .data import Entry from .models import GlobalConfiguration from .utils import verify_clamav_connection -def _subset_match(requested: str | None, checking: str) -> bool: - if not requested: - return True - return requested == checking - - -class ConfigCheckable(Protocol): - verbose_name: str - - def check_config(self) -> None: ... - - def get_config_actions(self) -> list[Action]: ... - - -def is_plugin(plugin: Any) -> TypeGuard[AbstractBasePlugin]: - if hasattr(plugin, "identifier"): - return True - return False - - class ConfigurationView(UserIsStaffMixin, PermissionRequiredMixin, TemplateView): template_name = "admin/config/overview.html" permission_required = [ @@ -55,48 +26,9 @@ def get_context_data(self, **kwargs: Any) -> dict[str, Any]: # undocumented query string support - helps for developers ;) module = self.request.GET.get("module") - # add custom non-generic - if _subset_match(module, "address_lookup"): - sections += [ - { - "name": _("Address lookup plugins"), - "entries": [ - self.get_plugin_entry(BAGCheck), # Location client - self.get_plugin_entry(LocatieServerCheck), # Kadaster search - ], - }, - ] - - if _subset_match(module, "validators"): - sections += [ - { - "name": _("Validator plugins"), - "entries": [ - # uses KVK 'zoeken' client - self.get_plugin_entry(KVKRemoteValidatorCheck), - self.get_plugin_entry(BRKValidatorCheck), - ], - }, - ] + check = ConfigurationCheck(requested_plugin=self.request.GET.get("plugin")) - # Iterate over all plugin registries. - plugin_registries = [ - ("appointments", _("Appointment plugins"), appointments_register), - ("registrations", _("Registration plugins"), registrations_register), - ("prefill", _("Prefill plugins"), prefill_register), - ("payments", _("Payment plugins"), payments_register), - ("dmn", _("DMN plugins"), dmn_register), - ] - - for registry_module, name, register in plugin_registries: - if not _subset_match(module, registry_module): - continue - sections.append( - { - "name": name, - "entries": list(self.get_register_entries(register)), - } - ) + sections += check.get_configuration_results(module) sections.append({"name": "Anti-virus", "entries": [self.get_clamav_entry()]}) @@ -104,46 +36,6 @@ def get_context_data(self, **kwargs: Any) -> dict[str, Any]: return context - def get_register_entries(self, register) -> Generator[Entry, None, None]: - for plugin in register.iter_enabled_plugins(): - if hasattr(plugin, "iter_config_checks"): - yield from plugin.iter_config_checks() - else: - yield self.get_plugin_entry(plugin) - - def get_plugin_entry(self, plugin: AbstractBasePlugin | ConfigCheckable) -> Entry: - # undocumented query string support - helps for developers ;) - requested_plugin = self.request.GET.get("plugin") - status, error = True, "" - if is_plugin(plugin) and not _subset_match(requested_plugin, plugin.identifier): - return Entry( - name=force_str(plugin.verbose_name), - status=None, - actions=[], - ) - - try: - plugin.check_config() - except Exception as e: - status, error = False, str(e) - - try: - actions = plugin.get_config_actions() - except Exception as e: - actions = [ - ( - _("Internal error: {exception}").format(exception=e), - "", - ) - ] - - return Entry( - name=force_str(plugin.verbose_name), - status=status, - actions=actions, - error=error, - ) - def get_clamav_entry(self): config = GlobalConfiguration.get_solo() assert isinstance(config, GlobalConfiguration)