Add support SUSE machines

coriolis/osmorphing/osmount/factory.py

@@ -8,6 +8,7 @@
 from coriolis import constants
 from coriolis import exception
 from coriolis.osmorphing.osmount import redhat
+from coriolis.osmorphing.osmount import suse
 from coriolis.osmorphing.osmount import ubuntu
 from coriolis.osmorphing.osmount import windows
 
@@ -17,7 +18,8 @@
 def get_os_mount_tools(os_type, connection_info, event_manager,
                        ignore_devices, operation_timeout):
     os_mount_tools = {constants.OS_TYPE_LINUX: [ubuntu.UbuntuOSMountTools,
-                                                redhat.RedHatOSMountTools],
+                                                redhat.RedHatOSMountTools,
+                                                suse.SUSEOSMountTools],
                       constants.OS_TYPE_WINDOWS: [windows.WindowsMountTools]}
 
     if os_type and os_type not in os_mount_tools:

coriolis/osmorphing/osmount/suse.py

@@ -0,0 +1,54 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from oslo_log import log as logging
+
+from coriolis import exception
+from coriolis.osmorphing.osmount import base
+from coriolis import utils
+
+LOG = logging.getLogger(__name__)
+
+SUSE_DISTRO_IDENTIFIERS = [
+    'sles', 'opensuse-leap', 'opensuse-tumbleweed', 'opensuse']
+
+SSHD_CONFIG_PATH = "/etc/ssh/sshd_config"
+USR_SSHD_CONFIG_PATH = "/usr/etc/ssh/sshd_config"
+
+
+class SUSEOSMountTools(base.BaseLinuxOSMountTools):
+    def check_os(self):
+        os_info = utils.get_linux_os_info(self._ssh)
+        if os_info and os_info[0] in SUSE_DISTRO_IDENTIFIERS:
+            return True
+
+    def _allow_ssh_env_vars(self):
+        if not utils.test_ssh_path(self._ssh, SSHD_CONFIG_PATH):
+            self._exec_cmd(
+                "sudo cp %s %s" % (USR_SSHD_CONFIG_PATH, SSHD_CONFIG_PATH))
+        self._exec_cmd(
+            'sudo sed -i -e "\\$aAcceptEnv *" %s' % SSHD_CONFIG_PATH)
+        try:
+            utils.restart_service(self._ssh, "sshd")
+        except exception.CoriolisException:
+            LOG.warning(
+                "Could not restart sshd service. The SSH connection "
+                "may have been reset during the restart.")
+        return True
+
+    def setup(self):
+        super(SUSEOSMountTools, self).setup()
+        if not self._check_pkg_installed("lvm2"):
+            retry_ssh_cmd = utils.retry_on_error(
+                max_attempts=10, sleep_seconds=30)(self._exec_cmd)
+            retry_ssh_cmd(
+                "sudo -E zypper --non-interactive install lvm2")
+        self._exec_cmd("sudo modprobe dm-mod")
+        self._exec_cmd("sudo rm -f /etc/lvm/devices/system.devices")
+
+    def _check_pkg_installed(self, pkg_name):
+        try:
+            self._exec_cmd("rpm -q %s" % pkg_name)
+            return True
+        except Exception:
+            return False

coriolis/tests/osmorphing/osmount/test_factory.py

@@ -25,11 +25,13 @@ def test_get_os_mount_tools_unsupported_os_type(self):
                        return_value=False)
     @mock.patch.object(factory.redhat.RedHatOSMountTools, 'check_os',
                        return_value=False)
