salt/tests/integration/utils/testprogram.py

# -*- coding: utf-8 -*-
"""
Classes for starting/stopping/status salt daemons, auxiliary
scripts, generic commands.
"""

from __future__ import absolute_import

import atexit
import copy
import errno
import getpass
import logging
import os
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
from datetime import datetime, timedelta

import pytest
import salt.defaults.exitcodes as exitcodes
import salt.utils.files
import salt.utils.platform
import salt.utils.process
import salt.utils.psutil_compat as psutils
import salt.utils.yaml
from salt.ext import six
from salt.ext.six.moves import range
from saltfactories.utils.processes.helpers import (
    terminate_process,
    terminate_process_list,
)
from tests.support.cli_scripts import ScriptPathMixin
from tests.support.runtests import RUNTIME_VARS
from tests.support.unit import TestCase

log = logging.getLogger(__name__)


if "TimeoutError" not in __builtins__:

    class TimeoutError(OSError):
        """Compatibility exception with python3"""

    __builtins__["TimeoutError"] = TimeoutError


@pytest.mark.windows_whitelisted
class TestProgramMeta(type):
    """
    Stack all inherited config_attrs and dirtree dirs from the base classes.
    """

    def __new__(mcs, name, bases, attrs):
        config_vals = {}
        config_attrs = set()
        dirtree = set()

        for base in bases:
            config_vals.update(getattr(base, "config_vals", {}))
            config_attrs.update(getattr(base, "config_attrs", {}))
            dirtree.update(getattr(base, "dirtree", []))

        config_vals.update(attrs.get("config_vals", {}))
        attrs["config_vals"] = config_vals

        config_attrs.update(attrs.get("config_attrs", {}))
        attrs["config_attrs"] = config_attrs

        dirtree.update(attrs.get("dirtree", []))
        attrs["dirtree"] = dirtree

        return super(TestProgramMeta, mcs).__new__(mcs, name, bases, attrs)


