Add tests for netpreserver

coriolis/osmorphing/debian.py

@@ -100,69 +100,6 @@ class BaseDebianMorphingTools(base.BaseLinuxOSMorphingTools):
             }
         return yaml.dump(cfg, default_flow_style=False)
 
-    def _preserve_static_netplan_configuration(self, nics_info):
-        ips_info = {}
-        for nic in nics_info:
-            mac_address = nic.get('mac_address')
-            if not mac_address:
-                LOG.warning(
-                    f"No MAC address found for NIC with ID '{nic.get('id')}. "
-                    f"This NIC will be skipped from static IP configuration")
-                continue
-
-            nic_ips = nic.get('ip_addresses', [])
-            if not nic_ips:
-                LOG.warning(
-                    f"Skipping NIC ('{mac_address}'). It has no detected IP "
-                    f"addresses")
-                continue
-            for nic_ip in nic_ips:
-                ips_info[nic_ip] = mac_address
-        if not ips_info:
-            LOG.warning("No IP information found for instance. Skipping "
-                        "static configuration")
-            return
-
-        for netfile in self._list_dir(self.netplan_base):
-            config_changed = False
-            if netfile.endswith('.yaml') or netfile.endswith('.yml'):
-                pth = f"{self.netplan_base}/{netfile}"
-                contents = self._read_file_sudo(pth).decode()
-                config = yaml.load(contents, Loader=yaml.SafeLoader)
-                ethernets = config.get('network', {}).get('ethernets', {})
-                for eth_name, eth_config in ethernets.items():
-                    ips = eth_config.get('addresses', [])
-                    for eth_ip in ips:
-                        # NOTE(dvincze): addresses can also be objects, where
-                        # netplan can set IP label and lifetime.
-                        if isinstance(eth_ip, dict):
-                            eth_ip_keys = list(eth_ip.keys())
-                            if not eth_ip_keys:
-                                LOG.warning(
-                                    "Found empty IP address object entry. "
-                                    "Skipping")
-                                continue
-                            eth_ip = eth_ip_keys[0]
-
-                        addr = eth_ip.split('/')[0]
-                        if ips_info.get(addr):
-                            LOG.debug(
-                                f"Found IP match for NIC {eth_name} for "
-                                f"'{addr}'")
-                            mac_addr = ips_info[addr]
-                            if not eth_config.get('match'):
-                                eth_config['match'] = dict()
-                            LOG.debug(f"Setting '{mac_addr}' MAC address "
-                                      f"match field for NIC '{eth_name}'")
-                            eth_config['match']['macaddress'] = mac_addr
-                            eth_config['set-name'] = eth_name
-                            config_changed = True
-
-                if config_changed:
-                    self._exec_cmd_chroot(f'cp {pth} {pth}.bak')
-                    new_config = yaml.dump(config, Dumper=yaml.SafeDumper)
-                    self._write_file_sudo(pth, new_config)
-
     def set_net_config(self, nics_info, dhcp):
         if not dhcp:
             LOG.info("Setting static IP configuration")

coriolis/osmorphing/netpreserver/base.py

@@ -11,5 +11,5 @@ class BaseNetPreserver(object):
     def check_net_preserver(self):
         pass
 
-    def parse_network(self, nics_info):
+    def parse_network(self):
         pass

coriolis/osmorphing/netpreserver/interfaces.py

@@ -29,6 +29,7 @@ class InterfacesNetPreserver(base.BaseNetPreserver):
         def _parse_iface_file(interface_file):
             nonlocal paths
             curr_ip_address = None
