salt/tests/unit/modules/test_vmctl.py

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

# Import Python libs
from __future__ import absolute_import

# Import Salt Libs
import salt.modules.vmctl as vmctl

# Import Salt Testing Libs
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.unit import TestCase


class VmctlTestCase(TestCase, LoaderModuleMockMixin):
    """
    test modules.vmctl functions
    """

    def setup_loader_modules(self):
        return {vmctl: {}}

    def test_create_disk(self):
        """
        Tests creating a new disk image.
        """
        ret = {}
        ret["stdout"] = "vmctl: imagefile created"
        ret["stderr"] = ""
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            self.assertTrue(vmctl.create_disk("/path/to/disk.img", "1G"))

    def test_load(self):
        """
        Tests loading a configuration file.
        """
        ret = {}
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            self.assertTrue(vmctl.load("/etc/vm.switches.conf"))

    def test_reload(self):
        """
        Tests reloading the configuration.
        """
        ret = {}
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            self.assertTrue(vmctl.reload())

    def test_reset(self):
        """
        Tests resetting VMM.
        """
        ret = {}
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            res = vmctl.reset()
            mock_cmd.assert_called_once_with(
                ["vmctl", "reset"], output_loglevel="trace", python_shell=False
            )
            self.assertTrue(res)

    def test_reset_vms(self):
        """
        Tests resetting VMs.
        """
        ret = {}
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            res = vmctl.reset(vms=True)
            mock_cmd.assert_called_once_with(
                ["vmctl", "reset", "vms"], output_loglevel="trace", python_shell=False
            )
            self.assertTrue(res)

    def test_reset_switches(self):
        """
        Tests resetting switches.
        """
        ret = {}
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            res = vmctl.reset(switches=True)
            mock_cmd.assert_called_once_with(
                ["vmctl", "reset", "switches"],
                output_loglevel="trace",
                python_shell=False,
            )
            self.assertTrue(res)

    def test_reset_all(self):
        """
        Tests resetting all.
        """
        ret = {}
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            res = vmctl.reset(all=True)
            mock_cmd.assert_called_once_with(
                ["vmctl", "reset", "all"], output_loglevel="trace", python_shell=False
            )
            self.assertTrue(res)

    def test_start_existing_vm(self):
        """
        Tests starting a VM that is already defined.
        """
        ret = {}
        ret["stderr"] = "vmctl: started vm 4 successfully, tty /dev/ttyp4"
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        expected = {"changes": True, "console": "/dev/ttyp4"}
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            self.assertDictEqual(expected, vmctl.start("4"))

    def test_start_new_vm(self):
        """
        Tests starting a new VM.
        """
        ret = {}
        ret["stderr"] = "vmctl: started vm 4 successfully, tty /dev/ttyp4"
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        mock_status = MagicMock(return_value={})
        expected = {"changes": True, "console": "/dev/ttyp4"}
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            with patch("salt.modules.vmctl.status", mock_status):
                res = vmctl.start("web1", bootpath="/bsd.rd", nics=2, disk="/disk.img")
                mock_cmd.assert_called_once_with(
                    [
                        "vmctl",
                        "start",
                        "web1",
                        "-i 2",
                        "-b",
                        "/bsd.rd",
                        "-d",
                        "/disk.img",
                    ],
                    output_loglevel="trace",
                    python_shell=False,
                )
                self.assertDictEqual(expected, res)

    def test_status(self):
        """
        Tests getting status for all VMs.
        """
        ret = {}
        ret["stdout"] = (
            "   ID   PID VCPUS  MAXMEM  CURMEM     TTY        OWNER NAME\n"
            "    1   123     1    2.9G     150M  ttyp5       john web1 - stopping\n"
            "    2   456     1    512M     301M  ttyp4       paul web2\n"
            "    3     -     1    512M       -       -       george web3\n"
        )
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        expected = {
            "web1": {
                "curmem": "150M",
                "id": "1",
                "maxmem": "2.9G",
                "owner": "john",
                "pid": "123",
                "state": "stopping",
                "tty": "ttyp5",
                "vcpus": "1",
            },
            "web2": {
                "curmem": "301M",
                "id": "2",
                "maxmem": "512M",
                "owner": "paul",
                "pid": "456",
                "state": "running",
                "tty": "ttyp4",
                "vcpus": "1",
            },
            "web3": {
                "curmem": "-",
                "id": "3",
                "maxmem": "512M",
                "owner": "george",
                "pid": "-",
                "state": "stopped",
                "tty": "-",
                "vcpus": "1",
            },
        }
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            self.assertEqual(expected, vmctl.status())

    def test_status_single(self):
        """
        Tests getting status for a single VM.
        """
        ret = {}
        ret["stdout"] = (
            "   ID   PID VCPUS  MAXMEM  CURMEM     TTY        OWNER NAME\n"
            "    1   123     1    2.9G     150M  ttyp5       ringo web4\n"
            "    2     -     1    512M       -       -       george web3\n"
        )
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        expected = {
            "web4": {
                "curmem": "150M",
                "id": "1",
                "maxmem": "2.9G",
                "owner": "ringo",
                "pid": "123",
                "state": "running",
                "tty": "ttyp5",
                "vcpus": "1",
            },
        }
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            self.assertEqual(expected, vmctl.status("web4"))

    def test_stop_when_running(self):
        """
        Tests stopping a VM that is running.
        """
        ret = {}
        ret["stdout"] = ""
        ret["stderr"] = "vmctl: sent request to terminate vm 14"
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            res = vmctl.stop("web1")
            mock_cmd.assert_called_once_with(
                ["vmctl", "stop", "web1"], output_loglevel="trace", python_shell=False
            )
            self.assertTrue(res["changes"])

    def test_stop_when_stopped(self):
        """
        Tests stopping a VM that is already stopped/stopping.
        """
        ret = {}
        ret["stdout"] = ""
        ret["stderr"] = "vmctl: terminate vm command failed: Invalid argument"
        ret["retcode"] = 0
        mock_cmd = MagicMock(return_value=ret)
        with patch.dict(vmctl.__salt__, {"cmd.run_all": mock_cmd}):
            res = vmctl.stop("web1")
            mock_cmd.assert_called_once_with(
                ["vmctl", "stop", "web1"], output_loglevel="trace", python_shell=False
            )
            self.assertFalse(res["changes"])