#!/usr/bin/python3
# SPDX-FileCopyrightText: 2014-2025 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only
import os.path
import traceback
from collections import OrderedDict
from collections.abc import Callable, Iterator
from os import listdir
from re import Match, Pattern
from typing import Any
from univention.config_registry import ucr
from univention.lib.i18n import Translation
from univention.management.console.log import MODULE
from univention.management.console.modules import Base
from univention.management.console.modules.decorators import sanitize, simple_response
from univention.management.console.modules.diagnostic import plugins
from univention.management.console.modules.mixins import ProgressMixin
from univention.management.console.modules.sanitizers import DictSanitizer, PatternSanitizer, StringSanitizer
_ = Translation('univention-management-console-module-diagnostic').translate
[docs]
class Problem(Exception):
def __init__(self, description: str = "", **kwargs: Any) -> None:
super().__init__(description)
self.kwargs = kwargs
kwargs['type'] = self.__class__.__name__.lower()
if description:
kwargs['description'] = description
# kwargs['success'] = False # debugging ;)
[docs]
class Success(Problem):
pass
[docs]
class Conflict(Problem):
pass
[docs]
class Warning(Problem):
pass
[docs]
class Critical(Problem):
pass
[docs]
class ProblemFixed(Problem):
pass
[docs]
class Instance(Base, ProgressMixin):
PLUGIN_DIR = os.path.dirname(plugins.__file__)
[docs]
def init(self) -> None:
self.modules: dict[str, Plugin] = {}
self.load()
[docs]
@sanitize(
plugin=StringSanitizer(required=True),
args=DictSanitizer({}),
)
@simple_response(with_progress=True)
def run(self, plugin: str, args: Any = None):
plug = self.get(plugin)
MODULE.process('Running %s', plug)
for line in plug.run_descr:
MODULE.process(line)
args = args or {}
return plug.execute(self, **args)
[docs]
def new_progress(self, *args: Any, **kwargs: Any) -> Any:
progress = super().new_progress(*args, **kwargs)
progress.retry_after = 600
return progress
[docs]
@sanitize(pattern=PatternSanitizer(default='.*'))
@simple_response
def query(self, pattern: Pattern[str]) -> list[dict[str, Any]]:
return [plugin.dict for plugin in self if plugin.match(pattern)]
@property
def plugins(self) -> Iterator[str]:
for plugin in listdir(self.PLUGIN_DIR):
if (
plugin.endswith('.py')
and plugin != '__init__.py'
and not ucr.is_true(f'diagnostic/check/disable/{plugin[:-3]}')
):
yield plugin[:-3]
[docs]
def load(self) -> None:
for plugin in self.plugins:
try:
self.modules[plugin] = Plugin(plugin)
except ImportError as exc:
MODULE.error('Could not load plugin %r: %r', plugin, exc)
raise
self.modules = OrderedDict(sorted(self.modules.items(), key=lambda t: t[0]))
[docs]
def get(self, plugin: str) -> "Plugin":
return self.modules[plugin]
def __iter__(self) -> Iterator["Plugin"]:
return iter(self.modules.values())
[docs]
class Plugin:
"""
A wrapper for a Python module underneath of "univention.management.console.modules.diagnostic.plugins".
These Python modules (plugins) may have the following properties:
:attr dict actions:
A mapping of valid action names to function callbacks.
These action names can be referenced by additional displayed buttons (see :attr:`buttons`).
If a called actions does not exists the run() function is taken as fallback.
example:
actions = {
'remove': my_remove_funct,
}
:attr str title:
A short description of the problem
example:
title = _('No space left on device')
:attr str description:
A more detailed description of the problem.
The description is able to contain HTML.
The description may contain expressions which are replaced by either links to UMC modules
or links to third party websites (e.g. an SDB article).
Expressions which are replaced look like:
UMC-Modules: either {module_id:flavor} or {module_id} if no flavor exists
Links: {link_name}
See attributes :attr:`umc_modules` and :attr:`links`.
example:
description = _('There is too few space left on the device /dev/sdb1.
Please use {directory_browser} to remove unneeded files. Further information can be found at {sdb}.')
:attr list umc_modules:
A list containing dicts with the definitions of UMC modules to create links which are either displayed inline the :attr:`description`
text or underneath of it. The definition has the same signature as umc.tools.linkToModule().
example:
umc_modules = [{
'module': 'udm',
'flavor': 'navigation',
'props': {
'openObject': {
'objectDN': 'uid=Administrator,cn=users,dc=foo,dc=bar',
'objectType': 'users/user'
}
}
}]
:attr list links:
A list of dicts which define regular inline text links (e.g. to SDB articles).
They are displayed either in the :attr:`description` or underneath of it.
example:
links = [{
'name': 'sdb',
'href': 'https://sdb.univention.de/foo',
'label': _('Solve problem XYZ'),
}]
:attr list buttons:
A list of umc.widgets.Button definitions which are displayed underneath of
the description and are able to execute the actions defined in :attr:`actions`.
A callback is automatically added in the frontend.
example:
[{
'action': 'remove',
'name': 'remove',
'label': _('Remove foo')
}]
The plugin module have to define at least a :method:`run()` function.
This function is executed as the primary default action for every interaction.
Every defined action callback may raise any of the following exceptions.
A callback gets the UMC instance as a first argument.
These exceptions allow the same attributes as the module so that an action is able to overwrite
the module attributes for the execution of that specific test.
Problem
+-- Success
+-- Conflict
+-- Warning
+-- Critical
+-- ProblemFixed
"""
@property
def title(self) -> str:
"""A title for the problem"""
return getattr(self.module, 'title', '')
@property
def description(self) -> str:
"""A description of the problem and how to solve it"""
return getattr(self.module, 'description', '')
@property
def buttons(self) -> list[dict[str, str]]:
"""Buttons which are displayed e.g. to automatically solve the problem"""
return list(getattr(self.module, 'buttons', []))
@property
def run_descr(self) -> list[str]:
return list(getattr(self.module, 'run_descr', []))
@property
def umc_modules(self) -> list[dict[str, Any]]:
"""
References to UMC modules which can help solving the problem.
(module, flavor, properties)
"""
return getattr(self.module, 'umc_modules', [])
@property
def links(self) -> list[dict[str, str]]:
"""
Links to e.g. related SDB articles
(url, link_name)
"""
return getattr(self.module, 'links', [])
@property
def actions(self) -> dict[str, Callable[[Instance], dict[str, Any] | None]]:
return getattr(self.module, 'actions', {})
def __init__(self, plugin: str) -> None:
self.plugin = plugin
self.load()
[docs]
def load(self) -> None:
self.module = __import__(
'univention.management.console.modules.diagnostic.plugins.%s' % (self.plugin,),
fromlist=['univention.management.console.modules.diagnostic'],
level=0,
)
[docs]
def execute(self, umc_module: Instance, action=None, **kwargs: Any) -> dict[str, Any]:
success = True
errors: dict[str, Any] = {}
execute = self.actions.get(action, self.module.run)
try:
try:
ret = execute(umc_module, **kwargs)
if isinstance(ret, dict):
errors.update(ret)
except Problem:
raise
except Exception:
raise Problem(traceback.format_exc())
except Problem as exc:
success = False
errors.update(exc.kwargs)
result: dict[str, Any] = {
"success": success,
"type": 'success',
}
result.update(self.dict)
result.update(errors)
result.setdefault('buttons', []).insert(0, {'label': _('Test again')})
return result
[docs]
def match(self, pattern: Pattern[str]) -> Match | None:
return pattern.match(self.title) or pattern.match(self.description)
def __str__(self) -> str:
return '%s' % (self.plugin,)
@property
def dict(self) -> dict[str, Any]:
return {
"id": str(self),
"plugin": str(self),
"title": self.title,
"description": self.description,
"umc_modules": self.umc_modules,
"links": self.links,
"buttons": self.buttons,
}
[docs]
def main() -> None:
print('TODO: someday implement?')
