Clean up, remove compat with py < 3.6

[pyi2ncommon] / src / cnfvar / store.py

# The software in this package is distributed under the GNU General
# Public License version 2 (with a special exception described below).
#
# A copy of GNU General Public License (GPL) is included in this distribution,
# in the file COPYING.GPL.
#
# As a special exception, if other files instantiate templates or use macros
# or inline functions from this file, or you compile this file and link it
# with other works to produce a work based on this file, this file
# does not by itself cause the resulting work to be covered
# by the GNU General Public License.
#
# However the source code for this file must still be made available
# in accordance with section (3) of the GNU General Public License.
#
# This exception does not invalidate any other reasons why a work based
# on this file might be covered by the GNU General Public License.
#
# Copyright (c) 2016-2022 Intra2net AG <info@intra2net.com>

"""
store: implementations of CNF stores using varlink and `*et_cnf` binaries.

Featuring:
    - CnfStore: the main store class that is implemented using the varlink API
    - BinaryCnfStore: alternative store class implementation using the `set_cnf`
      and `get_cnf` binaries

.. seealso:: Overview Diagram linked to from doc main page

.. codeauthor:: Intra2net
"""
import subprocess
import logging
import time
import html

from .. import arnied_wrapper, arnied_api
from . import CnfBinary, Cnf, CnfList

log = logging.getLogger("pyi2ncommon.cnfvar.store")


