starlingx/virtual-deployment
Code Issues Proposed changes
virtual-deployment/virtualbox/pybox/helper/tests/test_vboxmanage.py
Daniel Caires 7b07a8a6b2 Assigned local port doesn't stop installation 
Ensure that if given local port is already in use,
rather than stopping installation, the script asks
if the user would like to overwrite or no the currently
rule using the local port.

Test Plan:
PASS: User press y to override the rule
PASS: User press n or any other key to ignore the creation
of a port forwarding rule and continue installation

Story: 2005051
Task: 48091

Change-Id: Iaabf0d5b713cf6d73a9fccdcf7da6962ac5620c2
Signed-off-by: Daniel Caires <daniel.caires@encora.com>
2023-07-05 11:24:17 -03:00

1581 lines
48 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-compute-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-compute-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.hostname = "test-host"
self.vm_config = {
"uartbase": "0x3F8",
"uartport": "4",
"uartmode": "file",
"uartpath": "/path/to/uart/",
"prefix": "test-prefix"
}
@patch('vboxmanage.env')
@patch('vboxmanage.platform', new='win32')
def test_add_uart_windows(self, mock_env):
"""
Test _add_uart method for Windows platform
"""
mock_env.PORT = 1
result = vboxmanage._add_uart(self.hostname, self.vm_config)
expected = [
'--uart1', '0x3F8', '4', '--uartmode1', 'file', '1'
]
self.assertCountEqual(result, expected)
@patch('vboxmanage.platform', new='linux')
def test_add_uart_linux_controller(self):
"""
Test _add_uart method for Linux platform with 'controller-0' in hostname
"""
self.hostname = "test-host-controller-0"
result = vboxmanage._add_uart(self.hostname, self.vm_config)
expected = [
'--uart1', '0x3F8', '4', '--uartmode1', 'file',
'/path/to/uart/test-prefix_test-host-controller-0_serial'
]
self.assertCountEqual(result, expected)
@patch('vboxmanage.platform', new='linux')
def test_add_uart_linux_no_controller(self):
"""
Test _add_uart method for Linux platform without 'controller-0' in hostname
"""
result = vboxmanage._add_uart(self.hostname, self.vm_config)
expected = [
'--uart1', '0x3F8', '4', '--uartmode1', 'file',
'/path/to/uart/test-prefix_test-host'
]
self.assertCountEqual(result, expected)
@patch('vboxmanage.platform', new='linux')
def test_add_uart_linux_no_prefix(self):
"""
Test _add_uart method for Linux platform without prefix in vm_config
"""
del self.vm_config["prefix"]
result = vboxmanage._add_uart(self.hostname, self.vm_config)
expected = [
'--uart1', '0x3F8', '4', '--uartmode1', 'file',
'/path/to/uart/test-host'
]
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')
def test_vboxmanage_addportforward_error(self, mock_log, 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_log.assert_any_call(
"Creating port-forwarding rule to: %s", "rule1:tcp:[]:8080:[10.10.10.1]:80")
mock_log.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')
def test_vboxmanage_deleteportforward_error(self, mock_log, 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_log.assert_any_call(
"Removing previous forwarding rule '%s' from NAT network '%s'", "rule1", "NatNetwork")
mock_log.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