starlingx/virtual-deployment
Code Issues Proposed changes
77c6d3109b
virtual-deployment/virtualbox/pybox/helper/tests/test_vboxmanage.py
Daniel Caires b5768ae359 Review LOG levels 
Adjust LOG levels throughout the code.

Story: 2005051
Task: 48516

Change-Id: Ie54a0d6a50c373ebc8b1d272aa5b574d5821b399
Signed-off-by: Daniel Caires <daniel.caires@encora.com>
2023-09-25 12:24:55 +00:00

1535 lines
46 KiB
Python
Raw Blame History

"""
Unit tests related to vboxmanage
"""
import unittest
import subprocess
from unittest.mock import patch, call
import vboxmanage
class VboxmanageVersionTestCase(unittest.TestCase):
"""
Class to test vboxmanage_version method
"""
@patch("vboxmanage.subprocess")
def test_vboxmanage_version(self, mock_subprocess):
"""
Test vboxmanage_version method
"""
# Setup
expected_version = "2.25.95"
mock_subprocess.check_output.return_value = expected_version
# Run
version = vboxmanage.vboxmanage_version()
# Assert
mock_subprocess.check_output.assert_called_once_with(["vboxmanage", "--version"],
stderr=mock_subprocess.STDOUT)
self.assertEqual(version, expected_version)
class VboxmanageExtpackTestCase(unittest.TestCase):
"""
Class to test vboxmanage_extpack method
"""
@patch("vboxmanage.subprocess.check_output")
@patch("vboxmanage.vboxmanage_version")
def test_vboxmanage_extpack(self, mock_vboxmanage_version, mock_check_output):
"""
Test vboxmanage_extpack method
"""
# Setup
mock_vboxmanage_version.return_value = b"2.25.95r123456\n"
version_path = "2.25.95"
filename = f"Oracle_VM_VirtualBox_Extension_Pack-{version_path}.vbox-extpack"
expected_wget_args = ["wget", f"http://download.virtualbox.org/virtualbox/{version_path}/{filename}", "-P",
"/tmp"]
expected_vboxmanage_args = ["vboxmanage", "extpack", "install", "/tmp/" + filename, "--replace"]
# Run
vboxmanage.vboxmanage_extpack()
# Assert
mock_vboxmanage_version.assert_called_once()
call_args_list = mock_check_output.call_args_list
self.assertEqual(call_args_list[0][0][0], expected_wget_args)
self.assertEqual(call_args_list[1][0][0], expected_vboxmanage_args)
class GetAllVmsTestCase(unittest.TestCase):
"""
Class to test get_all_vms method
"""
@patch("vboxmanage.vboxmanage_list")
@patch("vboxmanage.vboxmanage_showinfo")
def test_get_all_vms(self, mock_showinfo, mock_list):
"""
Test get_all_vms method
"""
# Setup
labname = "lab1"
mock_list.return_value = [
b'"lab1-controller-0" {2f7f1b1c-40fe-4063-8182-ece45bbe229d}',
b'"lab1-worker-0" {1f7a1a1a-30ee-4062-8182-edc45bbe239d}',
b'"lab1-storage-0" {1f6a1a1b-30ee-4071-8182-edd45bbe239d}',
b'"not-matching-vm" {2f7f1b1c-40fe-4063-8182-ece45bbe229d}'
]
mock_showinfo.return_value = b'groups="/lab1"\n'
# Run
vms = vboxmanage.get_all_vms(labname)
# Assert
mock_list.assert_called_once_with("vms")
expected_vms = [
'"lab1-controller-0" {2f7f1b1c-40fe-4063-8182-ece45bbe229d}',
'"lab1-worker-0" {1f7a1a1a-30ee-4062-8182-edc45bbe239d}',
'"lab1-storage-0" {1f6a1a1b-30ee-4071-8182-edd45bbe239d}',
]
self.assertCountEqual(vms, expected_vms)
class TakeSnapshotTestCase(unittest.TestCase):
"""
Class to test take_snapshot method
"""
labname = "lab1"
snapshot_name = "snap1"
vms = ["vm1", "vm2"]
@patch("vboxmanage._wait_for_vms_to_run", return_value=None)
@patch("vboxmanage._resume_running_vms", return_value=None)
@patch("vboxmanage.vboxmanage_takesnapshot", return_value=None)
@patch("vboxmanage._pause_running_vms", return_value=None)
@patch("vboxmanage.get_all_vms")
def test_take_snapshot_no_running_vms(self, mock_get_all_vms, mock_pause, mock_takesnapshot,
mock_resume, mock_wait):
"""
Test take_snapshot method with no running vms
"""
# Setup
mock_get_all_vms.side_effect = [self.vms, []]
# Run
vboxmanage.take_snapshot(self.labname, self.snapshot_name)
# Assert
mock_get_all_vms.assert_any_call(self.labname, option="vms")
mock_get_all_vms.assert_any_call(self.labname, option="runningvms")
mock_pause.assert_called_once()
mock_takesnapshot.assert_called_once_with(self.vms, self.snapshot_name)
mock_resume.assert_called_once()
mock_wait.assert_not_called()
@patch("vboxmanage._wait_for_vms_to_run", return_value=None)
@patch("vboxmanage._resume_running_vms", return_value=None)
@patch("vboxmanage.vboxmanage_takesnapshot", return_value=None)
@patch("vboxmanage._pause_running_vms", return_value=None)
@patch("vboxmanage.get_all_vms")
def test_take_snapshot_with_running_vms(self, mock_get_all_vms, mock_pause, mock_takesnapshot,
mock_resume, mock_wait):
"""
Test take_snapshot method with running vms
"""
# Setup
running_vms = ["vm1"]
mock_get_all_vms.side_effect = [self.vms, running_vms]
# Run
vboxmanage.take_snapshot(self.labname, self.snapshot_name)
# Assert
mock_get_all_vms.assert_any_call(self.labname, option="vms")
mock_get_all_vms.assert_any_call(self.labname, option="runningvms")
mock_pause.assert_called_once()
mock_takesnapshot.assert_called_once_with(self.vms, self.snapshot_name)
mock_resume.assert_called_once_with(running_vms)
mock_wait.assert_called_once_with(self.labname, running_vms, self.vms)
class PauseRunningVmsTestCase(unittest.TestCase):
"""
Class to test _pause_running_vms method
"""
@patch("vboxmanage.os.waitpid", return_value=None)
@patch("vboxmanage.os._exit", return_value=None) # pylint: disable=protected-access
@patch("vboxmanage.vboxmanage_controlvms", return_value=None)
@patch("vboxmanage.os.fork")
def test_pause_running_vms_no_vms(self, mock_fork, mock_controlvms, mock_exit, mock_waitpid):
"""
Test _pause_running_vms method with no running vms
"""
# Setup
running_vms = []
vms = []
# Run
vboxmanage._pause_running_vms(running_vms, vms)
# Assert
mock_fork.assert_not_called()
mock_controlvms.assert_not_called()
mock_exit.assert_not_called()
mock_waitpid.assert_not_called()
@patch("vboxmanage.os.waitpid", return_value=None)
@patch("vboxmanage.os._exit", return_value=None) # pylint: disable=protected-access
@patch("vboxmanage.vboxmanage_controlvms", return_value=None)
@patch("vboxmanage.os.fork")
def test_pause_running_vms_with_vms(self, mock_fork, mock_controlvms, mock_exit, mock_waitpid):
"""
Test _pause_running_vms method with running vms
"""
# Setup
running_vms = ["vm1", "vm2"]
vms = ["vm1", "vm2", "vm3"]
mock_fork.return_value = 0
# Run
vboxmanage._pause_running_vms(running_vms, vms)
# Assert
mock_fork.assert_has_calls([call() for _ in running_vms])
mock_controlvms.assert_has_calls([call([vm], "pause") for vm in running_vms])
mock_exit.assert_has_calls([call(0) for _ in running_vms])
mock_waitpid.assert_has_calls([call(0, 0) for _ in vms])
class ResumeRunningVMsTestCase(unittest.TestCase):
"""
Class to test _resume_running_vms method
"""
@patch("vboxmanage.os.waitpid", return_value=None)
@patch("vboxmanage.os._exit", return_value=None)
@patch("vboxmanage.os.fork")
@patch("vboxmanage.vboxmanage_controlvms", return_value=None)
def test_resume_running_vms(self, mock_controlvms, mock_fork, mock_exit, mock_waitpid):
"""
Test _resume_running_vms method
"""
# Setup
runningvms = ["vm1", "vm2"]
mock_fork.side_effect = [0, 0]
# Run
vboxmanage._resume_running_vms(runningvms)
# Assert
mock_controlvms.assert_has_calls([call([runningvms[0]], "resume"), call([runningvms[1]], "resume")])
mock_exit.assert_has_calls([call(0), call(0)])
mock_waitpid.assert_has_calls([call(0, 0), call(0, 0)])
@patch("vboxmanage.vboxmanage_controlvms", return_value=None)
def test_resume_running_vms_one_vm(self, mock_controlvms):
"""
Test _resume_running_vms method with one running vm
"""
# Setup
runningvms = ["vm1"]
# Run
vboxmanage._resume_running_vms(runningvms)
# Assert
mock_controlvms.assert_not_called()
class WaitForVmsToRunTestCase(unittest.TestCase):
"""
Class to test _wait_for_vms_to_run method
"""
@patch("vboxmanage.LOG.info", return_value=None)
@patch("vboxmanage.get_all_vms")
def test_wait_for_vms_to_run_successful(self, mock_get_all_vms, mock_log_info):
"""
Test _wait_for_vms_to_run method
"""
# Setup
labname = "lab1"
runningvms = ["vm1"]
vms = ["vm1", "vm2", "vm3"]
mock_get_all_vms.side_effect = [vms, vms[0:2], [vms[0]]]
# Run
vboxmanage._wait_for_vms_to_run(labname, runningvms, vms)
# Assert
mock_get_all_vms.assert_has_calls([call(labname, option="runningvms"), call(labname, option="runningvms"),
call(labname, option="runningvms")])
mock_log_info.assert_called_with("All VMs %s are up running after taking snapshot...", vms)
@patch("vboxmanage.time.sleep", return_value=None)
@patch("vboxmanage.LOG.info", return_value=None)
@patch("vboxmanage.get_all_vms", return_value=["vm1"])
def test_wait_for_vms_to_run_no_retry(self, mock_get_all_vms, mock_log_info, mock_sleep):
"""
Test _wait_for_vms_to_run method with no need for retry
"""
# Setup
labname = "lab1"
runningvms = ["vm1"]
vms = ["vm1"]
# Run
vboxmanage._wait_for_vms_to_run(labname, runningvms, vms)
# Assert
mock_get_all_vms.assert_called_once_with(labname, option="runningvms")
mock_sleep.assert_not_called()
mock_log_info.assert_called_with("All VMs %s are up running after taking snapshot...", vms)
class RestoreSnapshotTestCase(unittest.TestCase):
"""
Class to test restore_snapshot method
"""
@patch("vboxmanage.time.sleep", return_value=None)
@patch("vboxmanage.vboxmanage_controlvms", return_value=None)
@patch("vboxmanage.vboxmanage_restoresnapshot", return_value=None)
@patch("vboxmanage.vboxmanage_startvm", return_value=None)
def test_restore_snapshot(self, mock_startvm, mock_restoresnapshot, mock_controlvms, mock_sleep):
"""
Test restore_snapshot method
"""
# Setup
node_list = ["controller-0", "vm1", "vm2"]
snapshot_name = "snapshot1"
# Run
vboxmanage.restore_snapshot(node_list, snapshot_name)
# Assert
mock_controlvms.assert_called_once_with(node_list, "poweroff")
mock_restoresnapshot.assert_has_calls(
[call(node_list[0], snapshot_name), call(node_list[1], snapshot_name), call(node_list[2], snapshot_name)])
mock_startvm.assert_has_calls([call(node_list[1]), call(node_list[2]), call(node_list[0])])
mock_sleep.assert_has_calls([call(5), call(5), call(5), call(10), call(10), call(10)])
@patch("vboxmanage.time.sleep", return_value=None)
@patch("vboxmanage.vboxmanage_controlvms", return_value=None)
@patch("vboxmanage.vboxmanage_restoresnapshot", return_value=None)
@patch("vboxmanage.vboxmanage_startvm", return_value=None)
def test_restore_snapshot_empty_node_list(self, mock_startvm, mock_restoresnapshot, mock_controlvms, mock_sleep):
"""
Test restore_snapshot method with empty node list
"""
# Setup
node_list = []
snapshot_name = "snapshot1"
# Run
vboxmanage.restore_snapshot(node_list, snapshot_name)
# Assert
mock_controlvms.assert_not_called()
mock_restoresnapshot.assert_not_called()
mock_startvm.assert_not_called()
mock_sleep.assert_not_called()
class VboxmanageListTestCase(unittest.TestCase):
"""
Class to test vboxmanage_list method
"""
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_list(self, mock_subprocess):
"""
Test vboxmanage_list method
"""
# Setup
expected_vms = [b"vm1", b"vm2", b"vm3"]
mock_subprocess.return_value = b'"vm1"\n"vm2"\n"vm3"\n'
# Run
vms = vboxmanage.vboxmanage_list()
# Assert
mock_subprocess.assert_called_once_with(["vboxmanage", "list", "vms"], stderr=vboxmanage.subprocess.STDOUT)
self.assertCountEqual(vms, expected_vms)
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_list_custom_option(self, mock_subprocess):
"""
Test vboxmanage_list method with custom option
"""
# Setup
expected_vms = [b"vm1", b"vm2"]
mock_subprocess.return_value = b'"vm1"\n"vm2"\n'
custom_option = "runningvms"
# Run
vms = vboxmanage.vboxmanage_list(custom_option)
# Assert
mock_subprocess.assert_called_once_with(["vboxmanage", "list", custom_option],
stderr=vboxmanage.subprocess.STDOUT)
self.assertCountEqual(vms, expected_vms)
class VboxmanageShowinfoTestCase(unittest.TestCase):
"""
Class to test vboxmanage_showinfo method
"""
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_showinfo(self, mock_subprocess):
"""
Test vboxmanage_showinfo method
"""
# Setup
expected_info = "Some VM info"
mock_subprocess.return_value = expected_info
host = "vm1"
# Run
info = vboxmanage.vboxmanage_showinfo(host)
# Assert
mock_subprocess.assert_called_once_with(["vboxmanage", "showvminfo", host, "--machinereadable"],
stderr=vboxmanage.subprocess.STDOUT)
self.assertEqual(info, expected_info)
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_showinfo_bytes(self, mock_subprocess):
"""
Test vboxmanage_showinfo method with bytes input
"""
# Setup
expected_info = "Some VM info"
mock_subprocess.return_value = expected_info
host = "vm1"
# Run
info = vboxmanage.vboxmanage_showinfo(host)
# Assert
mock_subprocess.assert_called_once_with(["vboxmanage", "showvminfo", "vm1", "--machinereadable"],
stderr=vboxmanage.subprocess.STDOUT)
self.assertEqual(info, expected_info)
class VboxmanageCreatevmTestCase(unittest.TestCase):
"""
Class to test vboxmanage_createvm method
"""
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_createvm(self, mock_subprocess):
"""
Test vboxmanage_createvm method
"""
# Setup
hostname = "vm1"
labname = "lab1"
# Run
vboxmanage.vboxmanage_createvm(hostname, labname)
# Assert
mock_subprocess.assert_called_once_with(
["vboxmanage", "createvm", "--name", hostname, "--register", "--ostype", "Linux_64", "--groups",
"/" + labname],
stderr=vboxmanage.subprocess.STDOUT
)
def test_vboxmanage_createvm_no_hostname(self):
"""
Test vboxmanage_createvm method with no hostname
"""
# Setup
hostname = None
labname = "lab1"
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_createvm(hostname, labname)
def test_vboxmanage_createvm_no_labname(self):
"""
Test vboxmanage_createvm method with no labname
"""
# Setup
hostname = "vm1"
labname = None
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_createvm(hostname, labname)
class VboxmanageDeletevmsTestCase(unittest.TestCase):
"""
Class to test vboxmanage_deletevms method
"""
@patch("vboxmanage.vboxmanage_deletemedium")
@patch("vboxmanage.vboxmanage_list")
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_deletevms(self, mock_subprocess, mock_list, mock_deletemedium):
"""
Test vboxmanage_deletevms method
"""
# Setup
hosts = ["vm1", "vm2"]
mock_list.return_value = []
# Run
vboxmanage.vboxmanage_deletevms(hosts)
# Assert
mock_subprocess.assert_has_calls(
[call(["vboxmanage", "unregistervm", host, "--delete"], stderr=vboxmanage.subprocess.STDOUT) for host in
hosts],
any_order=True
)
mock_deletemedium.assert_has_calls(
[call(host) for host in hosts],
any_order=True
)
mock_list.assert_called_once_with("vms")
def test_vboxmanage_deletevms_no_hosts(self):
"""
Test vboxmanage_deletevms method with no hosts
"""
# Setup
hosts = None
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_deletevms(hosts)
class VboxmanageHostonlyifcreateTestCase(unittest.TestCase):
"""
Class to test vboxmanage_hostonlyifcreate method
"""
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_hostonlyifcreate(self, mock_subprocess):
"""
Test vboxmanage_hostonlyifcreate method
"""
# Setup
name = "vboxnet0"
oam_ip = "192.168.0.1"
netmask = "255.255.255.0"
# Run
vboxmanage.vboxmanage_hostonlyifcreate(name, oam_ip, netmask)
# Assert
mock_subprocess.assert_has_calls([
call(["vboxmanage", "hostonlyif", "create"], stderr=vboxmanage.subprocess.STDOUT),
call(["vboxmanage", "hostonlyif", "ipconfig", name, "--ip", oam_ip, "--netmask", netmask],
stderr=vboxmanage.subprocess.STDOUT)
])
def test_vboxmanage_hostonlyifcreate_no_name(self):
"""
Test vboxmanage_hostonlyifcreate method with no network name
"""
# Setup
name = None
oam_ip = "192.168.0.1"
netmask = "255.255.255.0"
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_hostonlyifcreate(name, oam_ip, netmask)
def test_vboxmanage_hostonlyifcreate_no_oam_ip(self):
"""
Test vboxmanage_hostonlyifcreate method with no OAM IP
"""
# Setup
name = "vboxnet0"
oam_ip = None
netmask = "255.255.255.0"
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_hostonlyifcreate(name, oam_ip, netmask)
def test_vboxmanage_hostonlyifcreate_no_netmask(self):
"""
Test vboxmanage_hostonlyifcreate method with no OAM Netmask
"""
# Setup
name = "vboxnet0"
oam_ip = "192.168.0.1"
netmask = None
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_hostonlyifcreate(name, oam_ip, netmask)
class VboxmanageHostonlyifdeleteTestCase(unittest.TestCase):
"""
Class to test vboxmanage_hostonlyifdelete method
"""
@patch("vboxmanage.subprocess.check_output")
def test_vboxmanage_hostonlyifdelete(self, mock_subprocess):
"""
Test vboxmanage_hostonlyifdelete method
"""
# Setup
name = "vboxnet0"
# Run
vboxmanage.vboxmanage_hostonlyifdelete(name)
# Assert
mock_subprocess.assert_called_once_with(["vboxmanage", "hostonlyif", "remove", name],
stderr=vboxmanage.subprocess.STDOUT)
def test_vboxmanage_hostonlyifdelete_no_name(self):
"""
Test vboxmanage_hostonlyifdelete method with no network name
"""
# Setup
name = None
# Assert
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_hostonlyifdelete(name)
class VboxmanageModifyvmTestCase(unittest.TestCase):
"""
Class to test vboxmanage_modifyvm method
"""
@patch("vboxmanage.subprocess.check_output")
@patch("vboxmanage._contains_value")
@patch("vboxmanage._is_network_configured")
@patch("vboxmanage._is_nat_network_configured")
@patch("vboxmanage._is_uart_configured")
@patch("vboxmanage._get_network_configuration")
@patch("vboxmanage._add_uart")
def test_vboxmanage_modifyvm(self, mock_add_uart, mock_get_network_configuration, mock_is_uart_configured,
mock_is_nat_network_configured, mock_is_network_configured,
mock_contains_value, mock_subprocess):
"""
Test vboxmanage_modifyvm method
"""
# Setup
hostname = "test_host"
vm_config = {
"cpus": "2",
"memory": "1024",
"nicnum": "1",
"nicbootprio2": "1"
}
mock_contains_value.return_value = True
mock_is_network_configured.return_value = False
mock_is_nat_network_configured.return_value = False
mock_is_uart_configured.return_value = False
mock_get_network_configuration.return_value = []
mock_add_uart.return_value = []
expected_cmd = ["vboxmanage", "modifyvm", hostname, "--cpus", "2", "--memory", "1024", "--nicbootprio2", "1",
"--boot4", "net"]
# Run
vboxmanage.vboxmanage_modifyvm(hostname, vm_config)
# Assert
mock_subprocess.assert_called_once_with(expected_cmd, stderr=vboxmanage.subprocess.STDOUT)
class IsNetworkConfiguredTestCase(unittest.TestCase):
"""
Class to test _is_network_configured method
"""
@patch("vboxmanage._contains_value")
def test_is_network_configured(self, mock_contains_value):
"""
Test _is_network_configured method
"""
# Setup
vm_config = {
"nic": "test_nic",
"nictype": "test_nictype",
"nicpromisc": "test_nicpromisc",
"nicnum": "test_nicnum"
}
mock_contains_value.return_value = True
# Run
result = vboxmanage._is_network_configured(vm_config)
# Assert
self.assertTrue(result)
mock_contains_value.assert_any_call("nic", vm_config)
mock_contains_value.assert_any_call("nictype", vm_config)
mock_contains_value.assert_any_call("nicpromisc", vm_config)
mock_contains_value.assert_any_call("nicnum", vm_config)
class GetNetworkConfigurationTestCase(unittest.TestCase):
"""
Class to test _get_network_configuration method
"""
@patch("vboxmanage._contains_value")
def test_get_network_configuration(self, mock_contains_value):
"""
Test _get_network_configuration method
"""
# Setup
vm_config = {
"nic": "test_nic",
"nictype": "test_nictype",
"nicpromisc": "test_nicpromisc",
"nicnum": "1",
"intnet": "test_intnet",
"hostonlyadapter": "test_hostonlyadapter",
"natnetwork": "test_natnetwork",
"prefix": "test_prefix"
}
mock_contains_value.return_value = True
# Expected output
expected_output = [
'--nic1', 'test_nic',
'--nictype1', 'test_nictype',
'--nicpromisc1', 'test_nicpromisc',
'--intnet1', 'test_prefix-test_intnet',
'--hostonlyadapter1', 'test_hostonlyadapter',
'--nat-network1', 'test_natnetwork'
]
# Run
result = vboxmanage._get_network_configuration(vm_config)
# Assert
self.assertEqual(result, expected_output)
class IsNatNetworkConfiguredTestCase(unittest.TestCase):
"""
Class to test _is_nat_network_configured method
"""
@patch("vboxmanage._contains_value")
def test_is_nat_network_configured_true(self, mock_contains_value):
"""
Test _is_nat_network_configured method with nat
"""
# Setup
vm_config = {
"nicnum": "1",
"nictype": "nat",
}
mock_contains_value.return_value = True
# Run
result = vboxmanage._is_nat_network_configured(vm_config)
# Assert
self.assertEqual(result, True)
@patch("vboxmanage._contains_value")
def test_is_nat_network_configured_false(self, mock_contains_value):
"""
Test _is_nat_network_configured method with non nat
"""
# Setup
vm_config = {
"nicnum": "1",
"nictype": "non-nat",
}
mock_contains_value.return_value = True
# Run
result = vboxmanage._is_nat_network_configured(vm_config)
# Assert
self.assertEqual(result, False)
class IsUartConfiguredTestCase(unittest.TestCase):
"""
Class to test _is_uart_configured method
"""
@patch("vboxmanage._contains_value")
def test_is_uart_configured_true(self, mock_contains_value):
"""
Test _is_uart_configured method with all key values
"""
# Setup
vm_config = {
"uartbase": "0x3F8",
"uartport": "1",
"uartmode": "server",
"uartpath": "/tmp/uart",
}
mock_contains_value.return_value = True
# Run
result = vboxmanage._is_uart_configured(vm_config)
# Assert
self.assertEqual(result, True)
@patch("vboxmanage._contains_value")
def test_is_uart_configured_false(self, mock_contains_value):
"""
Test _is_uart_configured method without all key values
"""
# Setup
vm_config = {
"uartbase": "0x3F8",
"uartport": "1",
"uartmode": "server",
}
mock_contains_value.side_effect = [True, True, True, False]
# Run
result = vboxmanage._is_uart_configured(vm_config)
# Assert
self.assertEqual(result, False)
class AddUartTestCase(unittest.TestCase):
"""
Class to test _add_uart method
"""
def setUp(self):
self.vm_config = {
"uartbase": "0x3F8",
"uartport": "4",
"uartmode": "file",
"uartpath": "1",
}
def test_add_uart_windows(self):
"""
Test _add_uart method
"""
result = vboxmanage._add_uart(self.vm_config)
expected = [
'--uart1', '0x3F8', '4', '--uartmode1', 'file', '1'
]
self.assertCountEqual(result, expected)
class ContainsValueTestCase(unittest.TestCase):
"""
Class to test _contains_value method
"""
def setUp(self):
self.dictionary = {
"key1": "value1",
"key2": None,
}
def test_contains_value_key_present_value_truthy(self):
"""
Test _contains_value method with key present and value truthy
"""
result = vboxmanage._contains_value("key1", self.dictionary)
self.assertTrue(result)
def test_contains_value_key_present_value_falsy(self):
"""
Test _contains_value method with key present and value falsy
"""
result = vboxmanage._contains_value("key2", self.dictionary)
self.assertFalse(result)
def test_contains_value_key_absent(self):
"""
Test _contains_value method with key absent
"""
result = vboxmanage._contains_value("key3", self.dictionary)
self.assertFalse(result)
class StorageCtlTestCase(unittest.TestCase):
"""
Class to test vboxmanage_storagectl method
"""
@patch('vboxmanage.subprocess')
def test_vboxmanage_storagectl(self, mock_subprocess):
"""
Test vboxmanage_storagectl method
"""
hostname = "test-host"
storectl = "sata"
hostiocache = "off"
# Run
vboxmanage.vboxmanage_storagectl(hostname, storectl, hostiocache)
# Assert
cmd = [
"vboxmanage",
"storagectl",
hostname,
"--name",
storectl,
"--add",
storectl,
"--hostiocache",
hostiocache,
]
mock_subprocess.check_output.assert_called_once_with(cmd, stderr=mock_subprocess.STDOUT)
def test_vboxmanage_storagectl_no_hostname(self):
"""
Test vboxmanage_storagectl method without hostname
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_storagectl(None, "sata", "off")
def test_vboxmanage_storagectl_no_storectl(self):
"""
Test vboxmanage_storagectl method without storectl
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_storagectl("test-host", None, "off")
class StorageAttachTestCase(unittest.TestCase):
"""
Class to test vboxmanage_storageattach method
"""
@patch('vboxmanage.subprocess')
def test_vboxmanage_storageattach(self, mock_subprocess):
"""
Test vboxmanage_storageattach method
"""
hostname = "test-host"
storage_config = {
"storectl": "sata",
"storetype": "hdd",
"disk": "disk1",
"port_num": "0",
"device_num": "0",
}
# Run
vboxmanage.vboxmanage_storageattach(hostname, storage_config)
# Assert
cmd = [
"vboxmanage",
"storageattach",
hostname,
"--storagectl",
storage_config["storectl"],
"--medium",
storage_config["disk"],
"--type",
storage_config["storetype"],
"--port",
storage_config["port_num"],
"--device",
storage_config["device_num"],
]
mock_subprocess.check_output.assert_called_once_with(cmd, stderr=mock_subprocess.STDOUT)
def test_vboxmanage_storageattach_no_hostname(self):
"""
Test vboxmanage_storageattach method without hostname
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_storageattach(None, {"disk": "disk1"})
def test_vboxmanage_storageattach_no_storage_config(self):
"""
Test vboxmanage_storageattach method without storage_config
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_storageattach("test-host", None)
def test_vboxmanage_storageattach_no_disk_in_config(self):
"""
Test vboxmanage_storageattach method without 'disk' in storage_config
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_storageattach("test-host", {})
class DeleteMediumTestCase(unittest.TestCase):
"""
Class to test vboxmanage_deletemedium method
"""
@patch('vboxmanage.subprocess')
@patch('vboxmanage.getpass')
@patch('vboxmanage.os')
def test_vboxmanage_deletemedium(self, mock_os, mock_getpass, mock_subprocess):
"""
Test vboxmanage_deletemedium method
"""
hostname = "test-host"
vbox_home_dir = "/home"
username = "user1"
mock_getpass.getuser.return_value = username
mock_os.path.isfile.return_value = True
vbox_home_dir = f"{vbox_home_dir}/{username}/vbox_disks/"
disk_list = ["test-host-user1.vdi"]
mock_os.listdir.return_value = disk_list
# Run
vboxmanage.vboxmanage_deletemedium(hostname)
# Assert
cmd = [
"vboxmanage",
"closemedium",
"disk",
f"{vbox_home_dir}{disk_list[0]}",
"--delete",
]
mock_subprocess.check_output.assert_called_once_with(cmd, stderr=mock_subprocess.STDOUT)
mock_os.remove.assert_called_once_with(f"{vbox_home_dir}{disk_list[0]}")
def test_vboxmanage_deletemedium_no_hostname(self):
"""
Test vboxmanage_deletemedium method without hostname
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_deletemedium(None, "/home")
class CreateMediumTestCase(unittest.TestCase):
"""
Class to test vboxmanage_createmedium method
"""
@patch('vboxmanage.subprocess')
@patch('vboxmanage.getpass')
@patch('vboxmanage.LOG')
@patch('vboxmanage.vboxmanage_storageattach')
def test_vboxmanage_createmedium(self, mock_storageattach, mock_log, mock_getpass, mock_subprocess):
"""
Test vboxmanage_createmedium method
"""
hostname = "test-host"
disk_list = ["50000", "60000"]
vbox_home_dir = "/home"
username = "user1"
mock_getpass.getuser.return_value = username
# Run
vboxmanage.vboxmanage_createmedium(hostname, disk_list, vbox_home_dir)
# Assert
disk_count = 1
port_num = 0
device_num = 0
for disk in disk_list:
file_name = f"{vbox_home_dir}/{username}/vbox_disks/{hostname}_disk_{disk_count}"
cmd = [
"vboxmanage",
"createmedium",
"disk",
"--size",
disk,
"--filename",
file_name,
"--format",
"vdi",
"--variant",
"standard",
]
# Assert the logs and command for each disk
mock_log.info.assert_any_call(
"Creating disk %s of size %s on VM %s on device %s port %s",
file_name,
disk,
hostname,
device_num,
port_num,
)
mock_subprocess.check_output.assert_any_call(cmd, stderr=mock_subprocess.STDOUT)
# Assert the storageattach call for each disk
mock_storageattach.assert_any_call(
hostname,
{
"storectl": "sata",
"storetype": "hdd",
"disk": file_name + ".vdi",
"port_num": str(port_num),
"device_num": str(device_num),
},
)
# Update values for the next disk
disk_count += 1
port_num += 1
def test_vboxmanage_createmedium_no_hostname(self):
"""
Test vboxmanage_createmedium method without hostname
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_createmedium(None, ["50000", "60000"], "/home")
def test_vboxmanage_createmedium_no_disk_list(self):
"""
Test vboxmanage_createmedium method without disk_list
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_createmedium("test-host", None, "/home")
class StartVMTestCase(unittest.TestCase):
"""
Class to test vboxmanage_startvm method
"""
@patch('vboxmanage.vboxmanage_list')
@patch('vboxmanage.LOG')
def test_vboxmanage_startvm(self, mock_log, mock_list):
"""
Test vboxmanage_startvm method
"""
hostname = "test-host"
running_vms = [b"another-host", hostname.encode("utf-8")]
mock_list.return_value = running_vms
# Run
vboxmanage.vboxmanage_startvm(hostname)
# Assert
# Should check if VM is running and find it is running
mock_log.info.assert_any_call("Check if VM is running")
mock_list.assert_called_with(option="runningvms")
mock_log.info.assert_any_call("Host %s is already started", hostname)
def test_vboxmanage_startvm_no_hostname(self):
"""
Test vboxmanage_startvm method without hostname
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_startvm(None)
class ControlVMsTestCase(unittest.TestCase):
"""
Class to test vboxmanage_controlvms method
"""
@patch('vboxmanage.subprocess')
@patch('vboxmanage.LOG')
def test_vboxmanage_controlvms(self, mock_log, mock_subprocess):
"""
Test vboxmanage_controlvms method
"""
hosts = ["test-host1", "test-host2"]
action = "pause"
# Run
vboxmanage.vboxmanage_controlvms(hosts, action)
# Assert
# Should execute action on each VM
for host in hosts:
mock_log.info.assert_any_call("Executing %s action on VM %s", action, host)
mock_subprocess.call.assert_any_call(["vboxmanage", "controlvm", host, action],
stderr=mock_subprocess.STDOUT)
def test_vboxmanage_controlvms_no_hosts(self):
"""
Test vboxmanage_controlvms method without hosts
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_controlvms(None, "pause")
def test_vboxmanage_controlvms_no_action(self):
"""
Test vboxmanage_controlvms method without action
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_controlvms(["test-host"], None)
class TakeSnapshotTestCase2(unittest.TestCase):
"""
Class to test vboxmanage_takesnapshot method
"""
@patch('vboxmanage.subprocess')
def test_vboxmanage_takesnapshot(self, mock_subprocess):
"""
Test vboxmanage_takesnapshot method
"""
hosts = ["test-host1", "test-host2"]
name = "test-snapshot"
# Run
vboxmanage.vboxmanage_takesnapshot(hosts, name)
# Assert
# Should execute action on each VM
for host in hosts:
mock_subprocess.call.assert_any_call(["vboxmanage", "snapshot", host, "take", name],
stderr=mock_subprocess.STDOUT)
def test_vboxmanage_takesnapshot_no_hosts(self):
"""
Test vboxmanage_takesnapshot method without hosts
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_takesnapshot(None, "test-snapshot")
def test_vboxmanage_takesnapshot_no_name(self):
"""
Test vboxmanage_takesnapshot method without snapshot name
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_takesnapshot(["test-host"], None)
class RestoreSnapshotTestCase2(unittest.TestCase):
"""
Class to test vboxmanage_restoresnapshot method
"""
@patch('vboxmanage.subprocess')
def test_vboxmanage_restoresnapshot(self, mock_subprocess):
"""
Test vboxmanage_restoresnapshot method
"""
host = "test-host1"
name = "test-snapshot"
# Run
vboxmanage.vboxmanage_restoresnapshot(host, name)
# Assert
mock_subprocess.call.assert_called_once_with(["vboxmanage", "snapshot", host, "restore", name],
stderr=mock_subprocess.STDOUT)
def test_vboxmanage_restoresnapshot_no_hosts(self):
"""
Test vboxmanage_restoresnapshot method without hosts
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_restoresnapshot(None, "test-snapshot")
def test_vboxmanage_restoresnapshot_no_name(self):
"""
Test vboxmanage_restoresnapshot method without snapshot name
"""
with self.assertRaises(AssertionError):
vboxmanage.vboxmanage_restoresnapshot(["test-host"], None)
class VboxmanageAddportforwardTestCase(unittest.TestCase):
"""
Class to test vboxmanage_addportforward method
"""
def setUp(self):
"""
Method to set up the parameters used on the tests in this class
"""
# Set up the test parameters
self.rule_name = "rule1"
self.local_port = "8080"
self.guest_ip = "10.10.10.1"
self.guest_port = "80"
self.network = "NatNetwork"
@patch('subprocess.check_output')
@patch('utils.install_log.LOG.info')
def test_vboxmanage_addportforward_success(self, mock_log, mock_check_output):
"""
Test vboxmanage_addportforward method with success
"""
# Setup
mock_check_output.return_value = b''
# Run
result = vboxmanage.vboxmanage_addportforward(self.rule_name, self.local_port, self.guest_ip, self.guest_port,
self.network)
# Assert
mock_log.assert_called_once_with(
"Creating port-forwarding rule to: %s", "rule1:tcp:[]:8080:[10.10.10.1]:80")
mock_check_output.assert_called_once_with([
"vboxmanage",
"natnetwork",
"modify",
"--netname",
"NatNetwork",
"--port-forward-4",
"rule1:tcp:[]:8080:[10.10.10.1]:80"
], stderr=subprocess.STDOUT)
self.assertTrue(result)
@patch('subprocess.check_output')
@patch('utils.install_log.LOG.info')
@patch('utils.install_log.LOG.error')
def test_vboxmanage_addportforward_error(self, mock_error, mock_info, mock_check_output):
"""
Test vboxmanage_addportforward method with error
"""
# Setup
mock_check_output.side_effect = subprocess.CalledProcessError(returncode=1, cmd='vboxmanage')
# Run
result = vboxmanage.vboxmanage_addportforward(self.rule_name, self.local_port, self.guest_ip, self.guest_port,
self.network)
# Assert
mock_info.assert_any_call(
"Creating port-forwarding rule to: %s", "rule1:tcp:[]:8080:[10.10.10.1]:80")
mock_error.assert_any_call(
"Error while trying to create port-forwarding rule. Continuing installation!")
mock_check_output.assert_called_once_with([
"vboxmanage",
"natnetwork",
"modify",
"--netname",
"NatNetwork",
"--port-forward-4",
"rule1:tcp:[]:8080:[10.10.10.1]:80"
], stderr=subprocess.STDOUT)
self.assertFalse(result)
class VboxmanageDeleteportforwardTestCase(unittest.TestCase):
"""
Class to test vboxmanage_deleteportforward method
"""
def setUp(self):
"""
Method to set up the parameters used on the tests in this class
"""
# Set up the test parameters
self.rule_name = "rule1"
self.network = "NatNetwork"
@patch('subprocess.check_output')
@patch('utils.install_log.LOG.info')
def test_vboxmanage_deleteportforward_success(self, mock_log, mock_check_output):
"""
Test vboxmanage_deleteportforward method with success
"""
# Setup
mock_check_output.return_value = b''
# Run
result = vboxmanage.vboxmanage_deleteportforward(self.rule_name, self.network)
# Assert
mock_log.assert_called_once_with(
"Removing previous forwarding rule '%s' from NAT network '%s'", "rule1", "NatNetwork")
mock_check_output.assert_called_once_with([
"vboxmanage",
"natnetwork",
"modify",
"--netname",
"NatNetwork",
"--port-forward-4",
"delete",
"rule1"
], stderr=subprocess.STDOUT)
self.assertIsNone(result)
@patch('subprocess.check_output')
@patch('utils.install_log.LOG.info')
@patch('utils.install_log.LOG.error')
def test_vboxmanage_deleteportforward_error(self, mock_error, mock_info, mock_check_output):
"""
Test vboxmanage_deleteportforward method with error
"""
# Setup
mock_check_output.side_effect = subprocess.CalledProcessError(returncode=1, cmd='vboxmanage')
# Run
result = vboxmanage.vboxmanage_deleteportforward(self.rule_name, self.network)
# Assert
mock_info.assert_any_call(
"Removing previous forwarding rule '%s' from NAT network '%s'", "rule1", "NatNetwork")
mock_error.assert_any_call(
"Error while trying to delete port-forwarding rule. Continuing installation!")
mock_check_output.assert_called_once_with([
"vboxmanage",
"natnetwork",
"modify",
"--netname",
"NatNetwork",
"--port-forward-4",
"delete",
"rule1"
], stderr=subprocess.STDOUT)
self.assertFalse(result)
class VboxmanageGetrulenameTestcase(unittest.TestCase):
"""
Class to test vboxmanage_getrulename method
"""
def setUp(self):
"""
Method to set up the parameters used on the tests in this class
"""
# Mock the subprocess.check_output function to return sample output
self.mock_output = b'''
NetworkName: NatNetwork\nIP: 10.10.10.1\nNetwork: 10.10.10.0/24\nIPv6 Enabled: Yes\nIPv6 Prefix: fd17:625c:f037:2::/64\nDHCP Enabled: Yes\nEnabled: Yes\nPort-forwarding (ipv4)\n Rule1:tcp:[]:8080:[10.10.10.3]:80\n Rule2:tcp:[]:32000:[10.10.10.4]:53\nloopback mappings (ipv4)\n 127.0.0.1=2'''
# Set up the test parameters
self.network = "NatNetwork"
self.local_port = "8080"
self.existing_rule_name = "Rule1"
self.nonexistent_rule_name = ""
self.no_network = "NatNetwork1"
self.no_local_port = "1234"
@patch('subprocess.check_output')
def test_existing_rule(self, mock_check_output):
"""
Test vboxmanage_getrulename method with existing rule
"""
# Setup
mock_check_output.return_value = self.mock_output
# Run
rule_name = vboxmanage.vboxmanage_getrulename(self.network, self.local_port)
# Assert
self.assertEqual(rule_name, self.existing_rule_name)
@patch('subprocess.check_output')
def test_no_rule(self, mock_check_output):
"""
Test vboxmanage_getrulename method with no rule found
"""
# Setup
mock_check_output.return_value = self.mock_output
# Run
rule_name = vboxmanage.vboxmanage_getrulename(self.network, self.no_local_port)
# Assert
self.assertEqual(rule_name, self.nonexistent_rule_name)
@patch('subprocess.check_output')
def test_no_network(self, mock_check_output):
"""
Test vboxmanage_getrulename method with no network found
"""
# Setup
mock_check_output.return_value = self.mock_output
# Run
rule_name = vboxmanage.vboxmanage_getrulename(self.no_network, self.local_port)
# Assert
self.assertEqual(rule_name, self.nonexistent_rule_name)
if __name__ == '__main__':
unittest.main()
View Git Blame Copy Permalink