# pylint: disable=too-many-instance-attributes
@pytest.mark.windows_whitelisted
class TestProgram(six.with_metaclass(TestProgramMeta, object)):
    """
    Set up an arbitrary executable to run.

    :attribute dirtree: An iterable of directories to be created
    """

    empty_config = ""
    config_file = ""

    config_attrs = set(["name", "test_dir", "config_dirs"])
    config_vals = {}
    config_base = ""
    config_dir = os.path.join("etc")
    configs = {}
    config_types = (
        str,
        six.string_types,
    )

    dirtree = [
        "&config_dirs",
    ]

    @staticmethod
    def config_caster(cfg):
        return str(cfg)

    def __init__(
        self,
        program=None,
        name=None,
        env=None,
        shell=False,
        parent_dir=None,
        clean_on_exit=True,
        **kwargs
    ):
        self.program = program or getattr(self, "program", None)
        self.name = name or getattr(self, "name", "")
        self.env = env or {}
        self.shell = shell
        self._parent_dir = parent_dir or None
        self.clean_on_exit = clean_on_exit
        self._root_dir = kwargs.pop("root_dir", self.name)
        config_dir = kwargs.pop("config_dir", None)
        if config_dir is None:
            config_dir = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP)
        self.config_dir = config_dir

        config_attrs = copy.copy(self.config_attrs)
        config_attrs.update(kwargs.pop("config_attrs", set()))
        self.config_attrs = config_attrs

        config_vals = copy.copy(self.config_vals)
        config_vals.update(kwargs.pop("config_vals", {}))
        self.config_vals = config_vals

        config_base = copy.deepcopy(self.config_base)
        config_base = self.config_merge(
            config_base, kwargs.pop("config_base", self.config_types[0]())
        )
        self.config_base = config_base

        configs = copy.deepcopy(self.configs)
        for cname, cinfo in kwargs.pop("configs", {}).items():
            target = configs.setdefault(cname, {})
            if "path" in cinfo:
                target["path"] = cinfo["path"]
            if "map" in cinfo:
                target_map = target.setdefault("map", self.config_types[0]())
                target_map = self.config_merge(target_map, cinfo["map"])
                target["map"] = target_map
        self.configs = configs

        if not self.name:
            if not self.program:
                raise ValueError(
                    '"{0}" object must specify "program" parameter'.format(
                        self.__class__.__name__
                    )
                )
            self.name = os.path.basename(self.program)

        self.process = None
        self.created_parent_dir = False
        self._setup_done = False

        dirtree = set(self.dirtree)
        dirtree.update(kwargs.pop("dirtree", []))
        self.dirtree = dirtree

        # Register the exit clean-up before making anything needing clean-up
        atexit.register(self.cleanup)

    def __enter__(self):
        pass

    def __exit__(self, typ, value, traceback):
        pass

    @property
    def test_dir(self):
        """Directory that will contains all of the static and dynamic files for the daemon"""
        return os.path.join(self.parent_dir, self._root_dir)

    def config_file_get(self, config):
        """Get the filename (viz. path) to the configuration file"""
        cfgf = self.configs[config].get("path")
        if cfgf:
            cfgf.format(**self.config_subs())
        else:
            cfgf = os.path.join(self.config_dir, config)
        return cfgf

    def config_dir_get(self, config):
        """Get the parent directory for the configuration file"""
        return os.path.dirname(self.config_file_get(config))

    @property
    def config_dirs(self):
        """Return a list of configuration directories"""
        cdirs = [self.config_dir_get(config) for config in self.configs.keys()]
        return cdirs

    def abs_path(self, path):
        """Absolute path of file including the test_dir"""
        return os.path.join(self.test_dir, path)

    @property
    def start_pid(self):
        """PID of the called script prior to daemonizing."""
        return self.process.pid if self.process else None

    @property
    def parent_dir(self):
        """
        Directory that contains everything generated for running scripts - possibly
        for multiple scripts.
        """
        if self._parent_dir is None:
            self.created_parent_dir = True
            self._parent_dir = tempfile.mkdtemp(prefix="salt-testdaemon-")
        else:
            self._parent_dir = os.path.abspath(os.path.normpath(self._parent_dir))
            if not os.path.exists(self._parent_dir):
                self.created_parent_dir = True
                os.makedirs(self._parent_dir)
            elif not os.path.isdir(self._parent_dir):
                raise ValueError(
                    'Parent path "{0}" exists but is not a directory'.format(
                        self._parent_dir
                    )
                )
        return self._parent_dir

    def config_write(self, config):
        """Write out the config to a file"""
        if not config:
            return
        cpath = self.abs_path(self.config_file_get(config))
        with salt.utils.files.fopen(cpath, "w") as cfo:
            cfg = self.config_stringify(config)
            log.debug(
                "Writing configuration for {0} to {1}:\n{2}".format(
                    self.name, cpath, cfg
                )
            )
            cfo.write(cfg)
            cfo.flush()

    def configs_write(self):
        """Write all configuration files"""
        for config in self.configs:
            self.config_write(config)

    def config_type(self, config):
        """Check if a configuration is an acceptable type."""
        return isinstance(config, self.config_types)

    def config_cast(self, config):
        """Cast a configuration to the internal expected type."""
        if not self.config_type(config):
            config = self.config_caster(config)
        return config

    def config_subs(self):
        """Get the substitution values for use to generate the config"""
        subs = dict([(attr, getattr(self, attr, None)) for attr in self.config_attrs])
        for key, val in self.config_vals.items():
            subs[key] = val.format(**subs)
        return subs

    def config_stringify(self, config):
        """Get the configuration as a string"""
        cfg = self.config_get(config)
        cfg.format(**self.config_subs())
        return cfg

    def config_merge(self, base, overrides):
        """Merge two configuration hunks"""
        base = self.config_cast(base)
        overrides = self.config_cast(overrides)
        return "".join([base, overrides])

    def config_get(self, config):
        """Get the configuration data"""
        return self.configs[config]

    def config_set(self, config, val):
        """Set the configuration data"""
        self.configs[config] = val

    def make_dirtree(self):
        """Create directory structure."""
        subdirs = []
        for branch in self.dirtree:
            log.debug("checking dirtree: {0}".format(branch))
            if not branch:
                continue
            if isinstance(branch, six.string_types) and branch[0] == "&":
                log.debug('Looking up dirtree branch "{0}"'.format(branch))
                try:
                    dirattr = getattr(self, branch[1:], None)
                    log.debug('dirtree "{0}" => "{1}"'.format(branch, dirattr))
                except AttributeError:
                    raise ValueError(
                        'Unable to find dirtree attribute "{0}" on object "{1}.name = {2}: {3}"'.format(
                            branch, self.__class__.__name__, self.name, dir(self),
                        )
                    )

                if not dirattr:
                    continue

                if isinstance(dirattr, six.string_types):
                    subdirs.append(dirattr)
                elif hasattr(dirattr, "__iter__"):
                    subdirs.extend(dirattr)
                else:
                    raise TypeError(
                        "Branch type of {0} in dirtree is unhandled".format(branch)
                    )
            elif isinstance(branch, six.string_types):
                subdirs.append(branch)
            else:
                raise TypeError(
                    "Branch type of {0} in dirtree is unhandled".format(branch)
                )

        for subdir in subdirs:
            path = self.abs_path(subdir)
            if not os.path.exists(path):
                log.debug("make_dirtree: {0}".format(path))
                os.makedirs(path)

    def setup(self, *args, **kwargs):
        """Create any scaffolding for run-time"""

        # unused
        _ = args, kwargs

        if not self._setup_done:
            self.make_dirtree()
            self.configs_write()
            self._setup_done = True

    def cleanup(self, *args, **kwargs):
        """ Clean out scaffolding of setup() and any run-time generated files."""
        # Unused for now
        _ = (args, kwargs)

        if self.process:
            try:
                self.process.kill()
                self.process.wait()
            except OSError:
                pass
        if os.path.exists(self.test_dir):
            shutil.rmtree(self.test_dir)
        if self.created_parent_dir and os.path.exists(self.parent_dir):
            shutil.rmtree(self.parent_dir)

    def run(
        self,
        args=None,
        catch_stderr=False,
        with_retcode=False,
        timeout=None,
        raw=False,
        env=None,
        verbatim_args=False,
        verbatim_env=False,
    ):
        """
        Execute a command possibly using a supplied environment.

        :param args:
            A command string or a command sequence of arguments for the program.

        :param catch_stderr: A boolean whether to capture and return stderr.

        :param with_retcode: A boolean whether to return the exit code.

        :param timeout: A float of how long to wait for the process to
            complete before it is killed.

        :param raw: A boolean whether to return buffer strings for stdout and
            stderr or sequences of output lines.

        :param env: A dictionary of environment key/value settings for the
            command.

        :param verbatim_args: A boolean whether to automatically add inferred arguments.

        :param verbatim_env: A boolean whether to automatically add inferred
            environment values.

        :return list: (stdout [,stderr] [,retcode])
        """

        # unused for now
        _ = verbatim_args

        self.setup()

        if args is None:
            args = []

        if env is None:
            env = {}

        env_delta = {}
        env_delta.update(self.env)
        env_delta.update(env)

        if not verbatim_env:
            env_pypath = env_delta.get("PYTHONPATH", os.environ.get("PYTHONPATH"))
            if not env_pypath:
                env_pypath = sys.path
            else:
                env_pypath = env_pypath.split(":")
                for path in sys.path:
                    if path not in env_pypath:
                        env_pypath.append(path)
            # Always ensure that the test tree is searched first for python modules
            if RUNTIME_VARS.CODE_DIR != env_pypath[0]:
                env_pypath.insert(0, RUNTIME_VARS.CODE_DIR)
            if salt.utils.platform.is_windows():
                env_delta["PYTHONPATH"] = ";".join(env_pypath)
            else:
                env_delta["PYTHONPATH"] = ":".join(env_pypath)

        cmd_env = dict(os.environ)
        cmd_env.update(env_delta)

        if salt.utils.platform.is_windows() and six.PY2:
            for k, v in cmd_env.items():
                if isinstance(k, six.text_type) or isinstance(v, six.text_type):
                    cmd_env[k.encode("ascii")] = v.encode("ascii")

        popen_kwargs = {
            "shell": self.shell,
            "stdout": subprocess.PIPE,
            "env": cmd_env,
        }

        if catch_stderr is True:
            popen_kwargs["stderr"] = subprocess.PIPE

        if not sys.platform.lower().startswith("win"):
            popen_kwargs["close_fds"] = True

            def detach_from_parent_group():
                """
                A utility function that prevents child process from getting parent signals.
                """
                os.setpgrp()

            popen_kwargs["preexec_fn"] = detach_from_parent_group

        if salt.utils.platform.is_windows():
            self.argv = ["python.exe", self.program]
        else:
            self.argv = [self.program]
        self.argv.extend(args)
        log.debug("TestProgram.run: %s Environment %s", self.argv, env_delta)
        process = subprocess.Popen(self.argv, **popen_kwargs)
        self.process = process

        if timeout is not None:
            stop_at = datetime.now() + timedelta(seconds=timeout)
            term_sent = False
            while True:
                process.poll()

                if datetime.now() > stop_at:
                    try:
                        terminate_process(pid=process.pid, kill_children=True)
                        process.wait()
                    except OSError as exc:
                        if exc.errno != errno.ESRCH:
                            raise

                    out = process.stdout.read().splitlines()
                    out.extend(
                        [
                            "Process took more than {0} seconds to complete. "
                            "Process Killed!".format(timeout)
                        ]
                    )
                    if catch_stderr:
                        err = process.stderr.read().splitlines()
                        if with_retcode:
                            return out, err, process.returncode
                        else:
                            return out, err
                    if with_retcode:
                        return out, process.returncode
                    else:
                        return out

                if process.returncode is not None:
                    break

        if catch_stderr:
            out, err = process.communicate()
            # Force closing stderr/stdout to release file descriptors
            if process.stdout is not None:
                process.stdout.close()
            if process.stderr is not None:
                process.stderr.close()
            # pylint: disable=maybe-no-member
            try:
                if with_retcode:
                    if out is not None and err is not None:
                        if not raw:
                            return (
                                out.splitlines(),
                                err.splitlines(),
                                process.returncode,
                            )
                        else:
                            return out, err, process.returncode
                    return out.splitlines(), [], process.returncode
                else:
                    if out is not None and err is not None:
                        if not raw:
                            return out.splitlines(), err.splitlines()
                        else:
                            return out, err
                    if not raw:
                        return out.splitlines(), []
                    else:
                        return out, []
            finally:
                try:
                    process.terminate()
                except OSError as err:
                    # process already terminated
                    pass
            # pylint: enable=maybe-no-member

        data = process.communicate()
        process.stdout.close()

        try:
            if with_retcode:
                if not raw:
                    return data[0].splitlines(), process.returncode
                else:
                    return data[0], process.returncode
            else:
                if not raw:
                    return data[0].splitlines()
                else:
                    return data[0]
        finally:
            try:
                process.terminate()
            except OSError as err:
                # process already terminated
                pass


