diff --git a/pylint.rc b/pylint.rc index d4c7e6b..cca6ece 100755 --- a/pylint.rc +++ b/pylint.rc @@ -8,6 +8,7 @@ rcfile=pylint.rc # Add files or directories to the blacklist. Should be base names, not paths. ignore=tests +ignore-patterns=test_* # Pickle collected data for later comparisons. persistent=yes diff --git a/virtualbox/pybox/helper/tests/test_install_vbox.py b/virtualbox/pybox/helper/tests/test_install_vbox.py new file mode 100644 index 0000000..b5bdf68 --- /dev/null +++ b/virtualbox/pybox/helper/tests/test_install_vbox.py @@ -0,0 +1,128 @@ +""" +Unit tests related to install_vbox +""" + +import unittest +from unittest.mock import patch +from install_vbox import create_port_forward + + +class CreateportforwardTestCase(unittest.TestCase): + """ + Class to test create_port_forward method + """ + + def setUp(self): + """ + Method to set up the parameters used on the tests in this class + """ + + # Set up the test parameters + self.hostname = "TestVM" + self.network = "NatNetwork" + self.local_port = "8080" + self.guest_port = "80" + self.guest_ip = "10.10.10.1" + self.rule_name = "Rule1" + + @patch("helper.vboxmanage.vboxmanage_addportforward") + def test_successful_forwarding(self, mock_addport): + """ + Test create_port_forward method with successfull creation + """ + + # Setup + mock_addport.return_value = True + + # Run + create_port_forward(self.hostname, self.network, self.local_port, self.guest_port, self.guest_ip) + + # Assert + mock_addport.assert_called_once_with('TestVM', '8080', '10.10.10.1', '80', 'NatNetwork') + + @patch("helper.vboxmanage.vboxmanage_getrulename") + @patch("helper.vboxmanage.vboxmanage_addportforward") + @patch("helper.vboxmanage.vboxmanage_deleteportforward", return_value=None) + @patch('utils.install_log.LOG.info') + @patch("builtins.input") + def test_rewrite_rule(self, mock_input, mock_log, mock_deleteport, mock_addport, mock_getrule): + """ + Test create_port_forward method that fails to create rule and the user input 'y' to rewrite the existing rule + """ + + # Setup + mock_input.return_value = 'y' + mock_addport.side_effect = [False, True] + mock_getrule.return_value = "Rule1" + + # Run + result = create_port_forward(self.hostname, self.network, self.local_port, self.guest_port, self.guest_ip) + + # Assert + mock_log.assert_any_call( + "Trying to create a port-forwarding rule with port: %s, but it is already in use with rule name: %s", "8080", "Rule1") + mock_log.assert_any_call("Rewrite rule? (y/n)") + + mock_getrule.assert_called_once_with('NatNetwork', '8080') + mock_deleteport.assert_called_once_with('Rule1', 'NatNetwork') + mock_addport.assert_called_with('TestVM', '8080', '10.10.10.1', '80', 'NatNetwork') + + self.assertIsNone(result) + + @patch("helper.vboxmanage.vboxmanage_getrulename") + @patch("helper.vboxmanage.vboxmanage_addportforward") + @patch('utils.install_log.LOG.info') + @patch("builtins.input") + def test_dont_rewrite_rule(self, mock_input, mock_log, mock_addport, mock_getrule): + """ + Test create_port_forward method that fails to create rule and the user input 'n' to rewrite the existing rule + """ + + # Setup + mock_input.return_value = 'n' + mock_addport.return_value = False + mock_getrule.return_value = "Rule1" + + # Run + result = create_port_forward(self.hostname, self.network, self.local_port, self.guest_port, self.guest_ip) + + # Assert + mock_log.assert_any_call( + "Trying to create a port-forwarding rule with port: %s, but it is already in use with rule name: %s", "8080", "Rule1") + mock_log.assert_any_call("Rewrite rule? (y/n)") + mock_log.assert_any_call("Ignoring the creation of the port-forward rule and continuing installation!") + + mock_getrule.assert_called_once_with('NatNetwork', '8080') + mock_addport.assert_called_once_with('TestVM', '8080', '10.10.10.1', '80', 'NatNetwork') + + self.assertIsNone(result) + + @patch("helper.vboxmanage.vboxmanage_getrulename") + @patch("helper.vboxmanage.vboxmanage_addportforward") + @patch('utils.install_log.LOG.info') + @patch("sys.exit") + def test_norule(self, mock_exit, mock_log, mock_addport, mock_getrule): + """ + Test create_port_forward method that fails to create rule and no existing rule is found + """ + + # Setup + mock_addport.return_value = False + mock_getrule.return_value = "" + + mock_exit.side_effect = SystemExit(1) + + # Run + with self.assertRaises(SystemExit) as cm: + create_port_forward(self.hostname, self.network, self.local_port, self.guest_port, self.guest_ip) + + # Assert + self.assertEqual(cm.exception.code, 1) + + mock_log.assert_any_call( + "Could not add a port-forwarding rule using port %s, and could not find any rule already using it. Check the Nat Network and/or local port.", "8080") + mock_log.assert_any_call("Aborting!") + + +if __name__ == '__main__': + unittest.main() diff --git a/virtualbox/pybox/helper/tests/test_vboxmanage.py b/virtualbox/pybox/helper/tests/test_vboxmanage.py index 508001e..88c169c 100644 --- a/virtualbox/pybox/helper/tests/test_vboxmanage.py +++ b/virtualbox/pybox/helper/tests/test_vboxmanage.py @@ -3,6 +3,7 @@ Unit tests related to vboxmanage """ import unittest +import subprocess from unittest.mock import patch, call import vboxmanage @@ -953,56 +954,6 @@ class ContainsValueTestCase(unittest.TestCase): self.assertFalse(result) -class PortForwardTestCase(unittest.TestCase): - """ - Class to test vboxmanage_port_forward method - """ - - @patch('vboxmanage.subprocess') - def test_vboxmanage_port_forward(self, mock_subprocess): - """ - Test vboxmanage_port_forward method - """ - - hostname = "test-host" - network = "natnet1" - local_port = 1022 - guest_port = 22 - guest_ip = "192.168.15.5" - - # Run - vboxmanage.vboxmanage_port_forward(hostname, network, local_port, guest_port, guest_ip) - - # Assert - rule_name = f"{hostname}-{guest_port}" - - # Assert delete command was called - delete_cmd = [ - "vboxmanage", - "natnetwork", - "modify", - "--netname", - network, - "--port-forward-4", - "delete", - rule_name, - ] - mock_subprocess.check_output.assert_any_call(delete_cmd, stderr=mock_subprocess.STDOUT) - - # Assert create command was called - rule = f"{rule_name}:tcp:[]:{local_port}:[{guest_ip}]:{guest_port}" - create_cmd = [ - "vboxmanage", - "natnetwork", - "modify", - "--netname", - network, - "--port-forward-4", - rule, - ] - mock_subprocess.check_output.assert_any_call(create_cmd, stderr=mock_subprocess.STDOUT) - - class StorageCtlTestCase(unittest.TestCase): """ Class to test vboxmanage_storagectl method @@ -1400,6 +1351,230 @@ class RestoreSnapshotTestCase2(unittest.TestCase): 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() diff --git a/virtualbox/pybox/helper/vboxmanage.py b/virtualbox/pybox/helper/vboxmanage.py index 9cb48b6..41d878b 100644 --- a/virtualbox/pybox/helper/vboxmanage.py +++ b/virtualbox/pybox/helper/vboxmanage.py @@ -531,60 +531,6 @@ def _contains_value(key, dictionary): return key in dictionary and dictionary[key] -def vboxmanage_port_forward(hostname, network, local_port, guest_port, guest_ip): - """ - Configures port forwarding for a NAT network in VirtualBox. - - Args: - hostname (str): Name of the virtual machine. - network (str): Name of the NAT network. - local_port (int): The local port number to forward. - guest_port (int): The port number on the guest to forward to. - guest_ip (str): The IP address of the guest to forward to. - - Returns: - None - """ - - # VBoxManage natnetwork modify --netname natnet1 --port-forward-4 - # "ssh:tcp:[]:1022:[192.168.15.5]:22" - rule_name = f"{hostname}-{guest_port}" - # Delete previous entry, if any - LOG.info( - "Removing previous forwarding rule '%s' from NAT network '%s'", - rule_name, - network, - ) - cmd = [ - "vboxmanage", - "natnetwork", - "modify", - "--netname", - network, - "--port-forward-4", - "delete", - rule_name, - ] - try: - subprocess.check_output(cmd, stderr=subprocess.STDOUT) - except subprocess.CalledProcessError: - pass - - # Add new rule - rule = f"{rule_name}:tcp:[]:{local_port}:[{guest_ip}]:{guest_port}" - LOG.info("Updating port-forwarding rule to: %s", rule) - cmd = [ - "vboxmanage", - "natnetwork", - "modify", - "--netname", - network, - "--port-forward-4", - rule, - ] - subprocess.check_output(cmd, stderr=subprocess.STDOUT) - - def vboxmanage_storagectl(hostname=None, storectl="sata", hostiocache="off"): """ This creates a storage controller on the host. @@ -876,3 +822,104 @@ def vboxmanage_restoresnapshot(host=None, name=None): ["vboxmanage", "snapshot", host, "restore", name], stderr=subprocess.STDOUT ) time.sleep(10) + + +def vboxmanage_getrulename(network, local_port): + """ + Get port-forwarding rule for given NAT network and local port in VirtualBox. + + Args: + network (str): Name of the NAT network. + local_port (str): The local port number. + + Returns: + (str): Name of rule or empty + """ + # List information about all nat networks in VirtualBox + cmd = ["vboxmanage", "list", "natnets"] + result = subprocess.check_output(cmd, stderr=subprocess.STDOUT) + natpattern = r"NetworkName:(.*?)loopback mappings \(ipv4\)" + natnetworks = re.findall(natpattern,result.decode(),re.DOTALL) + + # Get the rule name of the given local port in the given natnetwork + for natnetwork in natnetworks: + natinfo = natnetwork.strip().split('\n') + if natinfo[0] == network: + try: + startindex = natinfo.index("Port-forwarding (ipv4)") + except ValueError: + # If no index is found the function return an empty string + return "" + for index in range (startindex+1,len(natinfo)): + rule = natinfo[index].strip() + + parsed_rule = rule.split(':') + if int(parsed_rule[3]) == int(local_port): + return parsed_rule[0] + return "" + + +def vboxmanage_addportforward(rule_name, local_port, guest_ip, guest_port, network): + """ + Add port-forwarding rule for a NAT network in VirtualBox. + + Args: + rule_name (str): Name of the port-forward rule to be added. + local_port (str): The local port number to forward. + guest_ip (str): The IP address of the guest to forward to. + guest_port (str): The port number on the guest to forward to. + network (str): Name of the NAT network. + + Returns: + True if the port was added + False if an error occurred when trying to add the port-forward rule. + """ + rule = f"{rule_name}:tcp:[]:{local_port}:[{guest_ip}]:{guest_port}" + LOG.info("Creating port-forwarding rule to: %s", rule) + cmd = [ + "vboxmanage", + "natnetwork", + "modify", + "--netname", + network, + "--port-forward-4", + rule, + ] + try: + subprocess.check_output(cmd, stderr=subprocess.STDOUT) + except subprocess.CalledProcessError: + LOG.info("Error while trying to create port-forwarding rule. Continuing installation!") + return False + return True + + +def vboxmanage_deleteportforward(rule_name, network): + """ + Delete port-forwarding rule for a NAT network in VirtualBox. + + Args: + rule_name (str): Name of the port-forward rule to be deleted. + network (str): Name of the NAT network. + + Returns: + None + """ + LOG.info( + "Removing previous forwarding rule '%s' from NAT network '%s'", + rule_name, + network, + ) + cmd = [ + "vboxmanage", + "natnetwork", + "modify", + "--netname", + network, + "--port-forward-4", + "delete", + rule_name, + ] + try: + subprocess.check_output(cmd, stderr=subprocess.STDOUT) + except subprocess.CalledProcessError: + LOG.info("Error while trying to delete port-forwarding rule. Continuing installation!") diff --git a/virtualbox/pybox/install_vbox.py b/virtualbox/pybox/install_vbox.py index 3ca5fd3..5db7b05 100755 --- a/virtualbox/pybox/install_vbox.py +++ b/virtualbox/pybox/install_vbox.py @@ -268,6 +268,41 @@ def get_disk_sizes(comma_list): return sizes +def create_port_forward(hostname, network, local_port, guest_port, guest_ip): + """ + Create a port forwarding rule for a NAT network in VirtualBox. + + Args: + hostname (str): Name of the virtual machine. + network (str): Name of the NAT network. + local_port (str): The local port number to forward. + guest_port (str): The port number on the guest to forward to. + guest_ip (str): The IP address of the guest to forward to. + + Returns: + None + """ + if not vboxmanage.vboxmanage_addportforward(hostname, local_port, guest_ip, guest_port, network): + rule_name = vboxmanage.vboxmanage_getrulename(network, local_port) + if not rule_name: + LOG.info( + "Could not add a port-forwarding rule using port %s, and could not find any rule already using it. Check the Nat Network and/or local port.", local_port) + LOG.info("Aborting!") + sys.exit(1) + LOG.info( + "Trying to create a port-forwarding rule with port: %s, but it is already in use with rule name: %s", + local_port, + rule_name) + + LOG.info("Rewrite rule? (y/n)") + choice = input().lower() + if choice == 'y': + vboxmanage.vboxmanage_deleteportforward(rule_name, network) + vboxmanage.vboxmanage_addportforward(hostname, local_port, guest_ip, guest_port, network) + else: + LOG.info("Ignoring the creation of the port-forward rule and continuing installation!") + + # pylint: disable=too-many-locals, too-many-branches, too-many-statements def create_lab(m_vboxoptions): """ @@ -424,13 +459,16 @@ def create_lab(m_vboxoptions): # Add port forward rule for StarlingX dashboard visualization at 8080 rule_name = m_vboxoptions.labname + "-horizon-dashbord" - vboxmanage.vboxmanage_port_forward(rule_name, - m_vboxoptions.vboxnet_name, local_port=m_vboxoptions.horizon_dashboard_port, guest_port='8080', guest_ip=ip_addr) + create_port_forward(rule_name, + m_vboxoptions.vboxnet_name, + local_port=m_vboxoptions.horizon_dashboard_port, + guest_port='8080', + guest_ip=ip_addr) elif 'controller-1' in node: local_port = m_vboxoptions.nat_controller1_local_ssh_port ip_addr = m_vboxoptions.controller1_ip - vboxmanage.vboxmanage_port_forward( + create_port_forward( node, m_vboxoptions.vboxnet_name, local_port=local_port, @@ -443,7 +481,7 @@ def create_lab(m_vboxoptions): local_port = m_vboxoptions.nat_controller_floating_local_ssh_port ip_addr = m_vboxoptions.controller_floating_ip name = m_vboxoptions.labname + 'controller-float' - vboxmanage.vboxmanage_port_forward(name, m_vboxoptions.vboxnet_name, + create_port_forward(name, m_vboxoptions.vboxnet_name, local_port=local_port, guest_port='22', guest_ip=ip_addr) ctrlr0 = m_vboxoptions.labname + '-controller-0' @@ -1378,9 +1416,11 @@ def stage_enable_kubernetes(ssh_client): ip_addr = V_BOX_OPTIONS.controller0_ip rule_name = V_BOX_OPTIONS.labname + "-kubernetes-dasboard" - vboxmanage.vboxmanage_port_forward(rule_name, V_BOX_OPTIONS.vboxnet_name, - local_port=V_BOX_OPTIONS.kubernetes_dashboard_port, - guest_port='32000', guest_ip=ip_addr) + create_port_forward(rule_name, + V_BOX_OPTIONS.vboxnet_name, + local_port=V_BOX_OPTIONS.kubernetes_dashboard_port, + guest_port='32000', + guest_ip=ip_addr) LOG.info("###### Installing Kubernetes dashboard ######")