Increase version to 1.7.3

[pyi2ncommon] / test / test_arnied_wrapper.py

#!/usr/bin/env python

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

import unittest
import unittest.mock as mock
import time

from src import arnied_wrapper


class DummyCmdOutputMapping:
    """
    Class to replace the :py:function:`arnied_wrapper.run_cmd` function.

    In the arnied_wrapper, when running a command, instead of calling the actual
    function, this class' constructor will be invoked and an instance returned
    instead. It stubs :py:class:`subprocess.CompletedProcess`, such that the
    returncode and stdout attributes are set depending on the command or on the
    test.
    """

    # whether to return 1 as a fail indicator
    fail_switch = False
    # mapping between expected commands and their mocked output + return code
    asserted_cmds = []

    def __init__(self, cmd="", ignore_errors=False, vm=None, timeout=60):
        """
        Class constructor to mimic the run function of the arnied wrapper.

        Arguments are the same of the mocked function.
        """
        self.returncode = 0
        self.stdout = b""
        self._get_result(cmd)
        if self.fail_switch:
            self.returncode = 1

    def __str__(self):
        """String representation of this class."""
        return "status %i, stdout %s" % (self.returncode, self.stdout)

    def _get_result(self, cmd):
        """
        Return the first (or raise) values in the mapping matching the command.

        :param str cmd: command to check the mapping against
        :raises: :py:class:`ValueError` if the command has no corresponding
                 mapping
        :returns: this instance with the return code and stdout attributes set
        :rtype: :py:class:`DummyCmdOutputMapping`
        """
        for dummy_cmd in self.asserted_cmds:
            if dummy_cmd['cmd'] == cmd:
                self.returncode = dummy_cmd['returncode']
                self.stdout = dummy_cmd['stdout']
                return
        raise ValueError("Could not locate the command '%s' among the known answers "
                         "for the universe" % cmd)


# make sure that invoking `run_cmd` returns an instance of DummyCmdOutputMapping
@mock.patch("src.arnied_wrapper.run_cmd", DummyCmdOutputMapping)
class ArniedWrapperTest(unittest.TestCase):
    def setUp(self):
        DummyCmdOutputMapping.fail_switch = False
        DummyCmdOutputMapping.asserted_cmds = []

    def test_verify_running(self):
        """Test checking for running programs."""
        DummyCmdOutputMapping.asserted_cmds = [{"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0}]
        arnied_wrapper.verify_running(timeout=1)
        DummyCmdOutputMapping.fail_switch = True
        with self.assertRaises(RuntimeError):
            arnied_wrapper.verify_running(timeout=1)

    def test_wait_for_arnied(self):
        """Test waiting for arnied to be ready."""
        DummyCmdOutputMapping.asserted_cmds = [{"cmd": "/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 10", "stdout": b"", "returncode": 0}]
        arnied_wrapper.wait_for_arnied(timeout=10)

    def test_go_online(self):
        DummyCmdOutputMapping.asserted_cmds = [{"cmd": 'tell-connd --online P1', "stdout": b"", "returncode": 0},
                                               {"cmd": '/usr/intranator/bin/get_var ONLINE', "stdout": b"DEFAULT: 2", "returncode": 0}]
        arnied_wrapper.go_online(1)

    def test_email_transfer(self):
        """Test e-mail transferring."""
        DummyCmdOutputMapping.asserted_cmds = [{"cmd": '/usr/intranator/bin/arnied_helper --transfer-mail', "stdout": b"", "returncode": 0}]
        arnied_wrapper.email_transfer()

    def test_wait_for_no_generate(self):
        """Test waiting for generate if there is no generate scheduled"""
        DummyCmdOutputMapping.asserted_cmds = [
            {"cmd": f'/usr/intranator/bin/arnied_helper --is-scheduled-or-running {job}', "stdout": b"", "returncode": 1}
                for job in ("GENERATE", "GENERATE_OFFLINE")]
        start_time = time.time()
        self.assertEqual(arnied_wrapper.wait_for_generate(),  True)
        self.assertLess(time.time() - start_time, 1)    # no sleep, just call and return at once

    def test_wait_for_generate(self):
        """Test waiting for generate if there is generate scheduled"""
        DummyCmdOutputMapping.asserted_cmds = [
            {"cmd": f'/usr/intranator/bin/arnied_helper --is-scheduled-or-running {job}', "stdout": b"", "returncode": 0}
            for job in ("GENERATE", "GENERATE_OFFLINE")]
        start_time = time.time()
        self.assertEqual(arnied_wrapper.wait_for_generate(timeout=2),  False)
        end_time = time.time()
        self.assertGreater(end_time - start_time, 1.5)
        self.assertLess(end_time - start_time, 2.5)

    def test_wait_for_email_transfer_timeout(self):
        """Test waiting for email transfer that reaches timeout."""
        DummyCmdOutputMapping.asserted_cmds = (
            {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0},
            {"cmd": 'postqueue -j', "stdout": b"not empty", "returncode": 0},
        )
        start_time = time.time()
        self.assertRaises(TimeoutError, arnied_wrapper.wait_for_email_transfer, 2)
        time_diff = time.time() - start_time
        self.assertGreater(time_diff, 1.5)
        self.assertLess(time_diff, 2.5)

    def test_wait_for_email_transfer_succeed(self):
        """Test waiting for email transfer that succeeds at once"""
        DummyCmdOutputMapping.asserted_cmds = (
            {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0},
            {"cmd": 'postqueue -j', "stdout": b"", "returncode": 0},
        )
        start_time = time.time()
        arnied_wrapper.wait_for_email_transfer()
        time_diff = time.time() - start_time
        self.assertLess(time_diff, 1)