+            curr_iface = None
             interfaces_contents = self.osmorphing_tool._read_file_sudo(
                 interface_file).decode()
             LOG.debug(

coriolis/tests/osmorphing/netpreserver/test_factory.py

@@ -0,0 +1,48 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import ddt
+from unittest import mock
+
+from coriolis.osmorphing.netpreserver import factory
+from coriolis.osmorphing.netpreserver import ifcfg
+from coriolis.osmorphing.netpreserver import interfaces
+from coriolis.osmorphing.netpreserver import netplan
+from coriolis.osmorphing.netpreserver import nmconnection
+from coriolis.tests import test_base
+
+
+@ddt.ddt
+class GetNetPreserverTestCase(test_base.CoriolisBaseTestCase):
+    def setUp(self):
+        super(GetNetPreserverTestCase, self).setUp()
+        self.osmorphing_tool = mock.MagicMock()
+
+    @ddt.data(
+        # (netplan, nmconnection, ifcfg, interfaces, expected_result)
+        (True, False, False, False, netplan.NetplanNetPreserver),
+        (False, True, False, False, nmconnection.NmconnectionNetPreserver),
+        (False, False, True, False, ifcfg.IfcfgNetPreserver),
+        (False, False, False, True, interfaces.InterfacesNetPreserver),
+        (False, False, False, False, None)
+    )
+    @ddt.unpack
+    def test_get_net_preserver_first_match(self, return_netplan,
+                                           return_nmconnection, return_ifcfg,
+                                           return_interfaces,
+                                           result_class):
+        with mock.patch.object(netplan.NetplanNetPreserver,
+                               "check_net_preserver",
+                               return_value=return_netplan), \
+             mock.patch.object(nmconnection.NmconnectionNetPreserver,
+                               "check_net_preserver",
+                               return_value=return_nmconnection), \
+             mock.patch.object(ifcfg.IfcfgNetPreserver,
+                               "check_net_preserver",
+                               return_value=return_ifcfg), \
+             mock.patch.object(interfaces.InterfacesNetPreserver,
+                               "check_net_preserver",
+                               return_value=return_interfaces):
+
+            res = factory.get_net_preserver(self.osmorphing_tool)
+            self.assertEqual(res, result_class)

coriolis/tests/osmorphing/netpreserver/test_ifcfg.py

@@ -0,0 +1,161 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.osmorphing import base
+from coriolis.osmorphing.netpreserver import ifcfg
+from coriolis.osmorphing import redhat
+from coriolis.tests import test_base
+
+
+class CoriolisTestException(Exception):
+    pass
+
+
+class IfcfgNetPreserverTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the IfcfgNetPreserver class."""
+
+    def setUp(self):
+        super(IfcfgNetPreserverTestCase, self).setUp()
+        self.event_manager = mock.MagicMock()
+        self.detected_os_info = {
+            'os_type': 'linux',
+            'distribution_name': redhat.RED_HAT_DISTRO_IDENTIFIER,
+            'release_version': '6',
+            'friendly_release_name': mock.sentinel.friendly_release_name,
+        }
+        self.netpreserver = ifcfg.IfcfgNetPreserver(
+            redhat.BaseRedHatMorphingTools(
+                mock.sentinel.conn, mock.sentinel.os_root_dir,
+                mock.sentinel.os_root_dir, mock.sentinel.hypervisor,
+                self.event_manager, self.detected_os_info,
+                mock.sentinel.osmorphing_parameters,
+                mock.sentinel.operation_timeout))
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_read_config_file')
+    @mock.patch.object(ifcfg.IfcfgNetPreserver, '_get_net_config_files')
+    def test_get_ifcfgs_by_type(self, mock_get_net_config_files,
+                                mock_read_config_file):
+        mock_get_net_config_files.return_value = [mock.sentinel.ifcfg_file]
+        mock_read_config_file.side_effect = [{"TYPE": "Ethernet"}]
+
+        result = self.netpreserver._get_ifcfgs_by_type(
+            "Ethernet", self.netpreserver.network_scripts_path)
+
+        mock_read_config_file.assert_called_once_with(mock.sentinel.ifcfg_file)
+        mock_get_net_config_files.assert_called_once_with(
+            self.netpreserver.network_scripts_path)
+
+        self.assertEqual(
+            result, [(mock.sentinel.ifcfg_file, {"TYPE": "Ethernet"})])
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_list_dir')
+    def test_get_net_config_files(self, mock_list_dir):
+        mock_list_dir.return_value = ['ifcfg-eth0', 'ifcfg-lo', 'other-file']
+
+        result = self.netpreserver._get_net_config_files(
+            self.netpreserver.network_scripts_path)
+
+        expected_result = [
+            'etc/sysconfig/network-scripts/ifcfg-eth0',
+            'etc/sysconfig/network-scripts/ifcfg-lo'
+        ]
+
+        mock_list_dir.assert_called_once_with(
+            self.netpreserver.network_scripts_path)
+
+        self.assertEqual(result, expected_result)
+
+    @mock.patch.object(ifcfg.IfcfgNetPreserver, '_get_ifcfgs_by_type')
+    @mock.patch.object(ifcfg.IfcfgNetPreserver, '_get_net_config_files')
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
+    def test_check_net_preserver_True(self, mock_test_path,
+                                      mock_get_net_config_files,
+                                      mock_get_ifcfgs_by_type) -> None:
+
+        mock_test_path.return_value = True
+        mock_get_net_config_files.return_value = [
+            "etc/sysconfig/network-scripts/ifcfg-eth0",
+            "etc/sysconfig/network-scripts/ifcfg-lo"
+        ]
+        mock_get_ifcfgs_by_type.return_value = [
+            {
+                "TYPE": "Ethernet",
+                "DEVICE": "eth0",
+                "HWADDR": "00:11:22:33:44:55",
+                "IPADDR": "192.168.1.10"
+            },
+            {
+                "TYPE": "Ethernet",
+                "HWADDR": "AA:BB:CC:DD:EE:FF",
+                "IPADDR": "192.168.1.11"
+            }]
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertTrue(result)
+
+    @mock.patch.object(ifcfg.IfcfgNetPreserver, '_get_ifcfgs_by_type')
+    @mock.patch.object(ifcfg.IfcfgNetPreserver, '_get_net_config_files')
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
+    def test_check_net_preserver_no_ifcfg_files(self, mock_test_path,
+                                                mock_get_net_config_files,
+                                                mock_get_ifcfgs_by_type):
+        mock_test_path.return_value = True
+        mock_get_net_config_files.return_value = [
+            "etc/sysconfig/network-scripts/ifcfg-eth0",
+            "etc/sysconfig/network-scripts/ifcfg-lo"
+        ]
+        mock_get_ifcfgs_by_type.return_value = []
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertFalse(result)
+
+    @mock.patch.object(ifcfg.IfcfgNetPreserver, '_get_ifcfgs_by_type')
+    def test_parse_network(self, mock_get_ifcfgs_by_type):
+        ifcfg_file_with_device = "etc/sysconfig/network-scripts/ifcfg-eth0"
+        ifcfg_with_device = {
+            "TYPE": "Ethernet",
+            "DEVICE": "eth0",
+            "HWADDR": "00:11:22:33:44:55",
+            "IPADDR": "192.168.1.10"
+        }
+        ifcfg_file_without_device = "etc/sysconfig/network-scripts/ifcfg-eth1"
+        ifcfg_without_device = {
+            "TYPE": "Ethernet",
+            "HWADDR": "AA:BB:CC:DD:EE:FF",
+            "IPADDR": "192.168.1.11"
+        }
+        ifcfg_file_without_hwaddr = "etc/sysconfig/network-scripts/ifcfg-eth2"
+        ifcfg_without_hwaddr = {
+            "TYPE": "Ethernet",
+            "DEVICE": "eth2",
+            "IPADDR": "192.168.1.12"
+        }
+
+        mock_get_ifcfgs_by_type.return_value = [
+            (ifcfg_file_with_device, ifcfg_with_device),
+            (ifcfg_file_without_device, ifcfg_without_device),
+            (ifcfg_file_without_hwaddr, ifcfg_without_hwaddr)
+        ]
+
+        self.netpreserver.parse_network()
+
+        expected_info = {
+            "eth0": {
+                "mac_address": "00:11:22:33:44:55",
+                "ip_addresses": ["192.168.1.10"]
+            },
+            "eth1": {
+                "mac_address": "AA:BB:CC:DD:EE:FF",
+                "ip_addresses": ["192.168.1.11"]
+            },
+            "eth2": {
+                "mac_address": None,
+                "ip_addresses": ["192.168.1.12"],
+            }
+        }
+
+        self.assertEqual(self.netpreserver.interface_info, expected_info)

coriolis/tests/osmorphing/netpreserver/test_interfaces.py

@@ -0,0 +1,146 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.osmorphing import debian
+from coriolis.osmorphing.netpreserver import interfaces
+from coriolis.tests import test_base
+
+
+class InterfacesNetPreserverTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the InterfacesNetPreserver class."""
+
+    def setUp(self):
+        super(InterfacesNetPreserverTestCase, self).setUp()
+        self.event_manager = mock.MagicMock()
+        self.detected_os_info = {
+            'os_type': 'linux',
+            "release_version": '10',
+            'distribution_name': debian.DEBIAN_DISTRO_IDENTIFIER,
+            'friendly_release_name': mock.sentinel.friendly_release_name,
+        }
+        self.netpreserver = interfaces.InterfacesNetPreserver(
+            debian.BaseDebianMorphingTools(
+                mock.sentinel.conn,
+                mock.sentinel.os_root_dir,
+                mock.sentinel.os_root_dir,
+                mock.sentinel.hypervisor,
+                self.event_manager,
+                self.detected_os_info,
+                mock.sentinel.osmorphing_parameters,
+                mock.sentinel.operation_timeout
+            )
+        )
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_check_net_preserver_true(self, mock_test_path, mock_list_dir):
+        mock_test_path.return_value = True
+        mock_list_dir.return_value = ["interfaces"]
+
+        result = self.netpreserver.check_net_preserver()
+
+        mock_test_path.assert_called_once_with(self.netpreserver.ifaces_file)
+        mock_list_dir.assert_called_once_with('etc/network')
+        self.assertTrue(result)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_check_net_preserver_false_no_file(self, mock_test_path,
+                                               mock_list_dir):
+        mock_test_path.return_value = False
+        mock_list_dir.return_value = []
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertFalse(result)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_read_file_sudo')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_parse_network_basic(self, mock_test_path, mock_exec_cmd_chroot,
+                                 mock_read_file_sudo):
+        interfaces_content = (
+            "iface eth0 inet static\n"
+            "hwaddress ether 00:11:22:33:44:55\n"
+            "address 192.168.1.10/24\n"
+            "\n"
+            "iface eth1 inet dhcp\n"
+            "address 192.168.1.20/24\n"
+        )
+        mock_read_file_sudo.return_value = interfaces_content.encode()
+        mock_test_path.return_value = True
+        mock_exec_cmd_chroot.return_value = ""
+
+        self.netpreserver.interface_info = {}
+        self.netpreserver.parse_network()
+
+        expected_info = {
+            "eth0": {
+                "mac_address": "00:11:22:33:44:55",
+                "ip_addresses": ["192.168.1.10"]
+            },
+            "eth1": {
+                "mac_address": "",
+                "ip_addresses": ["192.168.1.20"]
+            }
+        }
+        self.assertEqual(self.netpreserver.interface_info, expected_info)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_read_file_sudo')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_parse_network_with_source(self, mock_test_path,
+                                       mock_exec_cmd_chroot,
+                                       mock_read_file_sudo):
+        main_content = (
+            "iface eth0 inet static\n"
+            "hwaddress ether 00:11:22:33:44:55\n"
+            "address 192.168.1.10/24\n"
+            "source /etc/network/interfaces.d/\n"
+        )
+        extra_file = "extra_iface"
+        extra_content = (
+            "iface eth1 inet dhcp\n"
+            "address 192.168.1.20/24\n"
+        )
+
+        mock_test_path.return_value = True
+        mock_exec_cmd_chroot.return_value = extra_file
+        mock_read_file_sudo.side_effect = (
+            main_content.encode(), extra_content.encode()
+        )
+
+        self.netpreserver.parse_network()
+
+        expected_info = {
+            "eth0": {
+                "mac_address": "00:11:22:33:44:55",
+                "ip_addresses": ["192.168.1.10"]
+            },
+            "eth1": {
+                "mac_address": "",
+                "ip_addresses": ["192.168.1.20"]
+            }
+        }
+        self.assertEqual(self.netpreserver.interface_info, expected_info)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_read_file_sudo')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_parse_network_with_trailing_hwaddress(self, mock_test_path,
+                                                   mock_exec_cmd_chroot,
+                                                   mock_read_file_sudo):
+
+        interfaces_content = (
+            "hwaddress ether 00:11:22:33:44:55\n"
+        )
+        mock_read_file_sudo.return_value = interfaces_content.encode()
+        mock_test_path.return_value = True
+        mock_exec_cmd_chroot.return_value = ""
+
+        with self.assertLogs(level='WARNING'):
+            self.netpreserver.parse_network()
+
+        self.assertEqual(self.netpreserver.interface_info, {})