+    @mock.patch.object(factory.suse.SUSEOSMountTools, 'check_os',
+                       return_value=False)
     @mock.patch.object(factory.windows.WindowsMountTools, 'check_os',
                        return_value=False)
     def test_get_os_mount_tools_no_os_found(
-            self, mock_windows_check, mock_redhat_check, mock_ubuntu_check,
-            mock_exec_cmd, mock_connect):
+            self, mock_windows_check, mock_suse_check, mock_redhat_check,
+            mock_ubuntu_check, mock_exec_cmd, mock_connect):
         mock_exec_cmd.return_value = ("Ubuntu", "")
         self.assertRaises(
             exception.CoriolisException, factory.get_os_mount_tools,
@@ -39,6 +41,7 @@ def test_get_os_mount_tools_no_os_found(
 
         mock_redhat_check.assert_called_once_with()
         mock_ubuntu_check.assert_called_once_with()
+        mock_suse_check.assert_called_once_with()
         mock_windows_check.assert_not_called()
 
     @mock.patch.object(base.BaseSSHOSMountTools, '_connect')
@@ -47,11 +50,13 @@ def test_get_os_mount_tools_no_os_found(
                        return_value=True)
     @mock.patch.object(factory.redhat.RedHatOSMountTools, 'check_os',
                        return_value=False)
+    @mock.patch.object(factory.suse.SUSEOSMountTools, 'check_os',
+                       return_value=False)
     @mock.patch.object(factory.windows.WindowsMountTools, 'check_os',
                        return_value=False)
     def test_get_os_mount_tools_os_found(
-            self, mock_windows_check, mock_redhat_check, mock_ubuntu_check,
-            mock_exec_cmd, mock_connect):
+            self, mock_windows_check, mock_suse_check, mock_redhat_check,
+            mock_ubuntu_check, mock_exec_cmd, mock_connect):
         mock_exec_cmd.return_value = ("Ubuntu", "")
         tools = factory.get_os_mount_tools(
             factory.constants.OS_TYPE_LINUX, mock_connect,
@@ -61,4 +66,5 @@ def test_get_os_mount_tools_os_found(
 
         mock_ubuntu_check.assert_called_once_with()
         mock_redhat_check.assert_not_called()
+        mock_suse_check.assert_not_called()
         mock_windows_check.assert_not_called()

coriolis/tests/osmorphing/osmount/test_suse.py

@@ -0,0 +1,125 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis.osmorphing.osmount import suse
+from coriolis.tests import test_base
+
+
+class BaseSUSEOSMountToolsTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the SUSEOSMountTools class."""
+
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, '_connect')
+    def setUp(self, mock_connect):
+        super(BaseSUSEOSMountToolsTestCase, self).setUp()
+        self.ssh = mock.MagicMock()
+
+        self.tools = suse.SUSEOSMountTools(
+            self.ssh, mock.sentinel.event_manager,
+            mock.sentinel.ignore_devices,
+            mock.sentinel.operation_timeout)
+
+        mock_connect.assert_called_once_with()
+
+        self.tools._ssh = self.ssh
+
+    @mock.patch.object(suse.utils, 'get_linux_os_info')
+    def test_check_os(self, mock_get_linux_os_info):
+        mock_get_linux_os_info.return_value = ['sles']
+
+        result = self.tools.check_os()
+        self.assertTrue(result)
+
+    @mock.patch.object(suse.utils, 'get_linux_os_info')
+    def test_check_os_opensuse_leap(self, mock_get_linux_os_info):
+        mock_get_linux_os_info.return_value = ['opensuse-leap']
+
+        result = self.tools.check_os()
+        self.assertTrue(result)
+
+    @mock.patch.object(suse.utils, 'get_linux_os_info')
+    def test_check_os_opensuse_tumbleweed(self, mock_get_linux_os_info):
+        mock_get_linux_os_info.return_value = ['opensuse-tumbleweed']
+
+        result = self.tools.check_os()
+        self.assertTrue(result)
+
+    @mock.patch.object(suse.utils, 'get_linux_os_info')
+    def test_check_os_not_suse(self, mock_get_linux_os_info):
+        mock_get_linux_os_info.return_value = ['ubuntu']
+
+        result = self.tools.check_os()
+        self.assertIsNone(result)
+
+    @mock.patch.object(suse.utils, 'retry_on_error')
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, '_exec_cmd')
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, 'setup')
+    def test_setup_lvm2_not_installed(
+            self, mock_setup, mock_exec_cmd, mock_retry_on_error):
+        mock_retry_on_error.return_value = lambda f: f
+        mock_exec_cmd.side_effect = [
+            Exception("not installed"),
+            None, None, None]
+        result = self.tools.setup()
+        self.assertIsNone(result)
+
+        mock_setup.assert_called_once_with()
+        mock_retry_on_error.assert_called_once_with(
+            max_attempts=10, sleep_seconds=30)
+        mock_exec_cmd.assert_has_calls([
+            mock.call("rpm -q lvm2"),
+            mock.call(
+                "sudo -E zypper --non-interactive install lvm2"),
+            mock.call("sudo modprobe dm-mod"),
+            mock.call("sudo rm -f /etc/lvm/devices/system.devices")
+        ])
+
+    @mock.patch.object(suse.utils, 'retry_on_error')
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, '_exec_cmd')
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, 'setup')
+    def test_setup_lvm2_already_installed(
+            self, mock_setup, mock_exec_cmd, mock_retry_on_error):
+        result = self.tools.setup()
+        self.assertIsNone(result)
+
+        mock_setup.assert_called_once_with()
+        mock_retry_on_error.assert_not_called()
+        mock_exec_cmd.assert_has_calls([
+            mock.call("rpm -q lvm2"),
+            mock.call("sudo modprobe dm-mod"),
+            mock.call("sudo rm -f /etc/lvm/devices/system.devices")
+        ])
+
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, '_exec_cmd')
+    @mock.patch.object(suse.utils, 'restart_service')
+    @mock.patch.object(suse.utils, 'test_ssh_path', return_value=True)
+    def test__allow_ssh_env_vars(
+            self, mock_test_ssh_path, mock_restart_service, mock_exec_cmd):
+        result = self.tools._allow_ssh_env_vars()
+        self.assertTrue(result)
+
+        mock_test_ssh_path.assert_called_once_with(
+            self.ssh, suse.SSHD_CONFIG_PATH)
+        mock_exec_cmd.assert_called_once_with(
+            'sudo sed -i -e "\\$aAcceptEnv *" %s' % suse.SSHD_CONFIG_PATH)
+        mock_restart_service.assert_called_once_with(self.ssh, "sshd")
+
+    @mock.patch.object(suse.base.BaseSSHOSMountTools, '_exec_cmd')
+    @mock.patch.object(suse.utils, 'restart_service')
+    @mock.patch.object(suse.utils, 'test_ssh_path', return_value=False)
+    def test__allow_ssh_env_vars_usr_etc(
+            self, mock_test_ssh_path, mock_restart_service, mock_exec_cmd):
+        result = self.tools._allow_ssh_env_vars()
+        self.assertTrue(result)
+
+        mock_test_ssh_path.assert_called_once_with(
+            self.ssh, suse.SSHD_CONFIG_PATH)
+        mock_exec_cmd.assert_has_calls([
+            mock.call(
+                "sudo cp %s %s" % (
+                    suse.USR_SSHD_CONFIG_PATH, suse.SSHD_CONFIG_PATH)),
+            mock.call(
+                'sudo sed -i -e "\\$aAcceptEnv *" %s' %
+                suse.SSHD_CONFIG_PATH)])
+        mock_restart_service.assert_called_once_with(self.ssh, "sshd")

coriolis/tests/test_utils.py

@@ -915,14 +915,16 @@ def test_read_ssh_ini_config_file_check_true_path_not_exists(
     def test_write_systemd(self, mock_uuid, mock_test_ssh,
                            mock_write_ssh_file, mock_exec_ssh_cmd):
         mock_uuid.return_value = 'uuid'
-        mock_test_ssh.return_value = False
+        mock_test_ssh.side_effect = [True, False]
         mock_write_ssh_file.return_value = None
 
         utils._write_systemd(self.mock_ssh, 'cmdline', 'svc_name')
 
         mock_uuid.assert_called_once_with()
-        mock_test_ssh.assert_called_once_with(
-            self.mock_ssh, '/lib/systemd/system/svc_name.service')
+        mock_test_ssh.assert_has_calls([
+            mock.call(self.mock_ssh, '/lib/systemd/system'),
+            mock.call(self.mock_ssh,
+                      '/lib/systemd/system/svc_name.service')])
         mock_write_ssh_file.assert_called_once_with(self.mock_ssh,
                                                     '/tmp/uuid.service',
                                                     mock.ANY)
@@ -936,14 +938,37 @@ def test_write_systemd(self, mock_uuid, mock_test_ssh,
             mock.call(self.mock_ssh, 'sudo systemctl start svc_name',
                       get_pty=True)])
 
+    @mock.patch('coriolis.utils.exec_ssh_cmd')
+    @mock.patch('coriolis.utils.write_ssh_file')
+    @mock.patch('coriolis.utils.test_ssh_path')
+    @mock.patch.object(uuid, 'uuid4')
+    def test_write_systemd_usr_lib(self, mock_uuid, mock_test_ssh,
+                                   mock_write_ssh_file, mock_exec_ssh_cmd):
+        mock_uuid.return_value = 'uuid'
+        mock_test_ssh.side_effect = [False, False]
+        mock_write_ssh_file.return_value = None
+
+        utils._write_systemd(self.mock_ssh, 'cmdline', 'svc_name')
+
+        mock_test_ssh.assert_has_calls([
+            mock.call(self.mock_ssh, '/lib/systemd/system'),
+            mock.call(self.mock_ssh,
+                      '/usr/lib/systemd/system/svc_name.service')])
+        mock_exec_ssh_cmd.assert_has_calls([
+            mock.call(self.mock_ssh, 'sudo mv /tmp/uuid.service '
+                      '/usr/lib/systemd/system/svc_name.service',
+                      get_pty=True)])
+
     @mock.patch('coriolis.utils.test_ssh_path')
     def test_write_systemd_service_exists(self, mock_test_ssh):
         mock_test_ssh.return_value = True
 
         utils._write_systemd(self.mock_ssh, 'cmdline', 'svc_name')
 
-        mock_test_ssh.assert_called_once_with(
-            self.mock_ssh, '/lib/systemd/system/svc_name.service')
+        mock_test_ssh.assert_has_calls([
+            mock.call(self.mock_ssh, '/lib/systemd/system'),
+            mock.call(self.mock_ssh,
+                      '/lib/systemd/system/svc_name.service')])
 
     @mock.patch('coriolis.utils.exec_ssh_cmd')
     @mock.patch('coriolis.utils.write_ssh_file')
@@ -954,7 +979,7 @@ def test_write_systemd_service_selinux_exception(self, mock_uuid,
                                                      mock_write_ssh_file,
                                                      mock_exec_ssh_cmd):
         mock_uuid.return_value = 'uuid'
-        mock_test_ssh.return_value = False
+        mock_test_ssh.side_effect = [True, False]
         mock_write_ssh_file.return_value = None
         mock_exec_ssh_cmd.side_effect = [
             None, exception.CoriolisException(), None, None]
@@ -967,8 +992,10 @@ def test_write_systemd_service_selinux_exception(self, mock_uuid,
                                        start=True)
 
         mock_uuid.assert_called_once_with()
-        mock_test_ssh.assert_called_once_with(
-            self.mock_ssh, '/lib/systemd/system/svc_name.service')
+        mock_test_ssh.assert_has_calls([
+            mock.call(self.mock_ssh, '/lib/systemd/system'),
+            mock.call(self.mock_ssh,
+                      '/lib/systemd/system/svc_name.service')])
         mock_write_ssh_file.assert_called_once_with(self.mock_ssh,
                                                     '/tmp/uuid.service',
                                                     mock.ANY)
@@ -982,14 +1009,16 @@ def test_test_write_systemd_with_run_as(self, mock_uuid, mock_test_ssh,
                                             mock_exec_ssh_cmd):
 
         mock_uuid.return_value = 'uuid'
-        mock_test_ssh.return_value = False
+        mock_test_ssh.side_effect = [True, False]
 
         utils._write_systemd(self.mock_ssh, 'cmdline', 'svc_name',
                              run_as='test_user')
 
         mock_uuid.assert_called_once_with()
-        mock_test_ssh.assert_called_once_with(
-            self.mock_ssh, '/lib/systemd/system/svc_name.service')
+        mock_test_ssh.assert_has_calls([
+            mock.call(self.mock_ssh, '/lib/systemd/system'),
+            mock.call(self.mock_ssh,
+                      '/lib/systemd/system/svc_name.service')])
         mock_write_ssh_file.assert_called_once_with(
             self.mock_ssh, '/tmp/uuid.service',
             utils.SYSTEMD_TEMPLATE % {
@@ -1081,7 +1110,7 @@ def test_create_service_systemd(self, mock_test_ssh, mock_write_systemd):
     @mock.patch('coriolis.utils._write_upstart')
     @mock.patch('coriolis.utils.test_ssh_path')
     def test_create_service_upstart(self, mock_test_ssh, mock_write_upstart):
-        mock_test_ssh.side_effect = [False, True]
+        mock_test_ssh.side_effect = [False, False, True]
 
         utils.create_service(self.mock_ssh, 'cmdline', 'svc_name',
                              run_as='user', start=True)
@@ -1119,7 +1148,7 @@ def test_restart_service_with_systemd(self, mock_test_ssh,
     @mock.patch('coriolis.utils.test_ssh_path')
     def test_restart_service_with_upstart(self, mock_test_ssh,
                                           mock_exec_ssh_cmd):
-        mock_test_ssh.side_effect = [False, True]
+        mock_test_ssh.side_effect = [False, False, True]
 
         utils.restart_service(self.mock_ssh, 'svc_name')
 
@@ -1153,7 +1182,7 @@ def test_start_service_with_systemd(self, mock_test_ssh,
     @mock.patch('coriolis.utils.test_ssh_path')
     def test_start_service_with_upstart(self, mock_test_ssh,
                                         mock_exec_ssh_cmd):
-        mock_test_ssh.side_effect = [False, True]
+        mock_test_ssh.side_effect = [False, False, True]
 
         utils.start_service(self.mock_ssh, 'svc_name')
 
@@ -1187,7 +1216,7 @@ def test_stop_service_with_systemd(self, mock_test_ssh,
     @mock.patch('coriolis.utils.test_ssh_path')
     def test_stop_service_with_upstart(self, mock_test_ssh,
                                        mock_exec_ssh_cmd):
-        mock_test_ssh.side_effect = [False, True]
+        mock_test_ssh.side_effect = [False, False, True]
 
         utils.stop_service(self.mock_ssh, 'svc_name')