Skip to content

Commit

Permalink
Merge pull request #4279 from open-formulieren/improvement/move-confi…
Browse files Browse the repository at this point in the history
…guration-check-into-its-own-class

Move configuration check into its own class
  • Loading branch information
sergei-maertens authored May 8, 2024
2 parents 1df0654 + 60d0608 commit ca094b7
Show file tree
Hide file tree
Showing 2 changed files with 139 additions and 113 deletions.
134 changes: 134 additions & 0 deletions src/openforms/config/checks.py
Original file line number Diff line number Diff line change
@@ -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,
)
118 changes: 5 additions & 113 deletions src/openforms/config/views.py
Original file line number Diff line number Diff line change
@@ -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 = [
Expand All @@ -55,95 +26,16 @@ 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()]})

context.update({"sections": sections})

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)
Expand Down

0 comments on commit ca094b7

Please sign in to comment.