@pytest.mark.windows_whitelisted
class TestSaltProgramMeta(TestProgramMeta):
    """
    A Meta-class to set self.script from the class name when it is
    not specifically set by a "script" argument.
    """

    def __new__(mcs, name, bases, attrs):
        if attrs.get("script") is None:
            if "Salt" in name:
                script = "salt-{0}".format(name.rsplit("Salt", 1)[-1].lower())
            if script is None:
                raise AttributeError(
                    'Class {0}: Unable to set "script" attribute: class name'
                    ' must include "Salt" or "script" must be explicitly set.'.format(
                        name
                    )
                )
            attrs["script"] = script

        config_base = {}
        configs = {}

        for base in bases:
            if "Salt" not in base.__name__:
                continue
            config_base.update(getattr(base, "config_base", {}))
            configs.update(getattr(base, "configs", {}))

        config_base.update(attrs.get("config_base", {}))
        attrs["config_base"] = config_base

        configs.update(attrs.get("configs", {}))
        attrs["configs"] = configs

        return super(TestSaltProgramMeta, mcs).__new__(mcs, name, bases, attrs)


@pytest.mark.windows_whitelisted
class TestSaltProgram(
    six.with_metaclass(TestSaltProgramMeta, TestProgram, ScriptPathMixin)
):
    """
    This is like TestProgram but with some functions to run a salt-specific
    auxiliary program.
    """

    config_types = (dict,)
    config_attrs = set(["log_dir", "script_dir"])

    pub_port = 4505
    ret_port = 4506
    for port in [pub_port, ret_port]:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            connect = sock.bind(("localhost", port))
        except (socket.error, OSError):
            # these ports are already in use, use different ones
            pub_port = 4606
            ret_port = 4607
            break
        sock.close()

    config_base = {
        "root_dir": "{test_dir}",
        "publish_port": pub_port,
        "ret_port": ret_port,
    }
    configs = {}
    config_dir = os.path.join("etc", "salt")

    log_dir = os.path.join("var", "log", "salt")

    dirtree = [
        "&log_dir",
        "&script_dir",
    ]

    script = ""
    script_dir = "bin"

    @staticmethod
    def config_caster(cfg):
        return salt.utils.yaml.safe_load(cfg)

    def __init__(self, *args, **kwargs):
        if len(args) < 2 and "program" not in kwargs:
            # This is effectively a place-holder - it gets set correctly after super()
            kwargs["program"] = self.script
        super(TestSaltProgram, self).__init__(*args, **kwargs)
        self.program = self.get_script_path(self.script)

    def config_merge(self, base, overrides):
        _base = self.config_cast(copy.deepcopy(base))
        _overrides = self.config_cast(overrides)
        # NOTE: this simple update will not work for deep dictionaries
        _base.update(copy.deepcopy(_overrides))
        return _base

    def config_get(self, config):
        cfg_base = {}
        for key, val in self.config_base.items():
            _val = val
            if val and isinstance(val, six.string_types) and val[0] == "&":
                _val = getattr(self, val[1:], None)
                if _val is None:
                    continue
            cfg_base[key] = _val
        if config in self.configs:
            cfg = {}
            for key, val in self.configs.get(config, {}).get("map", {}).items():
                _val = val
                if val and isinstance(val, six.string_types) and val[0] == "&":
                    _val = getattr(self, val[1:], None)
                    if _val is None:
                        continue
                cfg[key] = _val
            cfg = self.config_merge(cfg_base, cfg)
        log.debug("Generated config => {0}".format(cfg))
        return cfg

    def config_stringify(self, config):
        """Transform the configuration data into a string (suitable to write to a file)"""
        subs = self.config_subs()
        cfg = {}
        for key, val in self.config_get(config).items():
            if isinstance(val, six.string_types):
                cfg[key] = val.format(**subs)
            else:
                cfg[key] = val
        return salt.utils.yaml.safe_dump(cfg, default_flow_style=False)

    def run(self, **kwargs):  # pylint: disable=arguments-differ
        if not kwargs.get("verbatim_args"):
            args = kwargs.setdefault("args", [])
            if "-c" not in args and "--config-dir" not in args:
                args.extend(["--config-dir", self.abs_path(self.config_dir)])
        return super(TestSaltProgram, self).run(**kwargs)