coriolis/tests/osmorphing/netpreserver/test_netplan.py

@@ -0,0 +1,166 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.osmorphing import debian
+from coriolis.osmorphing.netpreserver import netplan
+from coriolis.tests import test_base
+
+
+class NetplanNetPreserverTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the NetplanNetPreserver class."""
+
+    def setUp(self):
+        super(NetplanNetPreserverTestCase, self).setUp()
+        self.event_manager = mock.MagicMock()
+        self.detected_os_info = {
+            'os_type': 'linux',
+            "release_version": '10',
+            'distribution_name': debian.DEBIAN_DISTRO_IDENTIFIER,
+            'friendly_release_name': mock.sentinel.friendly_release_name,
+        }
+        self.netpreserver = netplan.NetplanNetPreserver(
+            debian.BaseDebianMorphingTools(
+                mock.sentinel.conn,
+                mock.sentinel.os_root_dir,
+                mock.sentinel.os_root_dir,
+                mock.sentinel.hypervisor,
+                self.event_manager,
+                self.detected_os_info,
+                mock.sentinel.osmorphing_parameters,
+                mock.sentinel.operation_timeout
+            )
+        )
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_check_net_preserver_true(self, mock_test_path, mock_list_dir):
+        mock_test_path.return_value = True
+        mock_list_dir.return_value = ["01-netplan.yaml", "notconfig.txt",
+                                      "02-other.yml"]
+
+        result = self.netpreserver.check_net_preserver()
+
+        mock_test_path.assert_called_once_with(self.netpreserver.netplan_base)
+        mock_list_dir.assert_called_once_with(self.netpreserver.netplan_base)
+        self.assertTrue(result)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_check_net_preserver_false_no_file(self, mock_test_path,
+                                               mock_list_dir):
+        mock_test_path.return_value = True
+        mock_list_dir.return_value = ["config.txt", "README.md"]
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertFalse(result)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
+    def test_check_net_preserver_false_no_directory(self, mock_test_path,
+                                                    mock_list_dir):
+        mock_test_path.return_value = False
+        mock_list_dir.return_value = []
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertFalse(result)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_read_file_sudo')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    def test_parse_network(self, mock_list_dir, mock_read_file_sudo):
+        mock_list_dir.return_value = ["01-netplan.yaml"]
+        yaml_content = """
+            network:
+            ethernets:
+                eth0:
+                    match:
+                        macaddress: "00:11:22:33:44:55"
+                    addresses:
+                        - "192.168.1.10/24"
+                eth1:
+                    addresses:
+                        - "192.168.1.20/24"
+                        - "192.168.1.21/24"
+                eth2:
+                    match:
+                        macaddress: "AA:BB:CC:DD:EE:FF"
+                eth4:
+                    addresses:
+                        - "192.168.1.31/24":
+                eth3:
+                    addresses:
+                        - "192.168.1.30/24":
+                            lifetime: 0
+                            label: "eth3"
+                eth5:
+                    addresses:
+                        - "192.168.1.32/24"
+                        - {}
+                eth6:
+                    addresses:
+                        - {}
+
+        """
+        mock_read_file_sudo.return_value = yaml_content.lstrip().encode()
+
+        self.netpreserver.parse_network()
+
+        expected_info = {
+            "eth0": {
+                "mac_address": "00:11:22:33:44:55",
+                "ip_addresses": ["192.168.1.10"]
+            },
+            "eth1": {
+                "mac_address": "",
+                "ip_addresses": ["192.168.1.20", "192.168.1.21"]
+            },
+            "eth2": {
+                "mac_address": "AA:BB:CC:DD:EE:FF",
+                "ip_addresses": []
+            },
+            "eth3": {
+                "mac_address": "",
+                "ip_addresses": ["192.168.1.30"]
+            },
+            "eth4": {
+                "mac_address": "",
+                "ip_addresses": ["192.168.1.31"]
+            },
+            "eth5": {
+                "mac_address": "",
+                "ip_addresses": ["192.168.1.32"]
+            },
+            "eth6": {
+                "mac_address": "",
+                "ip_addresses": []
+            }
+        }
+
+        self.assertEqual(self.netpreserver.interface_info, expected_info)
+
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_read_file_sudo')
+    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
+    def test_parse_network_invalid_file(self, mock_list_dir,
+                                        mock_read_file_sudo):
+        mock_list_dir.return_value = ["01-netplan.yaml"]
+        yaml_content = """
+            network:
+            ethernets:
+                eth0:
+                        match:
+                    macaddress: "00:11:22:33:44:55"
+                    addresses:
+                        - "192.168.1.10/24"
+        """
+
+        mock_read_file_sudo.return_value = yaml_content.lstrip().encode()
+
+        with self.assertLogs(level='WARNING'):
+            self.netpreserver.parse_network()
+
+        expected_info = {}
+
+        self.assertEqual(self.netpreserver.interface_info, expected_info)

coriolis/tests/osmorphing/netpreserver/test_nmconnection.py

@@ -0,0 +1,159 @@
+# Copyright 2025 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.osmorphing import base
+from coriolis.osmorphing.netpreserver import nmconnection
+from coriolis.osmorphing import redhat
+from coriolis.tests import test_base
+
+
+class NmconnectionNetPreserverTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the NmconnectionNetPreserver class."""
+
+    def setUp(self):
+        super(NmconnectionNetPreserverTestCase, self).setUp()
+        self.event_manager = mock.MagicMock()
+        self.detected_os_info = {
+            'os_type': 'linux',
+            'distribution_name': redhat.RED_HAT_DISTRO_IDENTIFIER,
+            'release_version': '6',
+            'friendly_release_name': mock.sentinel.friendly_release_name,
+        }
+        self.netpreserver = nmconnection.NmconnectionNetPreserver(
+            redhat.BaseRedHatMorphingTools(
+                mock.sentinel.conn, mock.sentinel.os_root_dir,
+                mock.sentinel.os_root_dir, mock.sentinel.hypervisor,
+                self.event_manager, self.detected_os_info,
+                mock.sentinel.osmorphing_parameters,
+                mock.sentinel.operation_timeout)
+        )
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_list_dir')
+    def test_get_nmconnection_files(self, mock_list_dir):
+        mock_list_dir.return_value = [
+            'eth0.nmconnection', 'eth1.nmconnection', 'other-file']
+        result = self.netpreserver._get_nmconnection_files(
+            self.netpreserver.nmconnection_file)
+        expected_result = [
+            'etc/NetworkManager/system-connections/eth0.nmconnection',
+            'etc/NetworkManager/system-connections/eth1.nmconnection'
+        ]
+        mock_list_dir.assert_called_once_with(
+            self.netpreserver.nmconnection_file)
+        self.assertEqual(result, expected_result)
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_read_config_file')
+    @mock.patch.object(nmconnection.NmconnectionNetPreserver,
+                       '_get_nmconnection_files')
+    def test_get_keyfiles_by_type(self, mock_get_nmconnection_files,
+                                  mock_read_config_file):
+        mock_get_nmconnection_files.return_value = [mock.sentinel.nmconn_file]
+        mock_read_config_file.side_effect = [{"type": "ethernet"}]
+
+        result = self.netpreserver._get_keyfiles_by_type(
+            "ethernet", self.netpreserver.nmconnection_file)
+
+        mock_get_nmconnection_files.assert_called_once_with(
+            self.netpreserver.nmconnection_file)
+        mock_read_config_file.assert_called_once_with(
+            mock.sentinel.nmconn_file)
+        self.assertEqual(result, [(mock.sentinel.nmconn_file,
+                                   {"type": "ethernet"})])
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
+    @mock.patch.object(nmconnection.NmconnectionNetPreserver,
+                       '_get_nmconnection_files')
+    @mock.patch.object(nmconnection.NmconnectionNetPreserver,
+                       '_get_keyfiles_by_type')
+    def test_check_net_preserver_True(self, mock_get_keyfiles_by_type,
+                                      mock_get_nmconnection_files,
+                                      mock_test_path):
+        mock_test_path.return_value = True
+        mock_get_nmconnection_files.return_value = ["eth0.nmconnection",
+                                                    "eth1.nmconnection"]
+        mock_get_keyfiles_by_type.return_value = [
+            (mock.sentinel.nmconn_file, {"type": "ethernet",
+                                         "connection": {"id": "eth0"}})
+        ]
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertTrue(result)
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
+    @mock.patch.object(nmconnection.NmconnectionNetPreserver,
+                       '_get_nmconnection_files')
+    def test_check_net_preserver_no_files(self, mock_get_nmconnection_files,
+                                          mock_test_path):
+        mock_test_path.return_value = True
+        mock_get_nmconnection_files.return_value = []
+
+        result = self.netpreserver.check_net_preserver()
+
+        self.assertFalse(result)
+
+    @mock.patch.object(nmconnection.NmconnectionNetPreserver,
+                       '_get_keyfiles_by_type')
+    def test_parse_network(self, mock_get_keyfiles_by_type):
+        nmconn_file_with_id = (
+            self.netpreserver.nmconnection_file + "/eth0.nmconnection"
+        )
+        nmconn_with_id = {
+            "connection": {"id": "eth0"},
+            "mac-address": "00:11:22:33:44:55",
+            "address1": "192.168.1.10/24"
+        }
+        nmconn_file_without_id = (
+            self.netpreserver.nmconnection_file + "/eth1.nmconnection"
+        )
+        nmconn_without_id = {
+            "mac-address": "AA:BB:CC:DD:EE:FF",
+            "Address2": "192.168.1.20/24, 192.168.1.21/24"
+        }
+        nmconn_file_without_mac_address = (
+            self.netpreserver.nmconnection_file + "/eth2.nmconnection"
+        )
+        nmconn_without_mac_address = {
+            "connection": {"id": "eth2"},
+            "address1": "192.168.1.30/24",
+        }
+        nmconn_file_without_mac_address_ip_address = (
+            self.netpreserver.nmconnection_file + "/eth3.nmconnection"
+        )
+        nmconn_without_mac_address_ip_address = {
+            "connection": {"id": "eth3"},
+        }
+        mock_get_keyfiles_by_type.return_value = [
+            (nmconn_file_with_id, nmconn_with_id),
+            (nmconn_file_without_id, nmconn_without_id),
+            (nmconn_file_without_mac_address, nmconn_without_mac_address),
+            (nmconn_file_without_mac_address_ip_address,
+             nmconn_without_mac_address_ip_address)
+        ]
+
+        self.netpreserver.interface_info = {}
+
+        self.netpreserver.parse_network()
+
+        expected_info = {
+            "eth0": {
+                "mac_address": "00:11:22:33:44:55",
+                "ip_addresses": ["192.168.1.10"]
+            },
+            "eth1": {
+                "mac_address": "AA:BB:CC:DD:EE:FF",
+                "ip_addresses": ["192.168.1.20", "192.168.1.21"]
+            },
+            "eth2": {
+                "mac_address": None,
+                "ip_addresses": ["192.168.1.30"]
+            },
+            "eth3": {
+                "mac_address": None,
+                "ip_addresses": []
+            }
+        }
+
+        self.assertEqual(self.netpreserver.interface_info, expected_info)

