salt/tests/conftest.py

# -*- coding: utf-8 -*-
'''
    :codeauthor: Pedro Algarvio (pedro@algarvio.me)

    tests.conftest
    ~~~~~~~~~~~~~~

    Prepare py.test for our test suite
'''
# pylint: disable=wrong-import-order,wrong-import-position,3rd-party-local-module-not-gated
# pylint: disable=redefined-outer-name,invalid-name

# Import python libs
from __future__ import absolute_import, print_function, unicode_literals
import os
import sys
import stat
import pprint
import shutil
import socket
import fnmatch
import logging
import tempfile
import textwrap
from functools import partial, wraps
from contextlib import contextmanager

TESTS_DIR = os.path.dirname(os.path.normpath(os.path.abspath(__file__)))
CODE_DIR = os.path.dirname(TESTS_DIR)

# Change to code checkout directory
os.chdir(CODE_DIR)

# Make sure the current directory is the first item in sys.path
if CODE_DIR in sys.path:
    sys.path.remove(CODE_DIR)
sys.path.insert(0, CODE_DIR)

# Import test libs
from tests.support.runtests import RUNTIME_VARS
from tests.support.sminion import create_sminion

# Import pytest libs
import pytest
import _pytest.logging
import _pytest.skipping
from _pytest.mark.evaluate import MarkEvaluator

# Import 3rd-party libs
import psutil
from salt.ext import six

# Import salt libs
import salt.loader
import salt.config
import salt.utils.files
import salt.utils.path
import salt.log.setup
import salt.log.mixins
import salt.utils.platform
import salt.utils.win_functions
from salt.serializers import yaml
from salt.utils.immutabletypes import freeze

# Import Pytest Salt libs
from pytestsalt.utils import cli_scripts

# Coverage
if 'COVERAGE_PROCESS_START' in os.environ:
    MAYBE_RUN_COVERAGE = True
    COVERAGERC_FILE = os.environ['COVERAGE_PROCESS_START']
else:
    COVERAGERC_FILE = os.path.join(CODE_DIR, '.coveragerc')
    MAYBE_RUN_COVERAGE = sys.argv[0].endswith('pytest.py') or '_COVERAGE_RCFILE' in os.environ
    if MAYBE_RUN_COVERAGE:
        # Flag coverage to track suprocesses by pointing it to the right .coveragerc file
        os.environ[str('COVERAGE_PROCESS_START')] = str(COVERAGERC_FILE)

# Define the pytest plugins we rely on
pytest_plugins = ['tempdir', 'helpers_namespace', 'salt-runtests-bridge']

# Define where not to collect tests from
collect_ignore = ['setup.py']
# pylint: enable=invalid-name


# Patch PyTest logging handlers
# pylint: disable=protected-access,too-many-ancestors
class LogCaptureHandler(salt.log.mixins.ExcInfoOnLogLevelFormatMixIn,
                        _pytest.logging.LogCaptureHandler):
    '''
    Subclassing PyTest's LogCaptureHandler in order to add the
    exc_info_on_loglevel functionality.
    '''


_pytest.logging.LogCaptureHandler = LogCaptureHandler


class LiveLoggingStreamHandler(salt.log.mixins.ExcInfoOnLogLevelFormatMixIn,
                               _pytest.logging._LiveLoggingStreamHandler):
    '''
    Subclassing PyTest's LiveLoggingStreamHandler in order to add the
    exc_info_on_loglevel functionality.
    '''


_pytest.logging._LiveLoggingStreamHandler = LiveLoggingStreamHandler
# pylint: enable=protected-access,too-many-ancestors

# Reset logging root handlers
for handler in logging.root.handlers[:]:
    logging.root.removeHandler(handler)


# Reset the root logger to it's default level(because salt changed it)
logging.root.setLevel(logging.WARNING)

log = logging.getLogger('salt.testsuite')


# ----- PyTest Tempdir Plugin Hooks --------------------------------------------------------------------------------->
def pytest_tempdir_temproot():
    # Taken from https://github.com/saltstack/salt/blob/v2019.2.0/tests/support/paths.py
    # Avoid ${TMPDIR} and gettempdir() on MacOS as they yield a base path too long
    # for unix sockets: ``error: AF_UNIX path too long``
    # Gentoo Portage prefers ebuild tests are rooted in ${TMPDIR}
    if not sys.platform.startswith('darwin'):
        tempdir = os.environ.get('TMPDIR') or tempfile.gettempdir()
    else:
        tempdir = '/tmp'
    return os.path.abspath(os.path.realpath(tempdir))


def pytest_tempdir_basename():
    '''
    Return the temporary directory basename for the salt test suite.
    '''
    return 'salt-tests-tmpdir'
# <---- PyTest Tempdir Plugin Hooks ----------------------------------------------------------------------------------


