salt/tests/unit/utils/test_systemd.py

# -*- coding: utf-8 -*-

# Import python libs
from __future__ import absolute_import, print_function, unicode_literals

import errno
import os

# Import Salt libs
import salt.utils.systemd as _systemd
from salt.exceptions import SaltInvocationError
from tests.support.mock import Mock, patch

# Import Salt Testing libs
from tests.support.unit import TestCase


def _booted_effect(path):
    return True if path == "/run/systemd/system" else os.stat(path)


def _not_booted_effect(path):
    if path == "/run/systemd/system":
        raise OSError(errno.ENOENT, "No such file or directory", path)
    return os.stat(path)


class SystemdTestCase(TestCase):
    """
    Tests the functions in salt.utils.systemd
    """

    def test_booted(self):
        """
        Test that salt.utils.systemd.booted() returns True when minion is
        systemd-booted.
        """
        # Ensure that os.stat returns True. os.stat doesn't return a bool
        # normally, but the code is doing a simple truth check on the return
        # data, so it is sufficient enough to mock it as True for these tests.
        with patch("os.stat", side_effect=_booted_effect):
            # Test without context dict passed
            self.assertTrue(_systemd.booted())
            # Test that context key is set when context dict is passed
            context = {}
            self.assertTrue(_systemd.booted(context))
            self.assertEqual(context, {"salt.utils.systemd.booted": True})

    def test_not_booted(self):
        """
        Test that salt.utils.systemd.booted() returns False when minion is not
        systemd-booted.
        """
        # Ensure that os.stat raises an exception even if test is being run on
        # a systemd-booted host.
        with patch("os.stat", side_effect=_not_booted_effect):
            # Test without context dict passed
            self.assertFalse(_systemd.booted())
            # Test that context key is set when context dict is passed
            context = {}
            self.assertFalse(_systemd.booted(context))
            self.assertEqual(context, {"salt.utils.systemd.booted": False})

    def test_booted_return_from_context(self):
        """
        Test that the context data is returned when present. To ensure we're
        getting data from the context dict, we use a non-boolean value to
        differentiate it from the True/False return this function normally
        produces.
        """
        context = {"salt.utils.systemd.booted": "foo"}
        self.assertEqual(_systemd.booted(context), "foo")

    def test_booted_invalid_context(self):
        """
        Test with invalid context data. The context value must be a dict, so
        this should raise a SaltInvocationError.
        """
        # Test with invalid context data
        with self.assertRaises(SaltInvocationError):
            _systemd.booted(99999)

    def test_version(self):
        """
        Test that salt.utils.systemd.booted() returns True when minion is
        systemd-booted.
        """
        with patch("subprocess.Popen") as popen_mock:
            _version = 231
            output = "systemd {0}\n-SYSVINIT".format(_version)
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: (output, None),
                pid=lambda: 12345,
                retcode=0,
            )

            # Test without context dict passed
            self.assertEqual(_systemd.version(), _version)
            # Test that context key is set when context dict is passed
            context = {}
            self.assertTrue(_systemd.version(context))
            self.assertEqual(context, {"salt.utils.systemd.version": _version})

    def test_version_generated_from_git_describe(self):
        """
        Test with version string matching versions generated by git describe
        in systemd. This feature is used in systemd>=241.
        """
        with patch("subprocess.Popen") as popen_mock:
            _version = 241
            output = "systemd {0} ({0}.0-0-dist)\n-SYSVINIT".format(_version)
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: (output, None),
                pid=lambda: 12345,
                retcode=0,
            )

            # Test without context dict passed
            self.assertEqual(_systemd.version(), _version)
            # Test that context key is set when context dict is passed
            context = {}
            self.assertTrue(_systemd.version(context))
            self.assertEqual(context, {"salt.utils.systemd.version": _version})

    def test_version_return_from_context(self):
        """
        Test that the context data is returned when present. To ensure we're
        getting data from the context dict, we use a non-integer value to
        differentiate it from the integer return this function normally
        produces.
        """
        context = {"salt.utils.systemd.version": "foo"}
        self.assertEqual(_systemd.version(context), "foo")

    def test_version_invalid_context(self):
        """
        Test with invalid context data. The context value must be a dict, so
        this should raise a SaltInvocationError.
        """
        # Test with invalid context data
        with self.assertRaises(SaltInvocationError):
            _systemd.version(99999)

    def test_version_parse_problem(self):
        """
        Test with invalid context data. The context value must be a dict, so
        this should raise a SaltInvocationError.
        """
        with patch("subprocess.Popen") as popen_mock:
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: ("invalid", None),
                pid=lambda: 12345,
                retcode=0,
            )
            # Test without context dict passed
            self.assertIsNone(_systemd.version())
            # Test that context key is set when context dict is passed. A failure
            # to parse the systemctl output should not set a context key, so it
            # should not be present in the context dict.
            context = {}
            self.assertIsNone(_systemd.version(context))
            self.assertEqual(context, {})

    def test_has_scope_systemd204(self):
        """
        Scopes are available in systemd>=205. Make sure that this function
        returns the expected boolean. We do three separate unit tests for
        versions 204 through 206 because mock doesn't like us altering the
        return_value in a loop.
        """
        with patch("subprocess.Popen") as popen_mock:
            _expected = False
            _version = 204
            _output = "systemd {0}\n-SYSVINIT".format(_version)
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: (_output, None),
                pid=lambda: 12345,
                retcode=0,
            )
            # Ensure that os.stat returns True. os.stat doesn't return a bool
            # normally, but the code is doing a simple truth check on the
            # return data, so it is sufficient enough to mock it as True for
            # these tests.
            with patch("os.stat", side_effect=_booted_effect):
                # Test without context dict passed
                self.assertEqual(_systemd.has_scope(), _expected)
                # Test that context key is set when context dict is passed
                context = {}
                self.assertEqual(_systemd.has_scope(context), _expected)
                self.assertEqual(
                    context,
                    {
                        "salt.utils.systemd.booted": True,
                        "salt.utils.systemd.version": _version,
                    },
                )

    def test_has_scope_systemd205(self):
        """
        Scopes are available in systemd>=205. Make sure that this function
        returns the expected boolean. We do three separate unit tests for
        versions 204 through 206 because mock doesn't like us altering the
        return_value in a loop.
        """
        with patch("subprocess.Popen") as popen_mock:
            _expected = True
            _version = 205
            _output = "systemd {0}\n-SYSVINIT".format(_version)
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: (_output, None),
                pid=lambda: 12345,
                retcode=0,
            )
            # Ensure that os.stat returns True. os.stat doesn't return a bool
            # normally, but the code is doing a simple truth check on the
            # return data, so it is sufficient enough to mock it as True for
            # these tests.
            with patch("os.stat", side_effect=_booted_effect):
                # Test without context dict passed
                self.assertEqual(_systemd.has_scope(), _expected)
                # Test that context key is set when context dict is passed
                context = {}
                self.assertEqual(_systemd.has_scope(context), _expected)
                self.assertEqual(
                    context,
                    {
                        "salt.utils.systemd.booted": True,
                        "salt.utils.systemd.version": _version,
                    },
                )

    def test_has_scope_systemd206(self):
        """
        Scopes are available in systemd>=205. Make sure that this function
        returns the expected boolean. We do three separate unit tests for
        versions 204 through 206 because mock doesn't like us altering the
        return_value in a loop.
        """
        with patch("subprocess.Popen") as popen_mock:
            _expected = True
            _version = 206
            _output = "systemd {0}\n-SYSVINIT".format(_version)
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: (_output, None),
                pid=lambda: 12345,
                retcode=0,
            )
            # Ensure that os.stat returns True. os.stat doesn't return a bool
            # normally, but the code is doing a simple truth check on the
            # return data, so it is sufficient enough to mock it as True for
            # these tests.
            with patch("os.stat", side_effect=_booted_effect):
                # Test without context dict passed
                self.assertEqual(_systemd.has_scope(), _expected)
                # Test that context key is set when context dict is passed
                context = {}
                self.assertEqual(_systemd.has_scope(context), _expected)
                self.assertEqual(
                    context,
                    {
                        "salt.utils.systemd.booted": True,
                        "salt.utils.systemd.version": _version,
                    },
                )

    def test_has_scope_no_systemd(self):
        """
        Test the case where the system is not systemd-booted. We should not be
        performing a version check in these cases as there is no need.
        """
        with patch("os.stat", side_effect=_not_booted_effect):
            # Test without context dict passed
            self.assertFalse(_systemd.has_scope())
            # Test that context key is set when context dict is passed.
            # Because we are not systemd-booted, there should be no key in the
            # context dict for the version check, as we shouldn't have
            # performed this check.
            context = {}
            self.assertFalse(_systemd.has_scope(context))
            self.assertEqual(context, {"salt.utils.systemd.booted": False})

    def test_has_scope_version_parse_problem(self):
        """
        Test the case where the system is systemd-booted, but we failed to
        parse the "systemctl --version" output.
        """
        with patch("subprocess.Popen") as popen_mock:
            popen_mock.return_value = Mock(
                communicate=lambda *args, **kwargs: ("invalid", None),
                pid=lambda: 12345,
                retcode=0,
            )
            with patch("os.stat", side_effect=_booted_effect):
                # Test without context dict passed
                self.assertFalse(_systemd.has_scope())
                # Test that context key is set when context dict is passed. A
                # failure to parse the systemctl output should not set a context
                # key, so it should not be present in the context dict.
                context = {}
                self.assertFalse(_systemd.has_scope(context))
                self.assertEqual(context, {"salt.utils.systemd.booted": True})

    def test_has_scope_invalid_context(self):
        """
        Test with invalid context data. The context value must be a dict, so
        this should raise a SaltInvocationError.
        """
        # Test with invalid context data
        with self.assertRaises(SaltInvocationError):
            _systemd.has_scope(99999)