@pytest.mark.windows_whitelisted
class TestProgramSalt(TestSaltProgram):
    """Class to manage salt"""

    configs = {"master": {}}
    script = "salt"


@pytest.mark.windows_whitelisted
class TestProgramSaltCall(TestSaltProgram):
    """Class to manage salt-call"""

    configs = {"minion": {"map": {"id": "{name}"}}}


@pytest.mark.windows_whitelisted
class TestProgramSaltRun(TestSaltProgram):
    """Class to manage salt-run"""

    configs = {"master": {}}

    def __init__(self, *args, **kwargs):
        cfgb = kwargs.setdefault("config_base", {})
        _ = cfgb.setdefault("user", getpass.getuser())
        super(TestProgramSaltRun, self).__init__(*args, **kwargs)


@pytest.mark.windows_whitelisted
class TestDaemon(TestProgram):
    """
    Run one of the standard daemons
    """

    script = None
    pid_file = None
    pid_dir = os.path.join("var", "run")

    dirtree = [
        "&pid_dir",
    ]

    def __init__(self, *args, **kwargs):
        self.script = kwargs.pop("script", self.script)
        self.pid_file = kwargs.pop(
            "pid_file",
            self.pid_file if self.pid_file else "{0}.pid".format(self.script),
        )
        self.pid_dir = kwargs.pop("pid_dir", self.pid_dir)
        self._shutdown = False
        if not args and "program" not in kwargs:
            # This is effectively a place-holder - it gets set correctly after super()
            kwargs["program"] = self.script
        super(TestDaemon, self).__init__(*args, **kwargs)

    @property
    def pid_path(self):
        """Path to the pid file created by the daemon"""
        return (
            os.path.join(self.pid_dir, self.pid_file)
            if os.path.sep not in self.pid_file
            else self.pid_file
        )

    @property
    def daemon_pid(self):
        """Return the daemon PID"""
        daemon_pid = None
        pid_path = self.abs_path(self.pid_path)
        if salt.utils.process.check_pidfile(pid_path):
            daemon_pid = salt.utils.process.get_pidfile(pid_path)
        return daemon_pid

    def wait_for_daemon_pid(self, timeout=10):
        """Wait up to timeout seconds for the PID file to appear and return the PID"""
        endtime = time.time() + timeout
        while True:
            pid = self.daemon_pid
            if pid:
                return pid
            if endtime < time.time():
                raise TimeoutError(
                    'Timeout waiting for "{0}" pid in "{1}"'.format(
                        self.name, self.abs_path(self.pid_path)
                    )
                )
            time.sleep(0.2)

    def is_running(self):
        """Is the daemon running?"""
        ret = False
        if not self._shutdown:
            try:
                pid = self.wait_for_daemon_pid()
                ret = psutils.pid_exists(pid)
            except TimeoutError:
                pass
        return ret

    def find_orphans(self, cmdline):
        """Find orphaned processes matching the specified cmdline"""
        ret = []
        if six.PY3:
            cmdline = " ".join(cmdline)
            for proc in psutils.process_iter():
                try:
                    for item in proc.cmdline():
                        if cmdline in item:
                            ret.append(proc)
                except psutils.NoSuchProcess:
                    # Process exited between when process_iter was invoked and
                    # when we tried to invoke this instance's cmdline() func.
                    continue
                except psutils.AccessDenied:
                    # We might get access denied if not running as root
                    if not salt.utils.platform.is_windows():
                        pinfo = proc.as_dict(attrs=["pid", "name", "username"])
                        log.error(
                            "Unable to access process %s, "
                            "running command %s as user %s",
                            pinfo["pid"],
                            pinfo["name"],
                            pinfo["username"],
                        )
                        continue
        else:
            cmd_len = len(cmdline)
            for proc in psutils.process_iter():
                try:
                    proc_cmdline = proc.cmdline()
                except psutils.NoSuchProcess:
                    # Process exited between when process_iter was invoked and
                    # when we tried to invoke this instance's cmdline() func.
                    continue
                except psutils.AccessDenied:
                    # We might get access denied if not running as root
                    if not salt.utils.platform.is_windows():
                        pinfo = proc.as_dict(attrs=["pid", "name", "username"])
                        log.error(
                            "Unable to access process %s, "
                            "running command %s as user %s",
                            pinfo["pid"],
                            pinfo["name"],
                            pinfo["username"],
                        )
                        continue
                if any(
                    (cmdline == proc_cmdline[n : n + cmd_len])
                    for n in range(len(proc_cmdline) - cmd_len + 1)
                ):
                    ret.append(proc)
        return ret

    def shutdown(self, signum=signal.SIGTERM, timeout=10, wait_for_orphans=0):
        """Shutdown a running daemon"""
        if not self._shutdown:
            try:
                pid = self.wait_for_daemon_pid(timeout)
                terminate_process(pid=pid, kill_children=True)
            except TimeoutError:
                pass
        if self.process:
            terminate_process(pid=self.process.pid, kill_children=True)
            self.process.wait()
            if wait_for_orphans:
                # NOTE: The process for finding orphans is greedy, it just
                # looks for processes with the same cmdline which are owned by
                # PID 1.
                orphans = self.find_orphans(self.argv)
                last = time.time()
                while True:
                    if orphans:
                        log.debug("Terminating orphaned child processes: %s", orphans)
                        terminate_process_list(orphans)
                        last = time.time()
                    if (time.time() - last) >= wait_for_orphans:
                        break
                    time.sleep(0.25)
                    orphans = self.find_orphans(self.argv)
            self.process = None
        self._shutdown = True

    def cleanup(self, *args, **kwargs):
        """Remove left-over scaffolding - antithesis of setup()"""

        # Shutdown if not alreadt shutdown
        self.shutdown()
        super(TestDaemon, self).cleanup(*args, **kwargs)