# ----- CLI Options Setup ------------------------------------------------------------------------------------------->
def pytest_addoption(parser):
    '''
    register argparse-style options and ini-style config values.
    '''
    parser.addoption(
        '--sysinfo',
        default=False,
        action='store_true',
        help='Print some system information.'
    )
    parser.addoption(
        '--transport',
        default='zeromq',
        choices=('zeromq', 'tcp'),
        help=('Select which transport to run the integration tests with, '
              'zeromq or tcp. Default: %default')
    )
    test_selection_group = parser.getgroup('Tests Selection')
    test_selection_group.addoption(
        '--ssh',
        '--ssh-tests',
        dest='ssh',
        action='store_true',
        default=False,
        help='Run salt-ssh tests. These tests will spin up a temporary '
             'SSH server on your machine. In certain environments, this '
             'may be insecure! Default: False'
    )
    test_selection_group.addoption(
        '--proxy',
        '--proxy-tests',
        dest='proxy',
        action='store_true',
        default=False,
        help='Run proxy tests'
    )
    test_selection_group.addoption(
        '--run-destructive',
        action='store_true',
        default=False,
        help='Run destructive tests. These tests can include adding '
             'or removing users from your system for example. '
             'Default: False'
    )
    test_selection_group.addoption(
        '--run-expensive',
        action='store_true',
        default=False,
        help='Run expensive tests. These tests usually involve costs '
             'like for example bootstrapping a cloud VM. '
             'Default: False'
    )
    output_options_group = parser.getgroup('Output Options')
    output_options_group.addoption(
        '--output-columns',
        default=80,
        type=int,
        help='Number of maximum columns to use on the output'
    )
    output_options_group.addoption(
        '--no-colors',
        '--no-colours',
        default=False,
        action='store_true',
        help='Disable colour printing.'
    )

    # ----- Test Groups --------------------------------------------------------------------------------------------->
    # This will allow running the tests in chunks
    test_selection_group.addoption(
        '--test-group-count', dest='test-group-count', type=int,
        help='The number of groups to split the tests into'
    )
    test_selection_group.addoption(
        '--test-group', dest='test-group', type=int,
        help='The group of tests that should be executed'
    )
    # <---- Test Groups ----------------------------------------------------------------------------------------------
# <---- CLI Options Setup --------------------------------------------------------------------------------------------


# ----- Register Markers -------------------------------------------------------------------------------------------->
@pytest.mark.trylast
def pytest_configure(config):
    '''
    called after command line options have been parsed
    and all plugins and initial conftest files been loaded.
    '''
    for dirname in os.listdir(CODE_DIR):
        if not os.path.isdir(dirname):
            continue
        if dirname != 'tests':
            config.addinivalue_line('norecursedirs', os.path.join(CODE_DIR, dirname))

    config.addinivalue_line('norecursedirs', os.path.join(CODE_DIR, 'templates'))
    config.addinivalue_line('norecursedirs', os.path.join(CODE_DIR, 'tests/kitchen'))
    config.addinivalue_line('norecursedirs', os.path.join(CODE_DIR, 'tests/support'))

    # Expose the markers we use to pytest CLI
    config.addinivalue_line(
        'markers',
        'destructive_test: Run destructive tests. These tests can include adding '
        'or removing users from your system for example.'
    )
    config.addinivalue_line(
        'markers',
        'skip_if_not_root: Skip if the current user is not `root`.'
    )
    config.addinivalue_line(
        'markers',
        'skip_if_binaries_missing(*binaries, check_all=False, message=None): Skip if '
        'any of the passed binaries are not found in path. If \'check_all\' is '
        '\'True\', then all binaries must be found.'
    )
    config.addinivalue_line(
        'markers',
        'requires_network(only_local_network=False): Skip if no networking is set up. '
        'If \'only_local_network\' is \'True\', only the local network is checked.'
    )
    config.addinivalue_line(
        'markers',
        'requires_salt_modules(*required_module_names): Skip if at least one module is not available. '
    )
    # Make sure the test suite "knows" this is a pytest test run
    RUNTIME_VARS.PYTEST_SESSION = True
# <---- Register Markers ---------------------------------------------------------------------------------------------


# ----- PyTest Tweaks ----------------------------------------------------------------------------------------------->
def set_max_open_files_limits(min_soft=3072, min_hard=4096):

    # Get current limits
    if salt.utils.platform.is_windows():
        import win32file
        prev_hard = win32file._getmaxstdio()
        prev_soft = 512
    else:
        import resource
        prev_soft, prev_hard = resource.getrlimit(resource.RLIMIT_NOFILE)

    # Check minimum required limits
    set_limits = False
    if prev_soft < min_soft:
        soft = min_soft
        set_limits = True
    else:
        soft = prev_soft

    if prev_hard < min_hard:
        hard = min_hard
        set_limits = True
    else:
        hard = prev_hard

    # Increase limits
    if set_limits:
        log.debug(
            ' * Max open files settings is too low (soft: %s, hard: %s) for running the tests. '
            'Trying to raise the limits to soft: %s, hard: %s',
            prev_soft,
            prev_hard,
            soft,
            hard
        )
        try:
            if salt.utils.platform.is_windows():
                hard = 2048 if hard > 2048 else hard
                win32file._setmaxstdio(hard)
            else:
                resource.setrlimit(resource.RLIMIT_NOFILE, (soft, hard))
        except Exception as err:
            log.error(
                'Failed to raise the max open files settings -> %s. Please issue the following command '
                'on your console: \'ulimit -u %s\'',
                err,
                soft,
            )
            exit(1)
    return soft, hard


def pytest_report_header():
    soft, hard = set_max_open_files_limits()
    return 'max open files; soft: {}; hard: {}'.format(soft, hard)


def pytest_runtest_logstart(nodeid):
    '''
    implements the runtest_setup/call/teardown protocol for
    the given test item, including capturing exceptions and calling
    reporting hooks.
    '''
    log.debug('>>>>> START >>>>> %s', nodeid)


def pytest_runtest_logfinish(nodeid):
    '''
    called after ``pytest_runtest_call``
    '''
    log.debug('<<<<< END <<<<<<< %s', nodeid)


