Add support for SUSE machines

Add `SUSEOSMountTools` to detect `SUSE` machines as
osmorphing workers. Also, handle `sshd_config` relocation
to `/usr/etc/ssh/` on newer `SUSE` versions by copying
the vendor config to `/etc/ssh/` before modifying it.
Skip `zypper install` when `lvm2` is already present,
as some minimal images ship without repositories
configured.

Signed-off-by: Mihaela Balutoiu <mbalutoiu@cloudbasesolutions.com>

Author: mihaelabalutoiu

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,45 @@
+# 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()
+        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 psmisc")
+        self._exec_cmd("sudo modprobe dm-mod")
+        self._exec_cmd("sudo rm -f /etc/lvm/devices/system.devices")

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,104 @@
+# 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(self, mock_setup, mock_exec_cmd, mock_retry_on_error):
+        mock_retry_on_error.return_value = lambda f: f
+        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(
+                "sudo -E zypper --non-interactive install lvm2 psmisc"),
+            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")