@pytest.mark.windows_whitelisted
class TestSaltDaemon(
    six.with_metaclass(TestSaltProgramMeta, TestDaemon, TestSaltProgram)
):
    """
    A class to run arbitrary salt daemons (master, minion, syndic, etc.)
    """


@pytest.mark.windows_whitelisted
class TestDaemonSaltMaster(TestSaltDaemon):
    """
    Manager for salt-master daemon.
    """

    configs = {"master": {}}

    def __init__(self, *args, **kwargs):
        cfgb = kwargs.setdefault("config_base", {})
        _ = cfgb.setdefault("user", getpass.getuser())
        super(TestDaemonSaltMaster, self).__init__(*args, **kwargs)


@pytest.mark.windows_whitelisted
class TestDaemonSaltMinion(TestSaltDaemon):
    """
    Manager for salt-minion daemon.
    """

    configs = {"minion": {"map": {"id": "{name}"}}}

    def __init__(self, *args, **kwargs):
        cfgb = kwargs.setdefault("config_base", {})
        _ = cfgb.setdefault("user", getpass.getuser())
        super(TestDaemonSaltMinion, self).__init__(*args, **kwargs)


@pytest.mark.windows_whitelisted
class TestDaemonSaltApi(TestSaltDaemon):
    """
    Manager for salt-api daemon.
    """