@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_collection_modifyitems(config, items):
    # Let PyTest or other plugins handle the initial collection
    yield
    groups_collection_modifyitems(config, items)

    log.warning('Mofifying collected tests to keep track of fixture usage')
    for item in items:
        for fixture in item.fixturenames:
            if fixture not in item._fixtureinfo.name2fixturedefs:
                continue
            for fixturedef in item._fixtureinfo.name2fixturedefs[fixture]:
                try:
                    node_ids = fixturedef.node_ids
                except AttributeError:
                    node_ids = fixturedef.node_ids = set()
                node_ids.add(item.nodeid)
                try:
                    fixturedef.finish.__wrapped__
                except AttributeError:
                    original_func = fixturedef.finish

                    def wrapper(func, fixturedef):

                        @wraps(func)
                        def wrapped(self, request):
                            if self.node_ids:
                                log.warning(
                                    '%s is still going to be used, not terminating it. '
                                    'Still in use on:\n%s',
                                    self,
                                    pprint.pformat(list(self.node_ids))
                                )
                                return
                            log.warning('Finish called on %s', self)
                            return func(request)
                        return partial(wrapped, fixturedef)

                    fixturedef.finish = wrapper(fixturedef.finish, fixturedef)
                    try:
                        fixturedef.finish.__wrapped__
                    except AttributeError:
                        fixturedef.finish.__wrapped__ = original_func


@pytest.hookimpl(trylast=True, hookwrapper=True)
def pytest_runtest_protocol(item, nextitem):
    used_fixture_defs = []
    for fixture in item.fixturenames:
        if fixture not in item._fixtureinfo.name2fixturedefs:
            continue
        for fixturedef in reversed(item._fixtureinfo.name2fixturedefs[fixture]):
            used_fixture_defs.append(fixturedef)
    for fixturedef in used_fixture_defs:
        log.warning('Before Test. Fixture: %s; Node IDs:\n%s', fixturedef, pprint.pformat(list(fixturedef.node_ids)))
    try:
        outcome = yield
    finally:
        for fixturedef in used_fixture_defs:
            fixturedef.node_ids.remove(item.nodeid)
            if not fixturedef.node_ids:
                # This fixture is not used in any more test functions
                log.warning('The fixture %s is not being used in any more tests, terminate it.', fixturedef)
                fixturedef.finish(item._request)
    for fixturedef in used_fixture_defs:
        log.warning(' After Test. Fixture: %s; Node IDs:\n%s', fixturedef, pprint.pformat(list(fixturedef.node_ids)))
    return outcome
# <---- PyTest Tweaks ------------------------------------------------------------------------------------------------


# ----- Test Setup -------------------------------------------------------------------------------------------------->
def _has_unittest_attr(item, attr):
    # XXX: This is a hack while we support both runtests.py and PyTest
    if hasattr(item.obj, attr):
        return True
    if item.cls and hasattr(item.cls, attr):
        return True
    if item.parent and hasattr(item.parent.obj, attr):
        return True
    return False


@pytest.hookimpl(tryfirst=True)
def pytest_runtest_setup(item):
    '''
    Fixtures injection based on markers or test skips based on CLI arguments
    '''
    destructive_tests_marker = item.get_closest_marker('destructive_test')
    if destructive_tests_marker is not None or _has_unittest_attr(item, '__destructive_test__'):
        if item.config.getoption('--run-destructive') is False:
            item._skipped_by_mark = True
            pytest.skip('Destructive tests are disabled')

    expensive_tests_marker = item.get_closest_marker('expensive_test')
    if expensive_tests_marker is not None or _has_unittest_attr(item, '__expensive_test__'):
        if item.config.getoption('--run-expensive') is False:
            item._skipped_by_mark = True
            pytest.skip('Expensive tests are disabled')

    skip_if_not_root_marker = item.get_closest_marker('skip_if_not_root')
    if skip_if_not_root_marker is not None or _has_unittest_attr(item, '__skip_if_not_root__'):
        if not sys.platform.startswith('win'):
            if os.getuid() != 0:
                item._skipped_by_mark = True
                pytest.skip('You must be logged in as root to run this test')
        else:
            current_user = salt.utils.win_functions.get_current_user()
            if current_user != 'SYSTEM':
                if not salt.utils.win_functions.is_admin(current_user):
                    item._skipped_by_mark = True
                    pytest.skip('You must be logged in as an Administrator to run this test')

    skip_if_binaries_missing_marker = item.get_closest_marker('skip_if_binaries_missing')
    if skip_if_binaries_missing_marker is not None:
        binaries = skip_if_binaries_missing_marker.args
        if len(binaries) == 1:
            if isinstance(binaries[0], (list, tuple, set, frozenset)):
                binaries = binaries[0]
        check_all = skip_if_binaries_missing_marker.kwargs.get('check_all', False)
        message = skip_if_binaries_missing_marker.kwargs.get('message', None)
        if check_all:
            for binary in binaries:
                if salt.utils.path.which(binary) is None:
                    item._skipped_by_mark = True
                    pytest.skip(
                        '{0}The "{1}" binary was not found'.format(
                            message and '{0}. '.format(message) or '',
                            binary
                        )
                    )
        elif salt.utils.path.which_bin(binaries) is None:
            item._skipped_by_mark = True
            pytest.skip(
                '{0}None of the following binaries was found: {1}'.format(
                    message and '{0}. '.format(message) or '',
                    ', '.join(binaries)
                )
            )

    requires_network_marker = item.get_closest_marker('requires_network')
    if requires_network_marker is not None:
        only_local_network = requires_network_marker.kwargs.get('only_local_network', False)
        has_local_network = False
        # First lets try if we have a local network. Inspired in verify_socket
        try:
            pubsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            retsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            pubsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            pubsock.bind(('', 18000))
            pubsock.close()
            retsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            retsock.bind(('', 18001))
            retsock.close()
            has_local_network = True
        except socket.error:
            # I wonder if we just have IPV6 support?
            try:
                pubsock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
                retsock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
                pubsock.setsockopt(
                    socket.SOL_SOCKET, socket.SO_REUSEADDR, 1
                )
                pubsock.bind(('', 18000))
                pubsock.close()
                retsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                retsock.bind(('', 18001))
                retsock.close()
                has_local_network = True
            except socket.error:
                # Let's continue
                pass

        if only_local_network is True:
            if has_local_network is False:
                # Since we're only supposed to check local network, and no
                # local network was detected, skip the test
                item._skipped_by_mark = True
                pytest.skip('No local network was detected')

        # We are using the google.com DNS records as numerical IPs to avoid
        # DNS lookups which could greatly slow down this check
        for addr in ('173.194.41.198', '173.194.41.199', '173.194.41.200',
                     '173.194.41.201', '173.194.41.206', '173.194.41.192',
                     '173.194.41.193', '173.194.41.194', '173.194.41.195',
                     '173.194.41.196', '173.194.41.197'):
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.settimeout(0.25)
                sock.connect((addr, 80))
                sock.close()
                # We connected? Stop the loop
                break
            except socket.error:
                # Let's check the next IP
                continue
            else:
                item._skipped_by_mark = True
                pytest.skip('No internet network connection was detected')

    requires_salt_modules_marker = item.get_closest_marker('requires_salt_modules')
    if requires_salt_modules_marker is not None:
        required_salt_modules = requires_salt_modules_marker.args
        if len(required_salt_modules) == 1 and isinstance(required_salt_modules[0], (list, tuple, set)):
            required_salt_modules = required_salt_modules[0]
        required_salt_modules = set(required_salt_modules)
        sminion = create_sminion()
        available_modules = list(sminion.functions)
        not_available_modules = set()
        try:
            cached_not_available_modules = sminion.__not_availiable_modules__
        except AttributeError:
            cached_not_available_modules = sminion.__not_availiable_modules__ = set()

        if cached_not_available_modules:
            for not_available_module in cached_not_available_modules:
                if not_available_module in required_salt_modules:
                    not_available_modules.add(not_available_module)
                    required_salt_modules.remove(not_available_module)

        for required_module_name in required_salt_modules:
            search_name = required_module_name
            if '.' not in search_name:
                search_name += '.*'
            if not fnmatch.filter(available_modules, search_name):
                not_available_modules.add(required_module_name)
                cached_not_available_modules.add(required_module_name)

        if not_available_modules:
            item._skipped_by_mark = True
            if len(not_available_modules) == 1:
                pytest.skip('Salt module \'{}\' is not available'.format(*not_available_modules))
            pytest.skip('Salt modules not available: {}'.format(', '.join(not_available_modules)))

    if salt.utils.platform.is_windows():
        if not item.fspath.fnmatch(os.path.join(CODE_DIR, 'tests', 'unit', '*')):
            # Unit tests are whitelisted on windows by default, so, we're only
            # after all other tests
            windows_whitelisted_marker = item.get_closest_marker('windows_whitelisted')
            if windows_whitelisted_marker is None:
                item._skipped_by_mark = True
                pytest.skip('Test is not whitelisted for Windows')