coriolis/tests/osmorphing/test_base.py

@@ -976,3 +976,78 @@ class BaseLinuxOSMorphingToolsTestBase(test_base.CoriolisBaseTestCase):
         mock_set_grub2_cmdline.assert_called_once_with(
             config_obj, ['console=tty0', 'console=ttyS0'])
         mock_apply_grub2_config.assert_called_once_with(config_obj, True)
+
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
+    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo')
+    def test__add_net_udev_rules(self, mock_write_file_sudo, mock_test_path):
+        mock_test_path.return_value = False
+        net_ifaces_info = [
+            ("eth0", "AA:BB:CC:DD:EE:FF"),
+            ("eth1", "FF:EE:DD:CC:BB:AA")
+        ]
+        content = (
+            'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", '
+            'ATTR{address}=="aa:bb:cc:dd:ee:ff", NAME="eth0"\n'
+            'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", '
+            'ATTR{address}=="ff:ee:dd:cc:bb:aa", NAME="eth1"\n'
+        )
+
+        self.os_morphing_tools._add_net_udev_rules(net_ifaces_info)
+
+        mock_write_file_sudo.assert_called_once_with(
+            "etc/udev/rules.d/70-persistent-net.rules", content
+        )
+
+    @ddt.data(
+        # (nics_info, expected_net_ifaces)
+        (
+            [
+                {"mac_address": "AA:BB:CC:DD:EE:FF",
+                 "ip_addresses": ["192.168.1.20"]},
+                {"mac_address": "00:11:22:33:44:55",
+                 "ip_addresses": ["192.168.1.11"]},
+                {"mac_address": "11:22:33:44:55:66",
+                 "ip_addresses": ["192.168.1.31"]},
+                {"mac_address": None,
+                 "ip_addresses": ["192.168.1.10"]},
+                {"mac_address": "FF:FF:FF:FF:FF:FF",
+                 "ip_addresses": ["192.168.1.10", "192.168.1.20"]}
+            ],
+            {
+                "eth0": "FF:FF:FF:FF:FF:FF",
+                "eth1": "FF:FF:FF:FF:FF:FF",
+                "eth2": "11:22:33:44:55:66"
+            }
+        ),
+    )
+    @ddt.unpack
+    def test__setup_network_preservation(self, nics_info, expected_net_ifaces):
+        class FakeNetPreserver:
+            def __init__(self, tool):
+                self.tool = tool
+                self.interface_info = {}
+
+            def parse_network(self):
+                self.interface_info = {
+                    "eth0": {"mac_address": "00:11:22:33:44:55",
+                             "ip_addresses": ["192.168.1.10"]},
+                    "eth1": {"mac_address": "AA:BB:CC:DD:EE:FF",
+                             "ip_addresses": ["192.168.1.20"]},
+                    "eth2": {"mac_address": "11:22:33:44:55:66",
+                             "ip_addresses": ["192.168.1.30"]},
+                }
+
+        with mock.patch(
+            "coriolis.osmorphing.netpreserver.factory.get_net_preserver",
+                return_value=FakeNetPreserver) as mock_get_np:
+
+            self.os_morphing_tools._add_net_udev_rules = mock.MagicMock()
+
+            self.os_morphing_tools._setup_network_preservation(nics_info)
+
+            result_net_ifaces = dict(
+                self.os_morphing_tools._add_net_udev_rules.call_args[0][0])
+
+            mock_get_np.assert_called_once_with(self.os_morphing_tools)
+            self.os_morphing_tools._add_net_udev_rules.assert_called_once()
+            self.assertEqual(result_net_ifaces, expected_net_ifaces)