@pytest.mark.windows_whitelisted
class TestDaemonSaltSyndic(TestSaltDaemon):
    """
    Manager for salt-syndic daemon.
    """

    configs = {
        "master": {"map": {"syndic_master": "localhost"}},
        "minion": {"map": {"id": "{name}"}},
    }

    def __init__(self, *args, **kwargs):
        cfgb = kwargs.setdefault("config_base", {})
        _ = cfgb.setdefault("user", getpass.getuser())
        super(TestDaemonSaltSyndic, self).__init__(*args, **kwargs)


@pytest.mark.windows_whitelisted
class TestDaemonSaltProxy(TestSaltDaemon):
    """
    Manager for salt-proxy daemon.
    """

    pid_file = "salt-minion.pid"
    configs = {"proxy": {}}

    def __init__(self, *args, **kwargs):
        cfgb = kwargs.setdefault("config_base", {})
        _ = cfgb.setdefault("user", getpass.getuser())
        super(TestDaemonSaltProxy, self).__init__(*args, **kwargs)

    def run(self, **kwargs):
        if not kwargs.get("verbatim_args"):
            args = kwargs.setdefault("args", [])
            if "--proxyid" not in args:
                args.extend(["--proxyid", self.name])
        return super(TestDaemonSaltProxy, self).run(**kwargs)