# <---- Test Setup ---------------------------------------------------------------------------------------------------


# ----- Test Groups Selection --------------------------------------------------------------------------------------->
def get_group_size(total_items, total_groups):
    '''
    Return the group size.
    '''
    return int(total_items / total_groups)


def get_group(items, group_count, group_size, group_id):
    '''
    Get the items from the passed in group based on group size.
    '''
    start = group_size * (group_id - 1)
    end = start + group_size
    total_items = len(items)

    if start >= total_items:
        pytest.fail("Invalid test-group argument. start({})>=total_items({})".format(start, total_items))
    elif start < 0:
        pytest.fail("Invalid test-group argument. Start({})<0".format(start))

    if group_count == group_id and end < total_items:
        # If this is the last group and there are still items to test
        # which don't fit in this group based on the group items count
        # add them anyway
        end = total_items

    return items[start:end]


def groups_collection_modifyitems(config, items):
    group_count = config.getoption('test-group-count')
    group_id = config.getoption('test-group')

    if not group_count or not group_id:
        # We're not selection tests using groups, don't do any filtering
        return

    total_items = len(items)

    group_size = get_group_size(total_items, group_count)
    tests_in_group = get_group(items, group_count, group_size, group_id)
    # Replace all items in the list
    items[:] = tests_in_group

    terminal_reporter = config.pluginmanager.get_plugin('terminalreporter')
    terminal_reporter.write(
        'Running test group #{0} ({1} tests)\n'.format(
            group_id,
            len(items)
        ),
        yellow=True
    )
# <---- Test Groups Selection ----------------------------------------------------------------------------------------