coriolis/tests/osmorphing/test_debian.py

@@ -1,10 +1,7 @@
 # Copyright 2024 Cloudbase Solutions Srl
 # All Rights Reserved.
-import copy
 from unittest import mock
 
-import yaml
-
 from coriolis import exception
 from coriolis.osmorphing import base
 from coriolis.osmorphing import debian
@@ -159,48 +156,6 @@ class BaseDebianMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
 
         self.assertEqual(result, expected_result)
 
-    @mock.patch.object(debian.BaseDebianMorphingTools, '_list_dir')
-    @mock.patch.object(debian.BaseDebianMorphingTools, '_read_file_sudo')
-    @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot')
-    @mock.patch.object(debian.BaseDebianMorphingTools, '_write_file_sudo')
-    def test__preserve_static_netplan_configuration(
-            self, mock_write_file_sudo, mock_exec_cmd_chroot,
-            mock_read_file_sudo, mock_list_dir):
-        ipv4 = "10.8.254.57"
-        ipv6 = "fe80::250:56ff:feba:f3aa"
-        mac = "00:50:56:ba:f3:aa"
-        eth0_config = {
-            "addresses": [{f"{ipv4}/24": {"limit": 0, "label": "maas"}},
-                          f"{ipv6}/64",
-                          {}]}
-        static_netplan_config = {
-            "network": {
-                "ethernets": {
-                    "eth0": eth0_config
-                }
-            }
-        }
-        nics_info = [
-            {"id": 1, "mac_address": "mac1", "ip_addresses": []},
-            {"id": 2, "mac_address": ""},
-            {"id": 3, "mac_address": mac, "ip_addresses": [ipv4, ipv6]}
-        ]
-        mock_list_dir.return_value = ["static.yml"]
-        mock_read_file_sudo.return_value = yaml.dump(
-            static_netplan_config, Dumper=yaml.SafeDumper).encode()
-        expected_eth0 = copy.deepcopy(eth0_config)
-        expected_eth0['match'] = {"macaddress": mac}
-        expected_eth0['set-name'] = 'eth0'
-        expected_netplan = {"network": {"ethernets": {"eth0": expected_eth0}}}
-
-        self.morpher._preserve_static_netplan_configuration(nics_info)
-
-        mock_write_file_sudo.assert_called_once_with(
-            "etc/netplan/static.yml",
-            yaml.dump(expected_netplan, Dumper=yaml.SafeDumper))
-        mock_exec_cmd_chroot.assert_called_once_with(
-            "cp etc/netplan/static.yml etc/netplan/static.yml.bak")
-
     @mock.patch.object(debian.BaseDebianMorphingTools, '_write_file_sudo')
     @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot')
     @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path')