class CnfStore:
    """Main CNF store class that uses the varlink API."""

    def __init__(self, backend_driver=arnied_api.Arnied):
        """
        Initialize this class.

        :param backend_driver: driver to use to talk to the cnfvar backend
        :type: :py:class:`arnied_api.Arnied`
        """
        self._driver = backend_driver
        log.debug("Initialized CnfStore with driver `%s`", type(backend_driver))

    def query(self, name=None, instance=None):
        """
        Query the CNF store and return a list of parsed CNFs.

        :param str name: optional name of the CNFs to query
        :param instance: optional CNF instance
        :type instance: str or int
        :returns: list of parsed CNF values
        :rtype: :py:class:`CnfList`

        Example::
            store = CnfStore()
            user_cnfs = store.query("USER")
        """
        log.debug("Querying CnfStore with name=%s and instance=%s via arnied API",
                  name, instance)

        # the API only expects an object if there is a name (and it's not case-insensitive)
        query = arnied_api.GetCnfQuery(name.upper(), instance) if name else None
        api_ret = self._driver.get_cnf(query)

        # NOTE: logging all output here would result in huge lines when querying
        # all variables via `store.query()`
        log.debug("Arnied API returned %d cnfvars", len(api_ret.vars))
        return CnfList.from_api_structure(api_ret.vars)

    def commit(self, cnf, fix_problems=False):
        """
        Update or insert CNF variables from a list.

        :param cnf: CNF instance or list of CNFs to update or insert
        :type cnf: :py:class:`Cnf` or :py:class:`CnfList`
        :raises: :py:class:`CommitException` if the arnied API complains
        :param bool fix_problems: whether to automatically fix errors in the vars

        .. note:: you can mix variables to insert and variables to update
                  in the same list as the system should handle it nicely

        Example::
            store = CnfStore()
            user_cnf = store.query("USER")\
                            .single_with_value("joe")
            user_cnf.add_child("user_group_member_ref", "3")
            store.commit(user_cnf)
        """
        cnf = self._cnf_or_list(cnf, operation="commit")
        self._autofix_instances(cnf)
        log.debug("Committing variables via arnied API:\n%s", cnf)

        arnied_cnfs = cnf.to_api_structure()
        self._do_commit(cnf, arnied_cnfs, fix_problems=fix_problems)

    def delete(self, cnf, fix_problems=False):
        """
        Delete a list of top-level CNF variables.

        :param cnf: a single CNF value or a list of values
        :type cnf: :py:class:`CnfList` or :py:class:`Cnf`
        :param bool fix_problems: whether to automatically fix errors in the vars

        Example::
            store = CnfStore()
            user_cnfs = store.query("USER")\
                             .where(lambda c: c.value in ["joe", "jane"])
            store.delete(user_cnfs)
        """
        cnf = self._cnf_or_list(cnf, operation="delete")
        if any((c.parent is not None for c in cnf)):
            raise RuntimeError("Calling delete is only supported on top-level CNF variables")
        log.debug("Deleting variables via arnied API:\n%s", cnf)

        arnied_cnfs = cnf.to_api_structure()
        for c in arnied_cnfs:
            c.deleted = True
            c.children.clear()
        self._do_commit(cnf, arnied_cnfs, fix_problems=fix_problems)

    def _autofix_instances(self, cnfs):
        """
        Auto-assign a valid instance value to all top-level vars in a list.

        :param cnfs: list of cnfvars to fix
        :type cnfs: :py:class`CnfList`

        When the instance of a cnfvar is -1, the backend initializes it
        automatically. However, it starts on 1, whereas many variables are not
        allowed to start on one (those that are meant to be unique, f.e.).
        This method can be used in child classes to use an alternative scheme,
        however for performance reasons the base API class uses the default and
        relies on the cnfvar backend to do this job.
        """

    def _do_commit(self, original_cnfs, arnied_cnfs, fix_problems=False):
        """
        Set cnfvars and commit changes.

        :param original_cnfs: list of CNFs to print in case of errors
        :type original_cnfs: :py:class:`CnfList`
        :param arnied_cnfs: list of cnfvars to pass to the arnied API
        :type arnied_cnfs: [:py:class:`arnied_api.CnfVar`]
        :param bool fix_problems: whether to automatically fix errors in the vars
        :raises: :py:class:`CommitException` if the arnied API complains
        """
        try:
            ret = self._driver.set_commit_cnf(vars=arnied_cnfs, username=None,
                                              nogenerate=False,
                                              fix_commit=fix_problems)
        except arnied_api.CnfCommitError as ex:
            # fatal errors, we will handle it in our custom exception
            ret = ex

        errors = []
        for r in ret.results:
            # message can contain HTML escape codes
            msg = html.unescape(r.result_msg)

            # `result_type` is defined as int in the varlink API,
            # but the meaning of the codes are:
            # enum result_type { OK=0, WARN=1, FAIL_TEMP=2, FAIL=3 }
            if r.result_type == 1:
                log.debug("Warning in `` %s,%s: \"%s\" ``: {msg} (code=%s)",
                          r.name, r.instance, r.data, r.result_type)
            elif r.result_type > 1:
                errors.append(f"Error in `` {r.name},{r.instance}: \"{r.data}\" ``"
                              f": {msg}")

        if len(errors) > 0:
            log.debug("Error sending variables:\n%s", arnied_cnfs)
            raise CommitException(original_cnfs, "\n".join(errors))

        self._wait_for_generate()

    def _wait_for_generate(self, timeout=300):
        """
        Wait for the 'generate' program to end.

        :param int timeout: program run timeout
        :raises: :py:class:`TimeoutError` if the program did not finish on time
        """
        def scheduled_or_running(progname):
            ret = self._driver.is_scheduled_or_running(progname)
            return ret.status in [arnied_api.ProgramStatus.Scheduled,
                                  arnied_api.ProgramStatus.Running]

        def wait_for_program(progname):
            log.debug("Waiting for `%s` to be running", progname)
            for _ in range(10):
                if scheduled_or_running(progname):
                    # if is running or scheduled, break to wait for completion
                    break
                time.sleep(1)
            else:
                # after trying and retrying, program is not scheduled nor
                # running, so it is safe to assume it has already executed
                return

            # program running or scheduled, wait
            log.debug("Waiting for `%s` to finish", progname)
            for _ in range(0, timeout):
                if not scheduled_or_running(progname):
                    # finished executing, bail out
                    return
                time.sleep(1)
            raise TimeoutError(f"Program `{progname}` did not end in time")

        wait_for_program("GENERATE")
        wait_for_program("GENERATE_OFFLINE")

    def _cnf_or_list(self, cnf, operation) -> CnfList:
        """
        Validate and wrap a CNF value into a list.

        :param cnf: a single CNF value or a list of values
        :type cnf: :py:class:`CnfList` or :py:class:`Cnf`
        :param str operation: name of the operation that is being done
        :raises: :py:class:`TypeError` if the type of the CNF object
                 is neither a list nor a CNF value
        :returns: wrapped CNF value
        :rtype: :py:class:`CnfList`
        """
        if isinstance(cnf, Cnf):
            cnf = CnfList([cnf])
        elif not isinstance(cnf, CnfList):
            raise TypeError(f"Cannot {operation} value(s) of type `{type(cnf)}`")
        return cnf