# ----- Pytest Helpers ---------------------------------------------------------------------------------------------->
if six.PY2:
    # backport mock_open from the python 3 unittest.mock library so that we can
    # mock read, readline, readlines, and file iteration properly

    file_spec = None

    def _iterate_read_data(read_data):
        # Helper for mock_open:
        # Retrieve lines from read_data via a generator so that separate calls to
        # readline, read, and readlines are properly interleaved
        data_as_list = ['{0}\n'.format(l) for l in read_data.split('\n')]

        if data_as_list[-1] == '\n':
            # If the last line ended in a newline, the list comprehension will have an
            # extra entry that's just a newline.  Remove this.
            data_as_list = data_as_list[:-1]
        else:
            # If there wasn't an extra newline by itself, then the file being
            # emulated doesn't have a newline to end the last line  remove the
            # newline that our naive format() added
            data_as_list[-1] = data_as_list[-1][:-1]

        for line in data_as_list:
            yield line

    @pytest.helpers.mock.register
    def mock_open(mock=None, read_data=''):
        """
        A helper function to create a mock to replace the use of `open`. It works
        for `open` called directly or used as a context manager.

        The `mock` argument is the mock object to configure. If `None` (the
        default) then a `MagicMock` will be created for you, with the API limited
        to methods or attributes available on standard file handles.

        `read_data` is a string for the `read` methoddline`, and `readlines` of the
        file handle to return.  This is an empty string by default.
        """
        _mock = pytest.importorskip('mock', minversion='2.0.0')

        # pylint: disable=unused-argument
        def _readlines_side_effect(*args, **kwargs):
            if handle.readlines.return_value is not None:
                return handle.readlines.return_value
            return list(_data)

        def _read_side_effect(*args, **kwargs):
            if handle.read.return_value is not None:
                return handle.read.return_value
            return ''.join(_data)
        # pylint: enable=unused-argument

        def _readline_side_effect():
            if handle.readline.return_value is not None:
                while True:
                    yield handle.readline.return_value
            for line in _data:
                yield line

        global file_spec  # pylint: disable=global-statement
        if file_spec is None:
            file_spec = file  # pylint: disable=undefined-variable

        if mock is None:
            mock = _mock.MagicMock(name='open', spec=open)

        handle = _mock.MagicMock(spec=file_spec)
        handle.__enter__.return_value = handle

        _data = _iterate_read_data(read_data)

        handle.write.return_value = None
        handle.read.return_value = None
        handle.readline.return_value = None
        handle.readlines.return_value = None

        handle.read.side_effect = _read_side_effect
        handle.readline.side_effect = _readline_side_effect()
        handle.readlines.side_effect = _readlines_side_effect

        mock.return_value = handle
        return mock
else:
    @pytest.helpers.mock.register
    def mock_open(mock=None, read_data=''):
        _mock = pytest.importorskip('mock', minversion='2.0.0')
        return _mock.mock_open(mock=mock, read_data=read_data)


@pytest.helpers.register
@contextmanager
def temp_directory(name=None):
    if name is not None:
        directory_path = os.path.join(RUNTIME_VARS.TMP, name)
    else:
        directory_path = tempfile.mkdtemp(dir=RUNTIME_VARS.TMP)

    yield directory_path

    shutil.rmtree(directory_path, ignore_errors=True)


@pytest.helpers.register
@contextmanager
def temp_file(name, contents=None, directory=None, strip_first_newline=True):
    if directory is None:
        directory = RUNTIME_VARS.TMP

    file_path = os.path.join(directory, name)
    file_directory = os.path.dirname(file_path)
    if contents is not None:
        if contents:
            if contents.startswith('\n') and strip_first_newline:
                contents = contents[1:]
            file_contents = textwrap.dedent(contents)
        else:
            file_contents = contents

    try:
        if not os.path.isdir(file_directory):
            os.makedirs(file_directory)
        if contents is not None:
            with salt.utils.files.fopen(file_path, 'w') as wfh:
                wfh.write(file_contents)

        yield file_path

    finally:
        try:
            os.unlink(file_path)
        except OSError:
            # Already deleted
            pass


@pytest.helpers.register
def temp_state_file(name, contents, saltenv='base', strip_first_newline=True):

    if saltenv == 'base':
        directory = RUNTIME_VARS.TMP_STATE_TREE
    elif saltenv == 'prod':
        directory = RUNTIME_VARS.TMP_PRODENV_STATE_TREE
    else:
        raise RuntimeError('"saltenv" can only be "base" or "prod", not "{}"'.format(saltenv))
    return temp_file(name, contents, directory=directory, strip_first_newline=strip_first_newline)
# <---- Pytest Helpers -----------------------------------------------------------------------------------------------


# ----- Fixtures Overrides ------------------------------------------------------------------------------------------>
# ----- Generate CLI Scripts ---------------------------------------------------------------------------------------->
@pytest.fixture(scope='session')
def cli_master_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_master.py'


@pytest.fixture(scope='session')
def cli_minion_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_minion.py'


@pytest.fixture(scope='session')
def cli_salt_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt.py'


@pytest.fixture(scope='session')
def cli_run_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_run.py'


@pytest.fixture(scope='session')
def cli_key_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_key.py'


@pytest.fixture(scope='session')
def cli_call_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_call.py'


@pytest.fixture(scope='session')
def cli_syndic_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_syndic.py'


@pytest.fixture(scope='session')
def cli_ssh_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_ssh.py'


@pytest.fixture(scope='session')
def cli_proxy_script_name():
    '''
    Return the CLI script basename
    '''
    return 'cli_salt_proxy.py'


@pytest.fixture(scope='session')
def cli_bin_dir(tempdir,
                request,
                python_executable_path,
                cli_master_script_name,
                cli_minion_script_name,
                cli_salt_script_name,
                cli_call_script_name,
                cli_key_script_name,
                cli_run_script_name,
                cli_ssh_script_name,
                cli_syndic_script_name,
                cli_proxy_script_name):
    '''
    Return the path to the CLI script directory to use
    '''
    tmp_cli_scripts_dir = tempdir.join('cli-scrips-bin')
    # Make sure we re-write the scripts every time we start the tests
    shutil.rmtree(tmp_cli_scripts_dir.strpath, ignore_errors=True)
    tmp_cli_scripts_dir.ensure(dir=True)
    cli_bin_dir_path = tmp_cli_scripts_dir.strpath

    # Now that we have the CLI directory created, lets generate the required CLI scripts to run salt's test suite
    for script_name in (cli_master_script_name,
                        cli_minion_script_name,
                        cli_call_script_name,
                        cli_key_script_name,
                        cli_run_script_name,
                        cli_salt_script_name,
                        cli_ssh_script_name,
                        cli_syndic_script_name,
                        cli_proxy_script_name):
        original_script_name = os.path.splitext(script_name)[0].split('cli_')[-1].replace('_', '-')
        cli_scripts.generate_script(
            bin_dir=cli_bin_dir_path,
            script_name=original_script_name,
            executable=sys.executable,
            code_dir=CODE_DIR,
            inject_sitecustomize=MAYBE_RUN_COVERAGE
        )

    # Return the CLI bin dir value
    return cli_bin_dir_path