@@ -251,9 +206,7 @@ class BaseDebianMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
         self.morpher.set_net_config(self.nics_info, dhcp)
 
         mock_disable_predictable_nic_names.assert_not_called()
-        mock_list_dir.assert_not_called()
         mock_write_file_sudo.assert_not_called()
-        mock_exec_cmd_chroot.assert_not_called()
 
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot')
     def test_get_installed_packages(self, mock_exec_cmd_chroot):

coriolis/tests/osmorphing/test_redhat.py

@@ -121,56 +121,6 @@ class BaseRedHatMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
             mock.call('/boot/grub2/grub.cfg')
         ])
 
-    @ddt.data(
-        (
-            [("/path/to/ifcfg-file", {"HWADDR": "mac", "NAME": "test-name"})],
-            [],
-            [("test-name", "mac")]
-        ),
-        (
-            [("/path/to/ifcfg-file", {"NAME": "test-name"})],
-            ["mac"],
-            [("test-name", "mac")]
-        ),
-        (
-            [("/path/to/ifcfg-file", {"HWADDR": "mac"})],
-            ["mac"],
-            [("file", "mac")]
-        ),
-        (
-            [("/path/to/ifcfg-file", {"NAME": "test-name"})],
-            [],
-            []
-        )
-    )
-    @ddt.unpack
-    def test__get_net_ifaces_info(self, ifcfgs_ethernet, mac_addresses,
-                                  expected):
-        result = self.morphing_tools._get_net_ifaces_info(
-            ifcfgs_ethernet, mac_addresses)
-        self.assertEqual(result, expected)
-
-    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
-    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo')
-    def test_add_net_udev_rules(self, mock_write_file_sudo, mock_test_path):
-        mock_test_path.return_value = False
-        net_ifaces_info = [
-            ("eth0", "AA:BB:CC:DD:EE:FF"),
-            ("eth1", "FF:EE:DD:CC:BB:AA")
-        ]
-        content = (
-            'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", '
-            'ATTR{address}=="aa:bb:cc:dd:ee:ff", NAME="eth0"\n'
-            'SUBSYSTEM=="net", ACTION=="add", DRIVERS=="?*", '
-            'ATTR{address}=="ff:ee:dd:cc:bb:aa", NAME="eth1"\n'
-        )
-
-        self.morphing_tools._add_net_udev_rules(net_ifaces_info)
-
-        mock_write_file_sudo.assert_called_once_with(
-            "etc/udev/rules.d/70-persistent-net.rules", content
-        )
-
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot')
     def test__has_systemd(self, mock_exec_cmd_chroot):
         result = self.morphing_tools._has_systemd()
