From 2223d26420ffe29801ee951613db085b03b8e4c5 Mon Sep 17 00:00:00 2001 From: Cristian Matiut Date: Wed, 13 Aug 2025 14:28:56 +0000 Subject: [PATCH 1/2] Refactor cloud-init config into core --- coriolis/osmorphing/base.py | 101 ++++++++++++++++++++++++---------- coriolis/osmorphing/debian.py | 13 +++++ coriolis/osmorphing/redhat.py | 11 ++++ coriolis/osmorphing/suse.py | 10 ++++ 4 files changed, 105 insertions(+), 30 deletions(-) diff --git a/coriolis/osmorphing/base.py b/coriolis/osmorphing/base.py index b1091c4e..33d53c60 100644 --- a/coriolis/osmorphing/base.py +++ b/coriolis/osmorphing/base.py @@ -24,6 +24,7 @@ REQUIRED_DETECTED_OS_FIELDS = [ "os_type", "distribution_name", "release_version", "friendly_release_name"] +DEFAULT_CLOUD_USER = "cloud-user" class BaseOSMorphingTools(object, with_metaclass(abc.ABCMeta)): @@ -393,39 +394,79 @@ def _set_selinux_autorelabel(self): except Exception as err: LOG.warning("Failed to set autorelabel: %r" % err) - def _configure_cloud_init_user_retention(self): - cloud_cfg_paths = ["/etc/cloud/cloud.cfg"] + def _write_cloud_init_mods_config(self, cloud_cfg): + cloud_config_path = "/etc/cloud/cloud.cfg.d/99_coriolis.cfg" cloud_cfgs_dir = "/etc/cloud/cloud.cfg.d" - if self._test_path(cloud_cfgs_dir): - for path in self._exec_cmd_chroot( - 'ls -1 %s' % cloud_cfgs_dir).decode().splitlines(): - if path.endswith('.cfg'): - cloud_cfg_paths.append("%s/%s" % (cloud_cfgs_dir, path)) - + if not self._test_path(cloud_cfgs_dir): + self._exec_cmd_chroot("mkdir -p /etc/cloud/cloud.cfg.d") self._event_manager.progress_update( - "Reconfiguring cloud-init to retain original user credentials") - for cloud_cfg_path in cloud_cfg_paths: - if not self._test_path(cloud_cfg_path): - LOG.warn("Could not find %s. Skipping reconfiguration." % ( - cloud_cfg_path)) - continue + "Customizing cloud-init configuration") + new_cloud_cfg = yaml.dump(cloud_cfg, Dumper=yaml.SafeDumper) + self._write_file_sudo(cloud_config_path, new_cloud_cfg) + + def _disable_installer_cloud_config(self): + installer_config_path = "/etc/cloud/cloud.cfg.d/99-installer.cfg" + if self._test_path(installer_config_path): + self._event_manager.progress_update( + "Disabling installer-generated cloud-config") + self._exec_cmd_chroot( + f"mv {installer_config_path} {installer_config_path}.bak") + + def _ensure_cloud_init_not_disabled(self): + disabler_file = "/etc/cloud/cloud-init.disabled" + system_conf_disabler = "/etc/systemd/system.conf" + grub_conf_disabler = "/etc/default/grub" + if self._test_path(disabler_file): + self._exec_cmd_chroot(f"rm {disabler_file}") + if self._test_path(system_conf_disabler): + self._exec_cmd_chroot( + "sed -i '/cloud-init=disabled/d' %s" % system_conf_disabler) + if self._test_path(grub_conf_disabler): + self._exec_cmd_chroot( + "sed -i '/cloud-init=disabled/d' %s" % grub_conf_disabler) + self._execute_update_grub() + + def _reset_cloud_init_run(self): + self._exec_cmd_chroot("cloud-init clean --logs") + + def _get_default_cloud_user(self): + cloud_cfg_path = 'etc/cloud/cloud.cfg' + if not self._test_path(cloud_cfg_path): + return DEFAULT_CLOUD_USER + cloud_cfg_content = self._read_file_sudo(cloud_cfg_path) + cloud_cfg = yaml.load(cloud_cfg_content, Loader=yaml.SafeLoader) + return cloud_cfg.get('system_info', {}).get('default_user', {}).get( + 'name', DEFAULT_CLOUD_USER) + + def _create_cloudinit_user(self): + cloud_user = self._get_default_cloud_user() + if not self._check_user_exists(cloud_user): + self._exec_cmd_chroot("useradd %s" % cloud_user) + + def _configure_cloud_init(self): + cloud_cfg_mods = {} + if "cloud-init" not in self.get_packages()[0]: + return + + self._disable_installer_cloud_config() + self._ensure_cloud_init_not_disabled() + self._reset_cloud_init_run() + + if self._osmorphing_parameters.get('retain_user_credentials', False): + cloud_cfg_user_retention = { + 'disable_root': False, + 'ssh_pwauth': True, + 'users': None + } + cloud_cfg_mods.update(cloud_cfg_user_retention) + else: + self._create_cloudinit_user() + + if not self._osmorphing_parameters.get('set_dhcp', True): + disabled_network_config = {"network": {"config": "disabled"}} + cloud_cfg_mods.update(disabled_network_config) - try: - self._exec_cmd_chroot( - "cp %s %s.bak" % (cloud_cfg_path, cloud_cfg_path)) - - cloud_cfg = yaml.load(self._read_file(cloud_cfg_path), - Loader=yaml.SafeLoader) - - cloud_cfg['disable_root'] = False - cloud_cfg['ssh_pwauth'] = True - cloud_cfg['users'] = None - new_cloud_cfg = yaml.dump(cloud_cfg) - self._write_file_sudo(cloud_cfg_path, new_cloud_cfg) - except Exception as err: - raise exception.CoriolisException( - "Failed to reconfigure cloud-init to retain user " - "credentials. Error was: %s" % str(err)) from err + self._write_cloud_init_mods_config(cloud_cfg_mods) def _test_path_chroot(self, path): # This method uses _exec_cmd_chroot() instead of SFTP stat() diff --git a/coriolis/osmorphing/debian.py b/coriolis/osmorphing/debian.py index 9c9b9bc6..0138075b 100644 --- a/coriolis/osmorphing/debian.py +++ b/coriolis/osmorphing/debian.py @@ -97,6 +97,9 @@ def _compose_netplan_cfg(self, nics_info): } return yaml.dump(cfg, default_flow_style=False) + def _has_systemd_chroot(self): + return self._test_path("/lib/systemd/system") + def set_net_config(self, nics_info, dhcp): if not dhcp: LOG.info("Setting static IP configuration") @@ -124,6 +127,11 @@ def set_net_config(self, nics_info, dhcp): cfg_name = "%s/coriolis_netplan.yaml" % self.netplan_base self._write_file_sudo(cfg_name, new_cfg) + def _configure_cloud_init(self): + super(BaseDebianMorphingTools, self)._configure_cloud_init() + if self._has_systemd_chroot(): + self._enable_systemd_service("cloud-init") + def get_installed_packages(self): cmd = "dpkg-query -f '${binary:Package}\\n' -W" try: @@ -155,6 +163,11 @@ def pre_packages_install(self, package_names): "OSMorphing minion machine on the target platform. Original " "error was: %s" % str(err)) from err + def post_packages_install(self, package_names): + self._configure_cloud_init() + super(BaseDebianMorphingTools, self).post_packages_install( + package_names) + def install_packages(self, package_names): try: # NOTE(aznashwan): in the rare event that a replica sync occurs diff --git a/coriolis/osmorphing/redhat.py b/coriolis/osmorphing/redhat.py index 22d7ebff..8fb28f27 100644 --- a/coriolis/osmorphing/redhat.py +++ b/coriolis/osmorphing/redhat.py @@ -265,6 +265,12 @@ def pre_packages_install(self, package_names): else: LOG.debug("Skipping package 'grubby' as it's already installed") + def post_packages_install(self, package_names): + self._configure_cloud_init() + self._run_dracut() + super(BaseRedHatMorphingTools, self).post_packages_install( + package_names) + def install_packages(self, package_names): enable_repos = self._get_repos_to_enable() self._yum_install(package_names, enable_repos) @@ -282,6 +288,11 @@ def _set_network_nozeroconf_config(self): network_cfg["NOZEROCONF"] = "yes" self._write_config_file(network_cfg_file, network_cfg) + def _configure_cloud_init(self): + super(BaseRedHatMorphingTools, self)._configure_cloud_init() + if self._has_systemd(): + self._enable_systemd_service("cloud-init") + def _write_config_file(self, chroot_path, config_data): content = self._get_config_file_content(config_data) self._write_file_sudo(chroot_path, content) diff --git a/coriolis/osmorphing/suse.py b/coriolis/osmorphing/suse.py index 5610a0b5..79538c5b 100644 --- a/coriolis/osmorphing/suse.py +++ b/coriolis/osmorphing/suse.py @@ -127,6 +127,16 @@ def _has_systemd(self): except Exception: return False + def _configure_cloud_init(self): + super(BaseSUSEMorphingTools, self)._configure_cloud_init() + if self._has_systemd(): + self._enable_systemd_service("cloud-init") + + def post_packages_install(self, package_names): + self._configure_cloud_init() + self._run_dracut() + super(BaseSUSEMorphingTools, self).post_packages_install(package_names) + def _enable_sles_module(self, module): available_modules = self._exec_cmd_chroot( "SUSEConnect --list-extensions").decode() From b0f1425c3be3b84bd10fd67fab67ae7543d080fa Mon Sep 17 00:00:00 2001 From: Cristian Matiut Date: Wed, 17 Sep 2025 12:53:47 +0000 Subject: [PATCH 2/2] Update osmorphing unit tests --- coriolis/osmorphing/base.py | 4 +- coriolis/tests/osmorphing/test_base.py | 277 ++++++++++++++++------- coriolis/tests/osmorphing/test_debian.py | 27 +++ coriolis/tests/osmorphing/test_redhat.py | 24 ++ coriolis/tests/osmorphing/test_suse.py | 23 ++ 5 files changed, 270 insertions(+), 85 deletions(-) diff --git a/coriolis/osmorphing/base.py b/coriolis/osmorphing/base.py index 33d53c60..6308153f 100644 --- a/coriolis/osmorphing/base.py +++ b/coriolis/osmorphing/base.py @@ -395,10 +395,10 @@ def _set_selinux_autorelabel(self): LOG.warning("Failed to set autorelabel: %r" % err) def _write_cloud_init_mods_config(self, cloud_cfg): - cloud_config_path = "/etc/cloud/cloud.cfg.d/99_coriolis.cfg" cloud_cfgs_dir = "/etc/cloud/cloud.cfg.d" + cloud_config_path = f"{cloud_cfgs_dir}/99_coriolis.cfg" if not self._test_path(cloud_cfgs_dir): - self._exec_cmd_chroot("mkdir -p /etc/cloud/cloud.cfg.d") + self._exec_cmd_chroot("mkdir -p %s" % cloud_cfgs_dir) self._event_manager.progress_update( "Customizing cloud-init configuration") new_cloud_cfg = yaml.dump(cloud_cfg, Dumper=yaml.SafeDumper) diff --git a/coriolis/tests/osmorphing/test_base.py b/coriolis/tests/osmorphing/test_base.py index 1619843f..b9a1fa7b 100644 --- a/coriolis/tests/osmorphing/test_base.py +++ b/coriolis/tests/osmorphing/test_base.py @@ -568,92 +568,203 @@ def test__set_selinux_autorelabel_with_exception(self, level=logging.WARNING): self.os_morphing_tools._set_selinux_autorelabel() - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_read_file') - def test__configure_cloud_init_user_retention( - self, mock_read_file, mock_test_path, mock_write_file_sudo, - mock_exec_cmd_chroot): - cloud_cfg_paths = ["/etc/cloud/cloud.cfg"] - cloud_cfgs_dir = "/etc/cloud/cloud.cfg.d" - - mock_test_path.return_value = True - mock_exec_cmd_chroot.return_value = b"10.cfg\n20.cfg" - mock_read_file.return_value = ( - b"disable_root: true\n" - b"ssh_pwauth: false\nusers: ['user1', 'user2']") - self.os_morphing_tools._configure_cloud_init_user_retention() - - mock_exec_cmd_chroot.assert_has_calls([ - mock.call('ls -1 %s' % cloud_cfgs_dir), - mock.call('cp %s %s.bak' % ( - cloud_cfg_paths[0], cloud_cfg_paths[0])), - mock.call('cp %s/10.cfg %s/10.cfg.bak' % ( - cloud_cfgs_dir, cloud_cfgs_dir)), - mock.call('cp %s/20.cfg %s/20.cfg.bak' % ( - cloud_cfgs_dir, cloud_cfgs_dir)), - ]) - mock_test_path.assert_has_calls([ - mock.call('/etc/cloud/cloud.cfg.d'), - mock.call('/etc/cloud/cloud.cfg'), - mock.call('/etc/cloud/cloud.cfg.d/10.cfg'), - mock.call('/etc/cloud/cloud.cfg.d/20.cfg'), - ]) - - mock_write_file_sudo.assert_has_calls([ - mock.call('/etc/cloud/cloud.cfg', - 'disable_root: false\nssh_pwauth: true\nusers: null\n'), - mock.call('/etc/cloud/cloud.cfg.d/10.cfg', - 'disable_root: false\nssh_pwauth: true\nusers: null\n'), - mock.call('/etc/cloud/cloud.cfg.d/20.cfg', - 'disable_root: false\nssh_pwauth: true\nusers: null\n'), - ]) + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_write_file_sudo") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_test_path") + def test__write_cloud_init_mods_config( + self, mock__test_path, mock__exec_cmd_chroot, + mock__write_file_sudo): + mock__test_path.return_value = True + cloud_cfg = { + "mock_key1": {"mock_key2": "mock_value1"}, + "mock_key3": "mock_value2"} + + self.os_morphing_tools._write_cloud_init_mods_config(cloud_cfg) + + mock__exec_cmd_chroot.assert_not_called() + mock__write_file_sudo.assert_called_once_with( + "/etc/cloud/cloud.cfg.d/99_coriolis.cfg", + 'mock_key1:\n mock_key2: mock_value1\nmock_key3: mock_value2\n') + + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_write_file_sudo") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_test_path") + def test__write_cloud_init_mods_config_no_directory( + self, mock__test_path, mock__exec_cmd_chroot, + mock__write_file_sudo): + mock__test_path.return_value = False + cloud_cfg = {} + + self.os_morphing_tools._write_cloud_init_mods_config(cloud_cfg) + + mock__exec_cmd_chroot.assert_called_once_with( + "mkdir -p /etc/cloud/cloud.cfg.d") + mock__write_file_sudo.assert_called_once_with( + "/etc/cloud/cloud.cfg.d/99_coriolis.cfg", + '{}\n') + + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_test_path") + def test__disable_installer_cloud_config( + self, mock__test_path, mock__exec_cmd_chroot): + mock__test_path.return_value = True + + self.os_morphing_tools._disable_installer_cloud_config() + + mock__exec_cmd_chroot.assert_called_once_with( + "mv /etc/cloud/cloud.cfg.d/99-installer.cfg " + "/etc/cloud/cloud.cfg.d/99-installer.cfg.bak") + + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_test_path") + def test__disable_installer_cloud_config_no_file( + self, mock__test_path, mock__exec_cmd_chroot): + mock__test_path.return_value = False + + self.os_morphing_tools._disable_installer_cloud_config() + + mock__exec_cmd_chroot.assert_not_called() - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path') - def test__configure_cloud_init_user_retention_path_not_found( - self, mock_test_path, mock_exec_cmd_chroot, mock_write_file_sudo): - mock_test_path.return_value = False - - with self.assertLogs('coriolis.osmorphing.base', level=logging.WARN): - self.os_morphing_tools._configure_cloud_init_user_retention() - - mock_test_path.assert_has_calls([ - mock.call('/etc/cloud/cloud.cfg.d'), - mock.call('/etc/cloud/cloud.cfg'), - ]) - mock_exec_cmd_chroot.assert_not_called() - mock_write_file_sudo.assert_not_called() - - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_write_file_sudo') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_read_file') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot') - @mock.patch.object(base.BaseLinuxOSMorphingTools, '_test_path') - def test__configure_cloud_init_user_retention_exception( - self, mock_test_path, mock_exec_cmd_chroot, mock_read_file, - mock_write_file_sudo): - mock_test_path.return_value = True - mock_exec_cmd_chroot.return_value = b"" - mock_read_file.return_value = ( - b"disable_root: true" - b"nssh_pwauth: false" - b"users: ['user1', 'user2']") - mock_write_file_sudo.side_effect = Exception() + @ddt.data( + ((False, False, False), [], False), + ((True, True, False), [ + "rm /etc/cloud/cloud-init.disabled", + "sed -i '/cloud-init=disabled/d' /etc/systemd/system.conf", + ], False), + ((False, False, True), [ + "sed -i '/cloud-init=disabled/d' /etc/default/grub", + ], True) + ) + @ddt.unpack + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_execute_update_grub") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_test_path") + def test__ensure_cloud_init_not_disabled( + self, test_path_results, expected_cmds, updates_grub, + mock__test_path, mock__exec_cmd_chroot, mock__execute_update_grub): + mock__test_path.side_effect = test_path_results + + self.os_morphing_tools._ensure_cloud_init_not_disabled() + + called_cmds = [ + call.args[0] for call in mock__exec_cmd_chroot.call_args_list] + self.assertEqual(called_cmds, expected_cmds) + if updates_grub: + mock__execute_update_grub.assert_called_once() + else: + mock__execute_update_grub.assset_not_called() + + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + def test__reset_cloud_init_run(self, mock__exec_cmd_chroot): + self.os_morphing_tools._reset_cloud_init_run() + + mock__exec_cmd_chroot.assert_called_once_with( + "cloud-init clean --logs") - self.assertRaises( - exception.CoriolisException, - self.os_morphing_tools._configure_cloud_init_user_retention) + @ddt.data( + (False, None, base.DEFAULT_CLOUD_USER), + (True, "system_info:\n default_user:\n name: mock_user\n", + "mock_user"), + (True, "{}", base.DEFAULT_CLOUD_USER), + ) + @ddt.unpack + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_read_file_sudo") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_test_path") + def test__get_default_cloud_user( + self, test_path_result, file_content, expected_user, + mock__test_path, mock__read_file_sudo): + mock__test_path.return_value = test_path_result + mock__read_file_sudo.return_value = file_content + + result = self.os_morphing_tools._get_default_cloud_user() + + self.assertEqual(result, expected_user) + + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_check_user_exists") + @mock.patch.object(base.BaseLinuxOSMorphingTools, + "_get_default_cloud_user") + def test__create_cloudinit_user( + self, mock__get_default_cloud_user, + mock__check_user_exists, mock__exec_cmd_chroot): + mock__get_default_cloud_user.return_value = "mock_user" + mock__check_user_exists.return_value = False + + self.os_morphing_tools._create_cloudinit_user() + + mock__exec_cmd_chroot.assert_called_once_with("useradd mock_user") + + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_exec_cmd_chroot") + @mock.patch.object(base.BaseLinuxOSMorphingTools, "_check_user_exists") + @mock.patch.object(base.BaseLinuxOSMorphingTools, + "_get_default_cloud_user") + def test__create_cloudinit_user_already_exists( + self, mock__get_default_cloud_user, + mock__check_user_exists, mock__exec_cmd_chroot): + mock__get_default_cloud_user.return_value = "mock_user" + mock__check_user_exists.return_value = True + + self.os_morphing_tools._create_cloudinit_user() + + mock__exec_cmd_chroot.assert_not_called() - mock_test_path.assert_has_calls([ - mock.call('/etc/cloud/cloud.cfg.d'), - mock.call('/etc/cloud/cloud.cfg'), - ]) - mock_exec_cmd_chroot.assert_has_calls([ - mock.call('ls -1 /etc/cloud/cloud.cfg.d'), - mock.call('cp /etc/cloud/cloud.cfg /etc/cloud/cloud.cfg.bak'), - ]) + @ddt.data( + ( + ["vim"], + {}, + False, + None + ), + ( + ["cloud-init"], + {"retain_user_credentials": True, "set_dhcp": False}, + False, + { + "disable_root": False, + "ssh_pwauth": True, + "users": None, + "network": {"config": "disabled"}, + } + ), + ( + ["cloud-init", "vim"], + {"retain_user_credentials": False, "set_dhcp": True}, + True, + {} + ), + ) + @ddt.unpack + @mock.patch.object(base.BaseLinuxOSMorphingTools, + '_write_cloud_init_mods_config') + @mock.patch.object(base.BaseLinuxOSMorphingTools, '_create_cloudinit_user') + @mock.patch.object(base.BaseLinuxOSMorphingTools, '_reset_cloud_init_run') + @mock.patch.object(base.BaseLinuxOSMorphingTools, + '_ensure_cloud_init_not_disabled') + @mock.patch.object(base.BaseLinuxOSMorphingTools, + '_disable_installer_cloud_config') + @mock.patch.object(base.BaseLinuxOSMorphingTools, 'get_packages') + def test__configure_cloud_init( + self, returned_packages, osmorphing_params, creates_cloudinit_user, + expected_result, mock_get_packages, + mock__disable_installer_cloud_config, + mock__ensure_cloud_init_not_disabled, mock__reset_cloud_init_run, + mock__create_cloudinit_user, mock__write_cloud_init_mods_config): + mock_get_packages.return_value = returned_packages + self.os_morphing_tools._osmorphing_parameters = osmorphing_params + + self.os_morphing_tools._configure_cloud_init() + + if expected_result is not None: + mock__ensure_cloud_init_not_disabled.assert_called_once() + mock__reset_cloud_init_run.assert_called_once() + mock__write_cloud_init_mods_config.assert_called_once_with( + expected_result) + if creates_cloudinit_user: + mock__create_cloudinit_user.assert_called_once() + else: + mock__create_cloudinit_user.assert_not_called() + else: + mock__disable_installer_cloud_config.assert_not_called() @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot') def test__test_path_chroot(self, mock_exec_cmd_chroot): diff --git a/coriolis/tests/osmorphing/test_debian.py b/coriolis/tests/osmorphing/test_debian.py index c7f6b942..26757f38 100644 --- a/coriolis/tests/osmorphing/test_debian.py +++ b/coriolis/tests/osmorphing/test_debian.py @@ -155,6 +155,12 @@ def test__compose_netplan_cfg(self): self.assertEqual(result, expected_result) + @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path') + def test__has_systemd_chroot(self, mock__test_path): + result = self.morpher._has_systemd_chroot() + + self.assertEqual(result, mock__test_path.return_value) + @mock.patch.object(debian.BaseDebianMorphingTools, '_write_file_sudo') @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot') @mock.patch.object(debian.BaseDebianMorphingTools, '_test_path') @@ -207,6 +213,17 @@ def test_set_net_config_no_dhcp( mock_disable_predictable_nic_names.assert_not_called() mock_write_file_sudo.assert_not_called() + @mock.patch.object(debian.BaseDebianMorphingTools, + '_enable_systemd_service') + @mock.patch.object(debian.BaseDebianMorphingTools, '_has_systemd_chroot') + def test__configure_cloud_init( + self, mock__has_systemd_chroot, mock__enable_systemd_service): + mock__has_systemd_chroot.return_value = True + + self.morpher._configure_cloud_init() + + mock__enable_systemd_service.assert_called_once_with("cloud-init") + @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot') def test_get_installed_packages(self, mock_exec_cmd_chroot): mock_exec_cmd_chroot.return_value = \ @@ -259,6 +276,16 @@ def test_pre_packages_install_with_exception(self, mock_exec_cmd_chroot, mock_pre_packages_install.assert_called_once_with(self.package_names) + @mock.patch.object(debian.BaseDebianMorphingTools, '_configure_cloud_init') + @mock.patch.object(base.BaseLinuxOSMorphingTools, 'post_packages_install') + def test_post_packages_install( + self, mock_post_packages_install, mock__configure_cloud_init): + + self.morpher.post_packages_install(self.package_names) + + mock__configure_cloud_init.assert_called_once() + mock_post_packages_install.assert_called_once_with(self.package_names) + @mock.patch.object(debian.BaseDebianMorphingTools, '_exec_cmd_chroot') def test_install_packages(self, mock_exec_cmd_chroot): self.morpher.install_packages(self.package_names) diff --git a/coriolis/tests/osmorphing/test_redhat.py b/coriolis/tests/osmorphing/test_redhat.py index c9adf734..1020654c 100644 --- a/coriolis/tests/osmorphing/test_redhat.py +++ b/coriolis/tests/osmorphing/test_redhat.py @@ -436,6 +436,19 @@ def test_pre_packages_install_has_grubby( mock_yum_clean_all.assert_called_once() mock_yum_install.assert_not_called() + @mock.patch.object(redhat.BaseRedHatMorphingTools, '_configure_cloud_init') + @mock.patch.object(redhat.BaseRedHatMorphingTools, '_run_dracut') + @mock.patch.object(base.BaseLinuxOSMorphingTools, 'post_packages_install') + def test_post_packages_install( + self, mock_post_packages_install, mock__run_dracut, + mock__configure_cloud_init): + + self.morphing_tools.post_packages_install(self.package_names) + + mock__configure_cloud_init.assert_called_once() + mock__run_dracut.assert_called_once() + mock_post_packages_install.assert_called_once_with(self.package_names) + @mock.patch.object(redhat.BaseRedHatMorphingTools, '_yum_install') @mock.patch.object(redhat.BaseRedHatMorphingTools, '_get_repos_to_enable') def test_install_packages(self, mock_get_repos_to_enable, @@ -470,6 +483,17 @@ def test__set_network_nozeroconf_config( mock_write_config_file.assert_called_once_with( "etc/sysconfig/network", mock_read_config_file.return_value) + @mock.patch.object(redhat.BaseRedHatMorphingTools, + '_enable_systemd_service') + @mock.patch.object(redhat.BaseRedHatMorphingTools, '_has_systemd') + def test__configure_cloud_init( + self, mock__has_systemd, mock__enable_systemd_service): + mock__has_systemd.return_value = True + + self.morphing_tools._configure_cloud_init() + + mock__enable_systemd_service.assert_called_once_with("cloud-init") + @mock.patch.object( redhat.BaseRedHatMorphingTools, '_get_config_file_content' ) diff --git a/coriolis/tests/osmorphing/test_suse.py b/coriolis/tests/osmorphing/test_suse.py index f1ca436b..bdaa067d 100644 --- a/coriolis/tests/osmorphing/test_suse.py +++ b/coriolis/tests/osmorphing/test_suse.py @@ -226,6 +226,29 @@ def test__has_systemd_with_exception(self, mock_exec_cmd_chroot): self.assertFalse(result) + @mock.patch.object(suse.BaseSUSEMorphingTools, '_enable_systemd_service') + @mock.patch.object(suse.BaseSUSEMorphingTools, '_has_systemd') + def test__configure_cloud_init( + self, mock__has_systemd, mock__enable_systemd_service): + mock__has_systemd.return_value = True + + self.morphing_tools._configure_cloud_init() + + mock__enable_systemd_service.assert_called_once_with("cloud-init") + + @mock.patch.object(suse.BaseSUSEMorphingTools, '_configure_cloud_init') + @mock.patch.object(suse.BaseSUSEMorphingTools, '_run_dracut') + @mock.patch.object(base.BaseLinuxOSMorphingTools, 'post_packages_install') + def test_post_packages_install( + self, mock_post_packages_install, mock__run_dracut, + mock__configure_cloud_init): + + self.morphing_tools.post_packages_install(self.package_names) + + mock__configure_cloud_init.assert_called_once() + mock__run_dracut.assert_called_once() + mock_post_packages_install.assert_called_once_with(self.package_names) + @mock.patch.object(base.BaseLinuxOSMorphingTools, '_exec_cmd_chroot') def test__enable_sles_module(self, mock_exec_cmd_chroot): mock_exec_cmd_chroot.return_value = b"module1\nmodule2\nmodule3"