Source code for univention.testing.ucr

# SPDX-FileCopyrightText: 2013-2025 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only

"""
BETA VERSION

Wrapper around Univention Configuration Registry that is able to revert
the UCR status after usage. For usage examples look at the end of this
file.

.. warning::
    changes to the ConfigRegistry object will also trigger the evaluation of templates
    and therefore changes in configuration files created by UCR!

.. warning:: The API is currently under development and may change before next UCS release!
"""


import copy
from types import TracebackType
from typing import Any, Self

import univention.config_registry
from univention.config_registry import ConfigRegistry



[docs]
class UCSTestConfigRegistry(ConfigRegistry):
    """
    Extension to ConfigRegistry to be able to clean up after
    several changes to UCR variables have been done.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """initialise object"""
        ConfigRegistry.__init__(self, *args, **kwargs)
        self.__original_registry: dict[int, dict[str, str]] | None = None


[docs]
    def ucr_update(self, *args):
        return univention.config_registry.frontend.ucr_update(*args)



[docs]
    def handler_set(self, *args):
        return univention.config_registry.handler_set(*args)



[docs]
    def handler_unset(self, *args):
        return univention.config_registry.handler_unset(*args)



[docs]
    def load(self) -> None:
        """call load() of superclass and save original registry values"""
        ConfigRegistry.load(self)
        if self.__original_registry is None:
            self.__original_registry = {
                regtype: copy.deepcopy(dict(reg))
                for (regtype, reg) in self._walk()
            }



[docs]
    def revert_to_original_registry(self) -> None:
        """revert UCR values back to original state"""
        # load current values again to perform correct comparison
        self.load()
        assert self.__original_registry is not None
        for regtype, option in (
                (ConfigRegistry.NORMAL, ''),
                (ConfigRegistry.LDAP, 'ldap-policy'),
                (ConfigRegistry.FORCED, 'force'),
                (ConfigRegistry.SCHEDULE, 'schedule'),
        ):
            # remove new variables
            keylist = set(self._registry[regtype]) - set(self.__original_registry[regtype])
            if keylist:
                self.handler_unset(list(keylist), {option: True})

            # add/revert existing variables
            changes = [
                f'{key}={origval}'
                for key, origval in self.__original_registry[regtype].items()
                if origval != self._registry[regtype].get(key)
            ]
            if changes:
                self.handler_set(changes, {option: True})
        # load new/original values
        self.load()


    def __enter__(self) -> Self:
        self.load()
        return self

    def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
        self.revert_to_original_registry()



if __name__ == '__main__':
    import time

    # Usage variant 1 "the manual way"

    print('Loading UCR...')
    ucr = UCSTestConfigRegistry()
    ucr.load()
    print('Setting some variables...')
    ucr.handler_set(['foo/bar=ding/dong'])
    ucr.handler_set(['repository/online/server=ftp.debian.org'])
    ucr.handler_unset(['server/role'])
    print('Waiting for 3 seconds...')
    time.sleep(3)
    print('Cleanup...')
    ucr.revert_to_original_registry()

    # Usage variant 2 "with statement"

    with UCSTestConfigRegistry() as ucr2:
        print('Old values...')
        print(ucr2.get('foo/bar', '<unset>'))
        print(ucr2.get('repository/online/server', '<unset>'))
        print(ucr2.get('server/role', '<unset>'))
        print('Setting some variables...')
        ucr2.handler_set(['foo/bar=ding/dong'])
        ucr2.handler_set(['repository/online/server=ftp.debian.org'])
        ucr2.handler_unset(['server/role'])
        print('Waiting for 3 seconds...')
        time.sleep(3)
        print('Cleanup...')
Univention Corporate Server Python API 5.2 documentation

Source code for univention.testing.ucr