@@ -215,21 +165,6 @@ class BaseRedHatMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
                       mock_read_config_file.return_value)
         ])
 
-    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_read_config_file')
-    @mock.patch.object(redhat.BaseRedHatMorphingTools, '_get_net_config_files')
-    def test__get_ifcfgs_by_type(self, mock_get_net_config_files,
-                                 mock_read_config_file):
-        mock_get_net_config_files.return_value = [mock.sentinel.ifcfg_file]
-        mock_read_config_file.side_effect = [{"TYPE": "Ethernet"}]
-
-        result = self.morphing_tools._get_ifcfgs_by_type("Ethernet")
-
-        mock_read_config_file.assert_called_once_with(mock.sentinel.ifcfg_file)
-        mock_get_net_config_files.assert_called_once()
-
-        self.assertEqual(
-            result, [(mock.sentinel.ifcfg_file, {"TYPE": "Ethernet"})])
-
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo')
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot')
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path')
@@ -341,12 +276,12 @@ class BaseRedHatMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
         redhat.BaseRedHatMorphingTools, 'disable_predictable_nic_names'
     )
     @mock.patch.object(redhat.BaseRedHatMorphingTools, '_write_nic_configs')
-    @mock.patch.object(redhat.BaseRedHatMorphingTools, '_get_ifcfgs_by_type')
-    @mock.patch.object(redhat.BaseRedHatMorphingTools, '_get_net_ifaces_info')
     @mock.patch.object(redhat.BaseRedHatMorphingTools, '_add_net_udev_rules')