# <---- Generate CLI Scripts -----------------------------------------------------------------------------------------


# ----- Salt Configuration ------------------------------------------------------------------------------------------>
@pytest.fixture(scope='session')
def session_master_of_masters_id():
    '''
    Returns the master of masters id
    '''
    return 'syndic_master'


@pytest.fixture(scope='session')
def session_master_id():
    '''
    Returns the session scoped master id
    '''
    return 'master'


@pytest.fixture(scope='session')
def session_minion_id():
    '''
    Returns the session scoped minion id
    '''
    return 'minion'


@pytest.fixture(scope='session')
def session_secondary_minion_id():
    '''
    Returns the session scoped secondary minion id
    '''
    return 'sub_minion'


@pytest.fixture(scope='session')
def session_syndic_id():
    '''
    Returns the session scoped syndic id
    '''
    return 'syndic'


@pytest.fixture(scope='session')
def session_proxy_id():
    '''
    Returns the session scoped proxy id
    '''
    return 'proxytest'


@pytest.fixture(scope='session')
def salt_fail_hard():
    '''
    Return the salt fail hard value
    '''
    return True


@pytest.fixture(scope='session')
def session_master_default_options(request, session_root_dir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'master')) as rfh:
        opts = yaml.deserialize(rfh.read())

        tests_known_hosts_file = session_root_dir.join('salt_ssh_known_hosts').strpath
        with salt.utils.files.fopen(tests_known_hosts_file, 'w') as known_hosts:
            known_hosts.write('')

        opts['known_hosts_file'] = tests_known_hosts_file
        opts['syndic_master'] = 'localhost'
        opts['transport'] = request.config.getoption('--transport')

        # Config settings to test `event_return`
        if 'returner_dirs' not in opts:
            opts['returner_dirs'] = []
        opts['returner_dirs'].append(os.path.join(RUNTIME_VARS.FILES, 'returners'))
        opts['event_return'] = 'runtests_noop'

        return opts


@pytest.fixture(scope='session')
def session_master_config_overrides(session_root_dir):
    ext_pillar = []
    if salt.utils.platform.is_windows():
        ext_pillar.append(
            {'cmd_yaml': 'type {0}'.format(os.path.join(RUNTIME_VARS.FILES, 'ext.yaml'))}
        )
    else:
        ext_pillar.append(
            {'cmd_yaml': 'cat {0}'.format(os.path.join(RUNTIME_VARS.FILES, 'ext.yaml'))}
        )
    ext_pillar.append(
        {
            'file_tree': {
                'root_dir':  os.path.join(RUNTIME_VARS.PILLAR_DIR, 'base', 'file_tree'),
                'follow_dir_links': False,
                'keep_newline': True
            }
        }
    )

    # We need to copy the extension modules into the new master root_dir or
    # it will be prefixed by it
    extension_modules_path = session_root_dir.join('extension_modules').strpath
    if not os.path.exists(extension_modules_path):
        shutil.copytree(
            os.path.join(
                RUNTIME_VARS.FILES, 'extension_modules'
            ),
            extension_modules_path
        )

    # Copy the autosign_file to the new  master root_dir
    autosign_file_path = session_root_dir.join('autosign_file').strpath
    shutil.copyfile(
        os.path.join(RUNTIME_VARS.FILES, 'autosign_file'),
        autosign_file_path
    )
    # all read, only owner write
    autosign_file_permissions = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH | stat.S_IWUSR
    os.chmod(autosign_file_path, autosign_file_permissions)

    pytest_stop_sending_events_file = session_root_dir.join('pytest_stop_sending_events_file').strpath
    with salt.utils.files.fopen(pytest_stop_sending_events_file, 'w') as wfh:
        wfh.write('')

    return {
        'pillar_opts': True,
        'ext_pillar': ext_pillar,
        'extension_modules': extension_modules_path,
        'file_roots': {
            'base': [
                os.path.join(RUNTIME_VARS.FILES, 'file', 'base'),
            ],
            # Alternate root to test __env__ choices
            'prod': [
                os.path.join(RUNTIME_VARS.FILES, 'file', 'prod'),
            ]
        },
        'pillar_roots': {
            'base': [
                os.path.join(RUNTIME_VARS.FILES, 'pillar', 'base'),
            ]
        },
        'reactor': [
            {
                'salt/minion/*/start': [
                    os.path.join(RUNTIME_VARS.FILES, 'reactor-sync-minion.sls')
                ],
            },
            {
                'salt/test/reactor': [
                    os.path.join(RUNTIME_VARS.FILES, 'reactor-test.sls')
                ],
            }
        ],
        'pytest_stop_sending_events_file': pytest_stop_sending_events_file
    }


@pytest.fixture(scope='session')
def session_minion_default_options(request, tempdir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'minion')) as rfh:
        opts = yaml.deserialize(rfh.read())

        opts['hosts.file'] = tempdir.join('hosts').strpath
        opts['aliases.file'] = tempdir.join('aliases').strpath
        opts['transport'] = request.config.getoption('--transport')

        return opts