@pytest.mark.windows_whitelisted
class TestProgramCase(TestCase):
    """
    Utilities for unit tests that use TestProgram()
    """

    def setUp(self):
        # Setup for scripts
        if not getattr(self, "_test_dir", None):
            self._test_dir = tempfile.mkdtemp(prefix="salt-testdaemon-")
        super(TestProgramCase, self).setUp()

    def tearDown(self):
        # shutdown for scripts
        if self._test_dir and os.path.sep == self._test_dir[0]:
            shutil.rmtree(self._test_dir)
            self._test_dir = None
        super(TestProgramCase, self).tearDown()

    def assert_exit_status(
        self, status, ex_status, message=None, stdout=None, stderr=None
    ):
        """
        Helper function to verify exit status and emit failure information.
        """

        ex_val = getattr(exitcodes, ex_status)
        _message = "" if not message else " ({0})".format(message)
        _stdout = "" if not stdout else "\nstdout: {0}".format(stdout)
        _stderr = "" if not stderr else "\nstderr: {0}".format(stderr)
        self.assertEqual(
            status,
            ex_val,
            "Exit status was {0}, must be {1} (salt.default.exitcodes.{2}){3}{4}{5}".format(
                status, ex_val, ex_status, _message, _stdout, _stderr,
            ),
        )