Ubuntu OSMorphing Refactor

coriolis/osmorphing/base.py

@@ -25,6 +25,8 @@
     "os_type", "distribution_name", "release_version",
     "friendly_release_name"]
 DEFAULT_CLOUD_USER = "cloud-user"
+CLOUD_INIT_SERVICE_UNIT_NAME = "cloud-init"
+CLOUD_INIT_SERVICE_UNIT_NAME_FALLBACK = "cloud-init-main"
 
 
 class BaseOSMorphingTools(object, with_metaclass(abc.ABCMeta)):
@@ -48,7 +50,8 @@ def __init__(
         self._osmorphing_parameters = osmorphing_parameters
         self._osmorphing_operation_timeout = operation_timeout
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def get_required_detected_os_info_fields(cls):
         raise NotImplementedError("Required OS params not defined.")
 
@@ -74,7 +77,8 @@ def check_detected_os_info_parameters(cls, detected_os_info):
                     extra_os_info_fields, cls.__name__, detected_os_info))
         return True
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def check_os_supported(cls, detected_os_info):
         raise NotImplementedError(
             "OS compatibility check not implemented for tools class %s" % (
@@ -472,7 +476,16 @@ def _configure_cloud_init(self):
         self._write_cloud_init_mods_config(cloud_cfg_mods)
 
         if self._has_systemd_chroot():
-            self._enable_systemd_service("cloud-init")
+            try:
+                self._enable_systemd_service(CLOUD_INIT_SERVICE_UNIT_NAME)
+            except exception.CoriolisException:
+                LOG.warning(
+                    "Failed to enable service unit with name "
+                    f"{CLOUD_INIT_SERVICE_UNIT_NAME}. Trying new name.")
+                # NOTE(dvincze): New versions of cloud-init (>26) got its
+                # unit service name renamed.
+                self._enable_systemd_service(
+                    CLOUD_INIT_SERVICE_UNIT_NAME_FALLBACK)
 
     def _test_path_chroot(self, path):
         # This method uses _exec_cmd_chroot() instead of SFTP stat()

coriolis/osmorphing/osdetect/base.py

@@ -25,7 +25,8 @@ def __init__(self, conn, os_root_dir, operation_timeout):
         self._environment = {}
         self._osdetect_operation_timeout = operation_timeout
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def returned_detected_os_info_fields(cls):
         raise NotImplementedError(
             "No returned OS info fields by class '%s'" % cls.__name__)

coriolis/osmorphing/ubuntu.py

@@ -24,15 +24,16 @@ def check_os_supported(cls, detected_os_info):
                 UBUNTU_DISTRO_IDENTIFIER):
             return False
 
-        lts_releases = [12.04, 14.04, 16.04, 18.04, 20.04]
-        for lts_release in lts_releases:
-            if cls._version_supported_util(
-                    detected_os_info['release_version'],
-                    minimum=lts_release, maximum=lts_release):
-                return True
+        _, subversion = detected_os_info['release_version'].split('.', 1)
+        if not subversion.startswith("04"):
+            LOG.warning(
+                "Detected Ubuntu release version "
+                f"'{detected_os_info['release_version']}' is not an LTS one. "
+                "Coriolis only supports morphing Ubuntu LTS versions.")
+            return False
 
         return cls._version_supported_util(
-            detected_os_info['release_version'], minimum=22.04)
+            detected_os_info['release_version'], minimum=12.04)
 
     def _set_netplan_ethernet_configs(
             self, nics_info, dhcp=False, iface_name_prefix=None):

coriolis/tasks/base.py

@@ -48,15 +48,17 @@ def get_shared_libs_for_providers(
 
         return required_libs
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def get_required_task_info_properties(cls):
         """ Returns a list of the string fields which are required
         to be present during the tasks' run method. """
         raise NotImplementedError(
             "No required task info properties specified for task class of "
             "type '%s'." % cls)
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def get_returned_task_info_properties(cls):
         """ Returns a list of the string fields which are returned by the
         tasks' run method to be added to the task info.
@@ -65,7 +67,8 @@ def get_returned_task_info_properties(cls):
             "No returned task info properties specified for task class of "
             "type '%s'." % cls)
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def get_required_provider_types(cls):
         """ Returns a dict with 'source/destination' as keys containing a list
         of all the provider types (constants.PROVIDER_TYPE_*) required for the
@@ -75,7 +78,8 @@ def get_required_provider_types(cls):
             "No required provider types specified for task class of "
             "type '%s'." % cls)
 
-    @abc.abstractclassmethod
+    @classmethod
+    @abc.abstractmethod
     def get_required_platform(cls):
         """ Returns whether the task operates on the source platform, the
         destination, or both. (constants.TASK_PLATFORM_*)

coriolis/tests/osmorphing/test_ubuntu.py

@@ -4,11 +4,14 @@
 import logging
 from unittest import mock
 
+import ddt
+
 from coriolis.osmorphing import base
 from coriolis.osmorphing import ubuntu
 from coriolis.tests import test_base
 
 
+@ddt.ddt
 class BaseUbuntuMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
     """Test suite for the BaseUbuntuMophingTools class."""
 
@@ -37,19 +40,20 @@ def test_check_os_supported_not_supported(self):
 
         self.assertFalse(result)
 
-    def test_check_os_supported_lts_release(self):
-        self.detected_os_info['release_version'] = '20.04'
-
-        result = ubuntu.BaseUbuntuMorphingTools.check_os_supported(
-            self.detected_os_info)
-
-        self.assertTrue(result)
+    @ddt.data(
+        ("20.04", True),
+        ("20.04.2 LTS", True),
+        ("20.10", False),
+        ("20.054", False),
+    )
+    @ddt.unpack
+    def test_check_os_supported_lts_release(self, release_version, expected):
+        self.detected_os_info['release_version'] = release_version
 
-    def test_check_os_supported_non_lts_release(self):
         result = ubuntu.BaseUbuntuMorphingTools.check_os_supported(
             self.detected_os_info)
 
-        self.assertTrue(result)
+        self.assertEqual(expected, result)
 
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd')
     @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo')