Increase version to 1.7.4

[pyi2ncommon] / src / cnfvar / binary.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>

"""
binary: wrappers around binaries that work with the CNF store.

Featuring:
    - CnfBinary: stateless class that can be used to invoke the different binaries
      in a Python-friendly way.

.. note:: It is written as a class on purpose for it to be easily extended to
          support invoking non-local binaries, but methods are all class methods since
          the class is stateless.

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

.. codeauthor:: Intra2net
"""

import subprocess
import logging
import shlex
import html
import re

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

#: default get_cnf binary
BIN_GET_CNF = "/usr/intranator/bin/get_cnf"
#: default set_cnf binary
BIN_SET_CNF = "/usr/intranator/bin/set_cnf"
#: encoding used by the get_cnf and set_cnf binaries
ENCODING = "latin1"


class CnfBinary:
    """Provide wrappers around the multiple binaries to handle the CNF store."""

    @classmethod
    def run_cmd(cls, cmd, cmd_input=None, ignore_errors=False, timeout=60, encoding=ENCODING):
        """
        Run commands on the local machine with input via stdin.

        :param cmd: command to run
        :type cmd: str or list
        :param str cmd_input: input to give to the command
        :param bool ignore_errors: whether to ignore the exit code or raise if not zero
        :param int timeout: amount of seconds to wait for the command to complete
        :param str encoding: encoding to use (pass latin1 when not working with JSON)
        :returns: command result
        :rtype: :py:class:`subprocess.CompletedProcess`
        """
        if isinstance(cmd, str):
            cmd = shlex.split(cmd)

        log.debug("Running binary cmd `%s` with input:\n%s",
                  ' '.join(cmd), cmd_input)
        retval = subprocess.run(cmd, input=cmd_input, check=not ignore_errors,
                                capture_output=True, encoding=encoding, timeout=timeout)
        log.debug("Command exited with \n"
                  "\treturncode=%d\n"
                  "\tstderr=%s\n"
                  "\tstdout=%s", retval.returncode, retval.stderr, retval.stdout)
        return retval

    @classmethod
    def get_cnf(cls, name=None, instance=None, no_children=False,
                as_json=False):
        """
        Wrapper around the `get_cnf` binary.

        :param str name: optional name of the CNFs to get
        :param instance: CNF instance
        :type instance: str or int
        :param bool no_children: whether to return child CNFs
        :param bool as_json: whether to return a JSON-formatted string
        :returns: output of the tool
        :rtype: str

        .. note:: being a wrapper, this function does not do anything extra
                  like checking if arnied is running or waiting for generate.
        """
        if name is None and instance is not None:
            raise ValueError("cannot pass `instance` without a `name`")

        if isinstance(instance, str):
            instance = int(instance)  # validate
        elif instance is not None and not isinstance(instance, int):
            raise TypeError(f"`instance` is of wrong type {type(instance)}")

        # make sure 0 instance cnfvars like e.g. NICs can also be filtered by instance
        cmd = f"{BIN_GET_CNF} {name or ''} {instance if instance is not None else ''}"

        encoding = ENCODING
        if no_children:
            cmd += " --no-childs"
        if as_json:
            cmd += " --json"
            encoding = "utf8"

        # TODO: should error handling be improved so error messages
        # from the binary are converted into specific exceptions types?
        output = cls.run_cmd(cmd, encoding=encoding).stdout.strip()
        return output

    @classmethod
    def set_cnf(cls, input_str=None, input_file=None, delete=False,
                delete_file=False, as_json=False, fix_problems=False, **kwargs):
        """
        Wrapper around the `set_cnf` binary.

        :param str input_str: string to send as input to the binary
        :param str input_file: path to a file to pass to the binary
        :param bool delete: whether to delete the corresponding CNFs
        :param bool delete_file: whether to delete the file passed as input
        :param bool as_json: whether to interpret input as JSON
        :param bool fix_problems: whether to automatically fix errors in the vars
        :param kwargs: extra arguments to pass to the binary - underscores are
                       replaced by dash, e.g. set_cnf(..., enable_queue=True)
                       becomes `/usr/intranator/bin/set_cnf --enable-queue`
        :raises: :py:class:`SetCnfException` in case the binary errors out

        .. note:: being a wrapper, this function does not do anything extra
                  like checking if arnied is running or waiting for generate.
        """
        if input_str is None and input_file is None:
            raise ValueError("Need to pass either a string or a file")

        if delete_file is True and input_file is None:
            raise ValueError("Cannot delete an unspecified file")

        cmd = f"{BIN_SET_CNF}"
        encoding = ENCODING
        if as_json:
            cmd += " --json"
            encoding = "utf8"
        if delete:
            cmd += " -x"
        if delete_file:
            cmd += " --delete-file"
        if fix_problems:
            cmd += " --fix-problems"

        for k, v in kwargs.items():
            if v is True:
                k = "-".join(k.split("_"))
                cmd += f" --{k}"

        if input_file:
            cmd += f" {input_file}"

        try:
            cls.run_cmd(cmd, cmd_input=input_str, encoding=encoding)
        except subprocess.CalledProcessError as ex:
            # clean up the error message
            ex.stderr = cls._clean_set_cnf_error(ex.stderr)
            raise

    @classmethod
    def _clean_set_cnf_error(cls, err):
        """
        Turn the error from the set_cnf binary into a more readable message.

        :param str err: error message from the binary
        :returns: the clean error message
        :rtype: str

        Keep only offending lines and strip out HTML tags.
        """
        def get_error_lines(output_lines):
            for cnfline in output_lines:
                buffer = ""
                quoted = False
                parts = []
                for c in cnfline:
                    last_chr = buffer[-1] if buffer else None
                    # this is a literal (unescaped) quote
                    if c == "\"" and last_chr != "\\":
                        quoted = not quoted

                    if c == " " and not quoted:
                        parts.append(buffer)
                        buffer = ""
                    else:
                        buffer += c
                parts.append(buffer)

                if parts[-2] == "0":
                    # no errors, ignore
                    continue

                lineno, name, instance, parent, value, exit_code, error = parts

                # the binary outputs HTML strings
                error = html.unescape(error)
                # replace line breaks for readability
                error = re.sub(r"(<br\s*>)+", " ", error)
                # clean up HTML tags
                error = re.sub(r"<[^<]+?>", "", error)

                if parent == "-1":
                    message = f"`` {lineno} {name},{instance}: {value} ``"
                else:
                    message = f"`` {lineno}   ({parent}) {name},{instance}: {value} ``"
                yield f"Error in {message}: {error} (code={exit_code})"

        lines = err.splitlines()
        if len(lines) == 0:
            return err

        if "fatal errors in changes" not in lines[0]:
            return err

        errors = list(get_error_lines(lines[1:]))
        return "\n".join(errors)