def _get_virtualenv_binary_path():
    try:
        return _get_virtualenv_binary_path.__virtualenv_binary__
    except AttributeError:
        # Under windows we can't seem to properly create a virtualenv off of another
        # virtualenv, we can on linux but we will still point to the virtualenv binary
        # outside the virtualenv running the test suite, if that's the case.
        try:
            real_prefix = sys.real_prefix
            # The above attribute exists, this is a virtualenv
            if salt.utils.platform.is_windows():
                virtualenv_binary = os.path.join(real_prefix, 'Scripts', 'virtualenv.exe')
            else:
                # We need to remove the virtualenv from PATH or we'll get the virtualenv binary
                # from within the virtualenv, we don't want that
                path = os.environ.get('PATH')
                if path is not None:
                    path_items = path.split(os.pathsep)
                    for item in path_items[:]:
                        if item.startswith(sys.base_prefix):
                            path_items.remove(item)
                    os.environ['PATH'] = os.pathsep.join(path_items)
                virtualenv_binary = salt.utils.path.which('virtualenv')
                if path is not None:
                    # Restore previous environ PATH
                    os.environ['PATH'] = path
                if not virtualenv_binary.startswith(real_prefix):
                    virtualenv_binary = None
            if virtualenv_binary and not os.path.exists(virtualenv_binary):
                # It doesn't exist?!
                virtualenv_binary = None
        except AttributeError:
            # We're not running inside a virtualenv
            virtualenv_binary = None
        _get_virtualenv_binary_path.__virtualenv_binary__ = virtualenv_binary
        return virtualenv_binary


@pytest.fixture(scope='session')
def session_minion_config_overrides():
    opts = {
        'file_roots': {
            'base': [
                os.path.join(RUNTIME_VARS.FILES, 'file', 'base'),
            ],
            # Alternate root to test __env__ choices
            'prod': [
                os.path.join(RUNTIME_VARS.FILES, 'file', 'prod'),
            ]
        },
        'pillar_roots': {
            'base': [
                os.path.join(RUNTIME_VARS.FILES, 'pillar', 'base'),
            ]
        },
    }
    virtualenv_binary = _get_virtualenv_binary_path()
    if virtualenv_binary:
        opts['venv_bin'] = virtualenv_binary
    return opts


@pytest.fixture(scope='session')
def session_secondary_minion_default_options(request, tempdir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'sub_minion')) as rfh:
        opts = yaml.deserialize(rfh.read())

        opts['hosts.file'] = tempdir.join('hosts').strpath
        opts['aliases.file'] = tempdir.join('aliases').strpath
        opts['transport'] = request.config.getoption('--transport')

        return opts


@pytest.fixture(scope='session')
def session_seconary_minion_config_overrides():
    opts = {}
    virtualenv_binary = _get_virtualenv_binary_path()
    if virtualenv_binary:
        opts['venv_bin'] = virtualenv_binary
    return opts


@pytest.fixture(scope='session')
def session_master_of_masters_default_options(request, tempdir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'syndic_master')) as rfh:
        opts = yaml.deserialize(rfh.read())

        opts['hosts.file'] = tempdir.join('hosts').strpath
        opts['aliases.file'] = tempdir.join('aliases').strpath
        opts['transport'] = request.config.getoption('--transport')

        return opts


@pytest.fixture(scope='session')
def session_master_of_masters_config_overrides(session_master_of_masters_root_dir):
    if salt.utils.platform.is_windows():
        ext_pillar = {'cmd_yaml': 'type {0}'.format(os.path.join(RUNTIME_VARS.FILES, 'ext.yaml'))}
    else:
        ext_pillar = {'cmd_yaml': 'cat {0}'.format(os.path.join(RUNTIME_VARS.FILES, 'ext.yaml'))}

    # We need to copy the extension modules into the new master root_dir or
    # it will be prefixed by it
    extension_modules_path = session_master_of_masters_root_dir.join('extension_modules').strpath
    if not os.path.exists(extension_modules_path):
        shutil.copytree(
            os.path.join(
                RUNTIME_VARS.FILES, 'extension_modules'
            ),
            extension_modules_path
        )

    # Copy the autosign_file to the new  master root_dir
    autosign_file_path = session_master_of_masters_root_dir.join('autosign_file').strpath
    shutil.copyfile(
        os.path.join(RUNTIME_VARS.FILES, 'autosign_file'),
        autosign_file_path
    )
    # all read, only owner write
    autosign_file_permissions = stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH | stat.S_IWUSR
    os.chmod(autosign_file_path, autosign_file_permissions)

    pytest_stop_sending_events_file = session_master_of_masters_root_dir.join('pytest_stop_sending_events_file').strpath
    with salt.utils.files.fopen(pytest_stop_sending_events_file, 'w') as wfh:
        wfh.write('')

    return {
        'ext_pillar': [ext_pillar],
        'extension_modules': extension_modules_path,
        'file_roots': {
            'base': [
                os.path.join(RUNTIME_VARS.FILES, 'file', 'base'),
            ],
            # Alternate root to test __env__ choices
            'prod': [
                os.path.join(RUNTIME_VARS.FILES, 'file', 'prod'),
            ]
        },
        'pillar_roots': {
            'base': [
                os.path.join(RUNTIME_VARS.FILES, 'pillar', 'base'),
            ]
        },
        'pytest_stop_sending_events_file': pytest_stop_sending_events_file
    }


@pytest.fixture(scope='session')
def session_syndic_master_default_options(request, tempdir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'syndic_master')) as rfh:
        opts = yaml.deserialize(rfh.read())

        opts['hosts.file'] = tempdir.join('hosts').strpath
        opts['aliases.file'] = tempdir.join('aliases').strpath
        opts['transport'] = request.config.getoption('--transport')

        return opts


@pytest.fixture(scope='session')
def session_syndic_default_options(request, tempdir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'syndic')) as rfh:
        opts = yaml.deserialize(rfh.read())

        opts['hosts.file'] = tempdir.join('hosts').strpath
        opts['aliases.file'] = tempdir.join('aliases').strpath
        opts['transport'] = request.config.getoption('--transport')

        return opts