+    @mock.patch.object(base.BaseLinuxOSMorphingTools,
+                       '_setup_network_preservation')
     def test_set_net_config_no_dhcp(
-            self, mock_add_net_udev_rules, mock_get_net_ifaces_info,
-            mock_get_ifcfgs_by_type, mock_write_nic_configs,
+            self, mock_setup_network_preservation, mock_add_net_udev_rules,
+            mock_write_nic_configs,
             mock_disable_predictable_nic_names):
         nics_info = [{
             'mac_address': mock.sentinel.mac_address,
@@ -357,11 +292,7 @@ class BaseRedHatMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
 
         mock_disable_predictable_nic_names.assert_not_called()
         mock_write_nic_configs.assert_not_called()
-        mock_get_ifcfgs_by_type.assert_called_once_with("Ethernet")
-        mock_get_net_ifaces_info.assert_called_once_with(
-            mock_get_ifcfgs_by_type.return_value, [mock.sentinel.mac_address])
-        mock_add_net_udev_rules.assert_called_once_with(
-            mock_get_net_ifaces_info.return_value)
+        mock_setup_network_preservation.assert_called_once()
 
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot')
     def test_get_installed_packages(self, mock_exec_cmd_chroot):
@@ -563,18 +494,3 @@ class BaseRedHatMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
         result = self.morphing_tools._get_config_file_content(config)
 
         self.assertEqual(result, 'key1="value1"\nkey2="value2"\n')
-
-    @mock.patch.object(base.BaseLinuxOSMorphingTools, '_list_dir')
-    def test__get_net_config_files(self, mock_list_dir):
-        mock_list_dir.return_value = ['ifcfg-eth0', 'ifcfg-lo', 'other-file']
-
-        result = self.morphing_tools._get_net_config_files()
-
-        expected_result = [
-            'etc/sysconfig/network-scripts/ifcfg-eth0',
-            'etc/sysconfig/network-scripts/ifcfg-lo'
-        ]
-
-        mock_list_dir.assert_called_once_with('etc/sysconfig/network-scripts')
-
-        self.assertEqual(result, expected_result)