class BinaryCnfStore(CnfStore):
    """Implementation of the CNF store that uses `get_cnf` and `set_cnf`."""

    #: how much to wait for arnied to report running
    ARNIED_TIMEOUT = 30

    def __init__(self, backend_driver=CnfBinary):
        """
        Initialize this class.

        :param backend_driver: driver to use to talk to the cnfvar backend
        :type: :py:class:`CnfBinary`
        """
        super().__init__(backend_driver=backend_driver)
        log.debug("Initialized BinaryCnfStore with driver `%s`", type(backend_driver))

    def query(self, name=None, instance=None):
        """
        Query the CNF store and return a list of parsed CNFs.

        :param str name: optional name of the CNFs to query
        :param instance: optional CNF instance
        :type instance: str or int
        :returns: list of parsed CNF values
        :rtype: :py:class:`CnfList`

        Example::
            store = CnfStore()
            user_cnfs = store.query("USER")
        """
        log.debug("Querying BinaryCnfStore with name=%s and instance=%s",
                  name, instance)
        output = self._driver.get_cnf(name, instance=instance)

        if len(output) == 0:
            # otherwise cnfvar raises a generic Malformed exception
            return CnfList()

        return CnfList.from_cnf_string(output)

    def commit(self, cnf, fix_problems=False):
        """
        Update or insert CNF variables from a list.

        :param cnf: CNF instance or list of CNFs to update or insert
        :type cnf: :py:class:`Cnf` or :py:class:`CnfList`
        :param bool fix_problems: whether to automatically fix errors in the vars

        .. note:: you can mix variables to insert and variables to update
                  in the same list as the system should handle it nicely

        Example::
            store = CnfStore()
            user_cnf = store.query("USER")\
                            .single_with_value("joe")
            user_cnf.add_child("user_group_member_ref", "3")
            store.commit(user_cnf)
        """
        cnf = self._cnf_or_list(cnf, operation="commit")
        self._autofix_instances(cnf)

        # set_cnf is demanding on lineno's
        cnf.renumber()
        log.debug("Committing variables via binaries:\n%s", cnf)

        self._call_arnied(arnied_wrapper.verify_running, timeout=self.ARNIED_TIMEOUT)
        try:
            self._driver.set_cnf(input_str=str(cnf), fix_problems=fix_problems)
        except subprocess.CalledProcessError as ex:
            raise CommitException(cnf, ex.stderr) from None
        self._call_arnied(arnied_wrapper.wait_for_generate)

    def delete(self, cnf, fix_problems=False):
        """
        Delete a list of top-level CNF variables.

        :param cnf: a single CNF value or a list of values
        :type cnf: :py:class:`CnfList` or :py:class:`Cnf`
        :param bool fix_problems: whether to automatically fix errors in the vars

        Example::
            store = CnfStore()
            user_cnfs = store.query("USER")\
                             .where(lambda c: c.value in ["joe", "jane"])
            store.delete(user_cnfs)
        """
        cnf = self._cnf_or_list(cnf, operation="delete")
        if any((c.parent is not None for c in cnf)):
            raise RuntimeError("Calling delete is only supported on top-level CNF variables")

        # set_cnf is demanding on lineno's
        cnf.renumber()
        log.debug("Deleting variables via binaries:\n%s", cnf)

        self._call_arnied(arnied_wrapper.verify_running, timeout=self.ARNIED_TIMEOUT)
        try:
            self._driver.set_cnf(input_str=str(cnf), delete=True, fix_problems=fix_problems)
        except subprocess.CalledProcessError as ex:
            raise CommitException(cnf, ex.stderr) from None
        self._call_arnied(arnied_wrapper.wait_for_generate)

    def _call_arnied(self, fn, *args, **kwargs):
        """
        Simple proxy around the arnied to be overwritten in child classes.

        :param fn: function to invoke in the arnied wrapper
        :type fn: py:function
        :param args: arguments to be passed to function
        :param kwargs: named arguments to be passed to function
        """
        return fn(*args, **kwargs)


#: pattern for more verbose error message for :py:class:`CommitException`
COMMIT_EXCEPTION_MESSAGE = """\
Error committing CNF variables!
----------------------------
Input:
{cnfvars}
----------------------------
Error:
{msg}
"""


class CommitException(Exception):
    """Custom exception for commit errors."""

    def __init__(self, cnfvars, msg):
        """
        Initialize this exception.

        :param cnfvars: list of CNF variables that would be committed
        :type cnfvars: CnfList
        :param str msg: error message
        """
        super().__init__(msg)
        self.message = COMMIT_EXCEPTION_MESSAGE.format(cnfvars=cnfvars, msg=msg)