@pytest.fixture(scope='session')
def session_proxy_default_options(request, tempdir):
    with salt.utils.files.fopen(os.path.join(RUNTIME_VARS.CONF_DIR, 'proxy')) as rfh:
        opts = yaml.deserialize(rfh.read())

        opts['hosts.file'] = tempdir.join('hosts').strpath
        opts['aliases.file'] = tempdir.join('aliases').strpath
        opts['transport'] = request.config.getoption('--transport')

        return opts


@pytest.fixture(scope='session', autouse=True)
def bridge_pytest_and_runtests(reap_stray_processes,
                               session_root_dir,
                               session_conf_dir,
                               session_secondary_conf_dir,
                               session_syndic_conf_dir,
                               session_master_of_masters_conf_dir,
                               session_base_env_pillar_tree_root_dir,
                               session_base_env_state_tree_root_dir,
                               session_prod_env_state_tree_root_dir,
                               session_master_config,
                               session_minion_config,
                               session_secondary_minion_config,
                               session_master_of_masters_config,
                               session_syndic_config):

    # Make sure unittest2 classes know their paths
    RUNTIME_VARS.TMP_ROOT_DIR = session_root_dir.realpath().strpath
    RUNTIME_VARS.TMP_CONF_DIR = session_conf_dir.realpath().strpath
    RUNTIME_VARS.TMP_SUB_MINION_CONF_DIR = session_secondary_conf_dir.realpath().strpath
    RUNTIME_VARS.TMP_SYNDIC_MASTER_CONF_DIR = session_master_of_masters_conf_dir.realpath().strpath
    RUNTIME_VARS.TMP_SYNDIC_MINION_CONF_DIR = session_syndic_conf_dir.realpath().strpath
    RUNTIME_VARS.TMP_PILLAR_TREE = session_base_env_pillar_tree_root_dir.realpath().strpath
    RUNTIME_VARS.TMP_STATE_TREE = session_base_env_state_tree_root_dir.realpath().strpath
    RUNTIME_VARS.TMP_PRODENV_STATE_TREE = session_prod_env_state_tree_root_dir.realpath().strpath

    # Make sure unittest2 uses the pytest generated configuration
    RUNTIME_VARS.RUNTIME_CONFIGS['master'] = freeze(session_master_config)
    RUNTIME_VARS.RUNTIME_CONFIGS['minion'] = freeze(session_minion_config)
    RUNTIME_VARS.RUNTIME_CONFIGS['sub_minion'] = freeze(session_secondary_minion_config)
    RUNTIME_VARS.RUNTIME_CONFIGS['syndic_master'] = freeze(session_master_of_masters_config)
    RUNTIME_VARS.RUNTIME_CONFIGS['syndic'] = freeze(session_syndic_config)
    RUNTIME_VARS.RUNTIME_CONFIGS['client_config'] = freeze(
        salt.config.client_config(session_conf_dir.join('master').strpath)
    )

    # Copy configuration files and directories which are not automatically generated
    for entry in os.listdir(RUNTIME_VARS.CONF_DIR):
        if entry in ('master', 'minion', 'sub_minion', 'syndic', 'syndic_master', 'proxy'):
            # These have runtime computed values and are handled by pytest-salt fixtures
            continue
        entry_path = os.path.join(RUNTIME_VARS.CONF_DIR, entry)
        if os.path.isfile(entry_path):
            shutil.copy(
                entry_path,
                os.path.join(RUNTIME_VARS.TMP_CONF_DIR, entry)
            )
        elif os.path.isdir(entry_path):
            shutil.copytree(
                entry_path,
                os.path.join(RUNTIME_VARS.TMP_CONF_DIR, entry)
            )
# <---- Salt Configuration -------------------------------------------------------------------------------------------
# <---- Fixtures Overrides -------------------------------------------------------------------------------------------


# ----- Custom Grains Mark Evaluator -------------------------------------------------------------------------------->
class GrainsMarkEvaluator(MarkEvaluator):
    _cached_grains = None

    def _getglobals(self):
        item_globals = super(GrainsMarkEvaluator, self)._getglobals()
        if GrainsMarkEvaluator._cached_grains is None:
            sminion = create_sminion()
            GrainsMarkEvaluator._cached_grains = sminion.opts['grains'].copy()
        item_globals['grains'] = GrainsMarkEvaluator._cached_grains.copy()
        return item_globals


# Patch PyTest's skipping MarkEvaluator to use our GrainsMarkEvaluator
_pytest.skipping.MarkEvaluator = GrainsMarkEvaluator
# <---- Custom Grains Mark Evaluator ---------------------------------------------------------------------------------


# ----- Custom Fixtures --------------------------------------------------------------------------------------------->
@pytest.fixture(scope='session')
def reap_stray_processes():
    # Run tests
    yield

    children = psutil.Process(os.getpid()).children(recursive=True)
    if not children:
        log.info('No astray processes found')
        return

    def on_terminate(proc):
        log.debug('Process %s terminated with exit code %s', proc, proc.returncode)

    if children:
        # Reverse the order, sublings first, parents after
        children.reverse()
        log.warning(
            'Test suite left %d astray processes running. Killing those processes:\n%s',
            len(children),
            pprint.pformat(children)
        )

        _, alive = psutil.wait_procs(children, timeout=3, callback=on_terminate)
        for child in alive:
            child.kill()

        _, alive = psutil.wait_procs(alive, timeout=3, callback=on_terminate)
        if alive:
            # Give up
            for child in alive:
                log.warning('Process %s survived SIGKILL, giving up:\n%s', child, pprint.pformat(child.as_dict()))


@pytest.fixture(scope='session')
def grains(request):
    sminion = create_sminion()
    return sminion.opts['grains'].copy()
# <---- Custom Fixtures ----------------------------------------------------------------------------------------------