summaryrefslogtreecommitdiff
path: root/tests/unittests/test_distros
diff options
context:
space:
mode:
authorBrett Holman <bholman.devel@gmail.com>2021-12-03 13:11:46 -0700
committerGitHub <noreply@github.com>2021-12-03 13:11:46 -0700
commit039c40f9b3d88ee8158604bb18ca4bf2fb5d5e51 (patch)
tree5f1b09486ccaf98ee8159de58d9a2a1ef0af5dc1 /tests/unittests/test_distros
parentffa6fc88249aa080aa31811a45569a45e567418a (diff)
downloadvyos-cloud-init-039c40f9b3d88ee8158604bb18ca4bf2fb5d5e51.tar.gz
vyos-cloud-init-039c40f9b3d88ee8158604bb18ca4bf2fb5d5e51.zip
Reorganize unit test locations under tests/unittests (#1126)
This attempts to standardize unit test file location under test/unittests/ such that any source file located at cloudinit/path/to/file.py may have a corresponding unit test file at test/unittests/path/to/test_file.py. Noteworthy Comments: ==================== Four different duplicate test files existed: test_{gpg,util,cc_mounts,cc_resolv_conf}.py Each of these duplicate file pairs has been merged together. This is a break in git history for these files. The test suite appears to have a dependency on test order. Changing test order causes some tests to fail. This should be rectified, but for now some tests have been modified in tests/unittests/config/test_set_passwords.py. A helper class name starts with "Test" which causes pytest to try executing it as a test case, which then throws warnings "due to Class having __init__()". Silence by changing the name of the class. # helpers.py is imported in many test files, import paths change cloudinit/tests/helpers.py -> tests/unittests/helpers.py # Move directories: cloudinit/distros/tests -> tests/unittests/distros cloudinit/cmd/devel/tests -> tests/unittests/cmd/devel cloudinit/cmd/tests -> tests/unittests/cmd/ cloudinit/sources/helpers/tests -> tests/unittests/sources/helpers cloudinit/sources/tests -> tests/unittests/sources cloudinit/net/tests -> tests/unittests/net cloudinit/config/tests -> tests/unittests/config cloudinit/analyze/tests/ -> tests/unittests/analyze/ # Standardize tests already in tests/unittests/ test_datasource -> sources test_distros -> distros test_vmware -> sources/vmware test_handler -> config # this contains cloudconfig module tests test_runs -> runs
Diffstat (limited to 'tests/unittests/test_distros')
-rw-r--r--tests/unittests/test_distros/__init__.py21
-rw-r--r--tests/unittests/test_distros/test_arch.py45
-rw-r--r--tests/unittests/test_distros/test_bsd_utils.py67
-rw-r--r--tests/unittests/test_distros/test_create_users.py236
-rw-r--r--tests/unittests/test_distros/test_debian.py174
-rw-r--r--tests/unittests/test_distros/test_dragonflybsd.py25
-rw-r--r--tests/unittests/test_distros/test_freebsd.py45
-rw-r--r--tests/unittests/test_distros/test_generic.py315
-rw-r--r--tests/unittests/test_distros/test_gentoo.py26
-rw-r--r--tests/unittests/test_distros/test_hostname.py42
-rw-r--r--tests/unittests/test_distros/test_hosts.py45
-rw-r--r--tests/unittests/test_distros/test_manage_service.py38
-rw-r--r--tests/unittests/test_distros/test_netbsd.py17
-rw-r--r--tests/unittests/test_distros/test_netconfig.py916
-rw-r--r--tests/unittests/test_distros/test_opensuse.py12
-rw-r--r--tests/unittests/test_distros/test_photon.py68
-rw-r--r--tests/unittests/test_distros/test_resolv.py65
-rw-r--r--tests/unittests/test_distros/test_sles.py12
-rw-r--r--tests/unittests/test_distros/test_sysconfig.py86
-rw-r--r--tests/unittests/test_distros/test_user_data_normalize.py372
20 files changed, 0 insertions, 2627 deletions
diff --git a/tests/unittests/test_distros/__init__.py b/tests/unittests/test_distros/__init__.py
deleted file mode 100644
index 5394aa56..00000000
--- a/tests/unittests/test_distros/__init__.py
+++ /dev/null
@@ -1,21 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-import copy
-
-from cloudinit import distros
-from cloudinit import helpers
-from cloudinit import settings
-
-
-def _get_distro(dtype, system_info=None):
- """Return a Distro class of distro 'dtype'.
-
- cfg is format of CFG_BUILTIN['system_info'].
-
- example: _get_distro("debian")
- """
- if system_info is None:
- system_info = copy.deepcopy(settings.CFG_BUILTIN['system_info'])
- system_info['distro'] = dtype
- paths = helpers.Paths(system_info['paths'])
- distro_cls = distros.fetch(dtype)
- return distro_cls(dtype, system_info, paths)
diff --git a/tests/unittests/test_distros/test_arch.py b/tests/unittests/test_distros/test_arch.py
deleted file mode 100644
index a95ba3b5..00000000
--- a/tests/unittests/test_distros/test_arch.py
+++ /dev/null
@@ -1,45 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.distros.arch import _render_network
-from cloudinit import util
-
-from cloudinit.tests.helpers import (CiTestCase, dir2dict)
-
-from . import _get_distro
-
-
-class TestArch(CiTestCase):
-
- def test_get_distro(self):
- distro = _get_distro("arch")
- hostname = "myhostname"
- hostfile = self.tmp_path("hostfile")
- distro._write_hostname(hostname, hostfile)
- self.assertEqual(hostname + "\n", util.load_file(hostfile))
-
-
-class TestRenderNetwork(CiTestCase):
- def test_basic_static(self):
- """Just the most basic static config.
-
- note 'lo' should not be rendered as an interface."""
- entries = {'eth0': {'auto': True,
- 'dns-nameservers': ['8.8.8.8'],
- 'bootproto': 'static',
- 'address': '10.0.0.2',
- 'gateway': '10.0.0.1',
- 'netmask': '255.255.255.0'},
- 'lo': {'auto': True}}
- target = self.tmp_dir()
- devs = _render_network(entries, target=target)
- files = dir2dict(target, prefix=target)
- self.assertEqual(['eth0'], devs)
- self.assertEqual(
- {'/etc/netctl/eth0': '\n'.join([
- "Address=10.0.0.2/255.255.255.0",
- "Connection=ethernet",
- "DNS=('8.8.8.8')",
- "Gateway=10.0.0.1",
- "IP=static",
- "Interface=eth0", ""]),
- '/etc/resolv.conf': 'nameserver 8.8.8.8\n'}, files)
diff --git a/tests/unittests/test_distros/test_bsd_utils.py b/tests/unittests/test_distros/test_bsd_utils.py
deleted file mode 100644
index 3a68f2a9..00000000
--- a/tests/unittests/test_distros/test_bsd_utils.py
+++ /dev/null
@@ -1,67 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import cloudinit.distros.bsd_utils as bsd_utils
-
-from cloudinit.tests.helpers import (CiTestCase, ExitStack, mock)
-
-RC_FILE = """
-if something; then
- do something here
-fi
-hostname={hostname}
-"""
-
-
-class TestBsdUtils(CiTestCase):
-
- def setUp(self):
- super().setUp()
- patches = ExitStack()
- self.addCleanup(patches.close)
-
- self.load_file = patches.enter_context(
- mock.patch.object(bsd_utils.util, 'load_file'))
-
- self.write_file = patches.enter_context(
- mock.patch.object(bsd_utils.util, 'write_file'))
-
- def test_get_rc_config_value(self):
- self.load_file.return_value = 'hostname=foo\n'
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), 'foo')
- self.load_file.assert_called_with('/etc/rc.conf')
-
- self.load_file.return_value = 'hostname=foo'
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), 'foo')
-
- self.load_file.return_value = 'hostname="foo"'
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), 'foo')
-
- self.load_file.return_value = "hostname='foo'"
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), 'foo')
-
- self.load_file.return_value = 'hostname=\'foo"'
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), "'foo\"")
-
- self.load_file.return_value = ''
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), None)
-
- self.load_file.return_value = RC_FILE.format(hostname='foo')
- self.assertEqual(bsd_utils.get_rc_config_value('hostname'), "foo")
-
- def test_set_rc_config_value_unchanged(self):
- # bsd_utils.set_rc_config_value('hostname', 'foo')
- # self.write_file.assert_called_with('/etc/rc.conf', 'hostname=foo\n')
-
- self.load_file.return_value = RC_FILE.format(hostname='foo')
- self.write_file.assert_not_called()
-
- def test_set_rc_config_value(self):
- bsd_utils.set_rc_config_value('hostname', 'foo')
- self.write_file.assert_called_with('/etc/rc.conf', 'hostname=foo\n')
-
- self.load_file.return_value = RC_FILE.format(hostname='foo')
- bsd_utils.set_rc_config_value('hostname', 'bar')
- self.write_file.assert_called_with(
- '/etc/rc.conf',
- RC_FILE.format(hostname='bar')
- )
diff --git a/tests/unittests/test_distros/test_create_users.py b/tests/unittests/test_distros/test_create_users.py
deleted file mode 100644
index 685f08ba..00000000
--- a/tests/unittests/test_distros/test_create_users.py
+++ /dev/null
@@ -1,236 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import re
-
-from cloudinit import distros
-from cloudinit import ssh_util
-from cloudinit.tests.helpers import (CiTestCase, mock)
-from tests.unittests.util import abstract_to_concrete
-
-
-@mock.patch("cloudinit.distros.util.system_is_snappy", return_value=False)
-@mock.patch("cloudinit.distros.subp.subp")
-class TestCreateUser(CiTestCase):
-
- with_logs = True
-
- def setUp(self):
- super(TestCreateUser, self).setUp()
- self.dist = abstract_to_concrete(distros.Distro)(
- name='test', cfg=None, paths=None
- )
-
- def _useradd2call(self, args):
- # return a mock call for the useradd command in args
- # with expected 'logstring'.
- args = ['useradd'] + args
- logcmd = [a for a in args]
- for i in range(len(args)):
- if args[i] in ('--password',):
- logcmd[i + 1] = 'REDACTED'
- return mock.call(args, logstring=logcmd)
-
- def test_basic(self, m_subp, m_is_snappy):
- user = 'foouser'
- self.dist.create_user(user)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-m']),
- mock.call(['passwd', '-l', user])])
-
- def test_no_home(self, m_subp, m_is_snappy):
- user = 'foouser'
- self.dist.create_user(user, no_create_home=True)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-M']),
- mock.call(['passwd', '-l', user])])
-
- def test_system_user(self, m_subp, m_is_snappy):
- # system user should have no home and get --system
- user = 'foouser'
- self.dist.create_user(user, system=True)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '--system', '-M']),
- mock.call(['passwd', '-l', user])])
-
- def test_explicit_no_home_false(self, m_subp, m_is_snappy):
- user = 'foouser'
- self.dist.create_user(user, no_create_home=False)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-m']),
- mock.call(['passwd', '-l', user])])
-
- def test_unlocked(self, m_subp, m_is_snappy):
- user = 'foouser'
- self.dist.create_user(user, lock_passwd=False)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-m'])])
-
- def test_set_password(self, m_subp, m_is_snappy):
- user = 'foouser'
- password = 'passfoo'
- self.dist.create_user(user, passwd=password)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '--password', password, '-m']),
- mock.call(['passwd', '-l', user])])
-
- @mock.patch("cloudinit.distros.util.is_group")
- def test_group_added(self, m_is_group, m_subp, m_is_snappy):
- m_is_group.return_value = False
- user = 'foouser'
- self.dist.create_user(user, groups=['group1'])
- expected = [
- mock.call(['groupadd', 'group1']),
- self._useradd2call([user, '--groups', 'group1', '-m']),
- mock.call(['passwd', '-l', user])]
- self.assertEqual(m_subp.call_args_list, expected)
-
- @mock.patch("cloudinit.distros.util.is_group")
- def test_only_new_group_added(self, m_is_group, m_subp, m_is_snappy):
- ex_groups = ['existing_group']
- groups = ['group1', ex_groups[0]]
- m_is_group.side_effect = lambda m: m in ex_groups
- user = 'foouser'
- self.dist.create_user(user, groups=groups)
- expected = [
- mock.call(['groupadd', 'group1']),
- self._useradd2call([user, '--groups', ','.join(groups), '-m']),
- mock.call(['passwd', '-l', user])]
- self.assertEqual(m_subp.call_args_list, expected)
-
- @mock.patch("cloudinit.distros.util.is_group")
- def test_create_groups_with_whitespace_string(
- self, m_is_group, m_subp, m_is_snappy):
- # groups supported as a comma delimeted string even with white space
- m_is_group.return_value = False
- user = 'foouser'
- self.dist.create_user(user, groups='group1, group2')
- expected = [
- mock.call(['groupadd', 'group1']),
- mock.call(['groupadd', 'group2']),
- self._useradd2call([user, '--groups', 'group1,group2', '-m']),
- mock.call(['passwd', '-l', user])]
- self.assertEqual(m_subp.call_args_list, expected)
-
- def test_explicit_sudo_false(self, m_subp, m_is_snappy):
- user = 'foouser'
- self.dist.create_user(user, sudo=False)
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-m']),
- mock.call(['passwd', '-l', user])])
-
- @mock.patch('cloudinit.ssh_util.setup_user_keys')
- def test_setup_ssh_authorized_keys_with_string(
- self, m_setup_user_keys, m_subp, m_is_snappy):
- """ssh_authorized_keys allows string and calls setup_user_keys."""
- user = 'foouser'
- self.dist.create_user(user, ssh_authorized_keys='mykey')
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-m']),
- mock.call(['passwd', '-l', user])])
- m_setup_user_keys.assert_called_once_with(set(['mykey']), user)
-
- @mock.patch('cloudinit.ssh_util.setup_user_keys')
- def test_setup_ssh_authorized_keys_with_list(
- self, m_setup_user_keys, m_subp, m_is_snappy):
- """ssh_authorized_keys allows lists and calls setup_user_keys."""
- user = 'foouser'
- self.dist.create_user(user, ssh_authorized_keys=['key1', 'key2'])
- self.assertEqual(
- m_subp.call_args_list,
- [self._useradd2call([user, '-m']),
- mock.call(['passwd', '-l', user])])
- m_setup_user_keys.assert_called_once_with(set(['key1', 'key2']), user)
-
- @mock.patch('cloudinit.ssh_util.setup_user_keys')
- def test_setup_ssh_authorized_keys_with_integer(
- self, m_setup_user_keys, m_subp, m_is_snappy):
- """ssh_authorized_keys warns on non-iterable/string type."""
- user = 'foouser'
- self.dist.create_user(user, ssh_authorized_keys=-1)
- m_setup_user_keys.assert_called_once_with(set([]), user)
- match = re.match(
- r'.*WARNING: Invalid type \'<(type|class) \'int\'>\' detected for'
- ' \'ssh_authorized_keys\'.*',
- self.logs.getvalue(),
- re.DOTALL)
- self.assertIsNotNone(
- match, 'Missing ssh_authorized_keys invalid type warning')
-
- @mock.patch('cloudinit.ssh_util.setup_user_keys')
- def test_create_user_with_ssh_redirect_user_no_cloud_keys(
- self, m_setup_user_keys, m_subp, m_is_snappy):
- """Log a warning when trying to redirect a user no cloud ssh keys."""
- user = 'foouser'
- self.dist.create_user(user, ssh_redirect_user='someuser')
- self.assertIn(
- 'WARNING: Unable to disable SSH logins for foouser given '
- 'ssh_redirect_user: someuser. No cloud public-keys present.\n',
- self.logs.getvalue())
- m_setup_user_keys.assert_not_called()
-
- @mock.patch('cloudinit.ssh_util.setup_user_keys')
- def test_create_user_with_ssh_redirect_user_with_cloud_keys(
- self, m_setup_user_keys, m_subp, m_is_snappy):
- """Disable ssh when ssh_redirect_user and cloud ssh keys are set."""
- user = 'foouser'
- self.dist.create_user(
- user, ssh_redirect_user='someuser', cloud_public_ssh_keys=['key1'])
- disable_prefix = ssh_util.DISABLE_USER_OPTS
- disable_prefix = disable_prefix.replace('$USER', 'someuser')
- disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
- m_setup_user_keys.assert_called_once_with(
- set(['key1']), 'foouser', options=disable_prefix)
-
- @mock.patch('cloudinit.ssh_util.setup_user_keys')
- def test_create_user_with_ssh_redirect_user_does_not_disable_auth_keys(
- self, m_setup_user_keys, m_subp, m_is_snappy):
- """Do not disable ssh_authorized_keys when ssh_redirect_user is set."""
- user = 'foouser'
- self.dist.create_user(
- user, ssh_authorized_keys='auth1', ssh_redirect_user='someuser',
- cloud_public_ssh_keys=['key1'])
- disable_prefix = ssh_util.DISABLE_USER_OPTS
- disable_prefix = disable_prefix.replace('$USER', 'someuser')
- disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
- self.assertEqual(
- m_setup_user_keys.call_args_list,
- [mock.call(set(['auth1']), user), # not disabled
- mock.call(set(['key1']), 'foouser', options=disable_prefix)])
-
- @mock.patch("cloudinit.distros.subp.which")
- def test_lock_with_usermod_if_no_passwd(self, m_which, m_subp,
- m_is_snappy):
- """Lock uses usermod --lock if no 'passwd' cmd available."""
- m_which.side_effect = lambda m: m in ('usermod',)
- self.dist.lock_passwd("bob")
- self.assertEqual(
- [mock.call(['usermod', '--lock', 'bob'])],
- m_subp.call_args_list)
-
- @mock.patch("cloudinit.distros.subp.which")
- def test_lock_with_passwd_if_available(self, m_which, m_subp,
- m_is_snappy):
- """Lock with only passwd will use passwd."""
- m_which.side_effect = lambda m: m in ('passwd',)
- self.dist.lock_passwd("bob")
- self.assertEqual(
- [mock.call(['passwd', '-l', 'bob'])],
- m_subp.call_args_list)
-
- @mock.patch("cloudinit.distros.subp.which")
- def test_lock_raises_runtime_if_no_commands(self, m_which, m_subp,
- m_is_snappy):
- """Lock with no commands available raises RuntimeError."""
- m_which.return_value = None
- with self.assertRaises(RuntimeError):
- self.dist.lock_passwd("bob")
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_debian.py b/tests/unittests/test_distros/test_debian.py
deleted file mode 100644
index a88c2686..00000000
--- a/tests/unittests/test_distros/test_debian.py
+++ /dev/null
@@ -1,174 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-from itertools import count, cycle
-from unittest import mock
-
-import pytest
-
-from cloudinit import distros, util
-from cloudinit.distros.debian import (
- APT_GET_COMMAND,
- APT_GET_WRAPPER,
-)
-from cloudinit.tests.helpers import FilesystemMockingTestCase
-from cloudinit import subp
-
-
-@mock.patch("cloudinit.distros.debian.subp.subp")
-class TestDebianApplyLocale(FilesystemMockingTestCase):
-
- def setUp(self):
- super(TestDebianApplyLocale, self).setUp()
- self.new_root = self.tmp_dir()
- self.patchOS(self.new_root)
- self.patchUtils(self.new_root)
- self.spath = self.tmp_path('etc/default/locale', self.new_root)
- cls = distros.fetch("debian")
- self.distro = cls("debian", {}, None)
-
- def test_no_rerun(self, m_subp):
- """If system has defined locale, no re-run is expected."""
- m_subp.return_value = (None, None)
- locale = 'en_US.UTF-8'
- util.write_file(self.spath, 'LANG=%s\n' % locale, omode="w")
- self.distro.apply_locale(locale, out_fn=self.spath)
- m_subp.assert_not_called()
-
- def test_no_regen_on_c_utf8(self, m_subp):
- """If locale is set to C.UTF8, do not attempt to call locale-gen"""
- m_subp.return_value = (None, None)
- locale = 'C.UTF-8'
- util.write_file(self.spath, 'LANG=%s\n' % 'en_US.UTF-8', omode="w")
- self.distro.apply_locale(locale, out_fn=self.spath)
- self.assertEqual(
- [['update-locale', '--locale-file=' + self.spath,
- 'LANG=%s' % locale]],
- [p[0][0] for p in m_subp.call_args_list])
-
- def test_rerun_if_different(self, m_subp):
- """If system has different locale, locale-gen should be called."""
- m_subp.return_value = (None, None)
- locale = 'en_US.UTF-8'
- util.write_file(self.spath, 'LANG=fr_FR.UTF-8', omode="w")
- self.distro.apply_locale(locale, out_fn=self.spath)
- self.assertEqual(
- [['locale-gen', locale],
- ['update-locale', '--locale-file=' + self.spath,
- 'LANG=%s' % locale]],
- [p[0][0] for p in m_subp.call_args_list])
-
- def test_rerun_if_no_file(self, m_subp):
- """If system has no locale file, locale-gen should be called."""
- m_subp.return_value = (None, None)
- locale = 'en_US.UTF-8'
- self.distro.apply_locale(locale, out_fn=self.spath)
- self.assertEqual(
- [['locale-gen', locale],
- ['update-locale', '--locale-file=' + self.spath,
- 'LANG=%s' % locale]],
- [p[0][0] for p in m_subp.call_args_list])
-
- def test_rerun_on_unset_system_locale(self, m_subp):
- """If system has unset locale, locale-gen should be called."""
- m_subp.return_value = (None, None)
- locale = 'en_US.UTF-8'
- util.write_file(self.spath, 'LANG=', omode="w")
- self.distro.apply_locale(locale, out_fn=self.spath)
- self.assertEqual(
- [['locale-gen', locale],
- ['update-locale', '--locale-file=' + self.spath,
- 'LANG=%s' % locale]],
- [p[0][0] for p in m_subp.call_args_list])
-
- def test_rerun_on_mismatched_keys(self, m_subp):
- """If key is LC_ALL and system has only LANG, rerun is expected."""
- m_subp.return_value = (None, None)
- locale = 'en_US.UTF-8'
- util.write_file(self.spath, 'LANG=', omode="w")
- self.distro.apply_locale(locale, out_fn=self.spath, keyname='LC_ALL')
- self.assertEqual(
- [['locale-gen', locale],
- ['update-locale', '--locale-file=' + self.spath,
- 'LC_ALL=%s' % locale]],
- [p[0][0] for p in m_subp.call_args_list])
-
- def test_falseish_locale_raises_valueerror(self, m_subp):
- """locale as None or "" is invalid and should raise ValueError."""
-
- with self.assertRaises(ValueError) as ctext_m:
- self.distro.apply_locale(None)
- m_subp.assert_not_called()
-
- self.assertEqual(
- 'Failed to provide locale value.', str(ctext_m.exception))
-
- with self.assertRaises(ValueError) as ctext_m:
- self.distro.apply_locale("")
- m_subp.assert_not_called()
- self.assertEqual(
- 'Failed to provide locale value.', str(ctext_m.exception))
-
-
-@mock.patch.dict('os.environ', {}, clear=True)
-@mock.patch("cloudinit.distros.debian.subp.which", return_value=True)
-@mock.patch("cloudinit.distros.debian.subp.subp")
-class TestPackageCommand:
- distro = distros.fetch("debian")("debian", {}, None)
-
- @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
- return_value=True)
- def test_simple_command(self, m_apt_avail, m_subp, m_which):
- self.distro.package_command('update')
- apt_args = [APT_GET_WRAPPER['command']]
- apt_args.extend(APT_GET_COMMAND)
- apt_args.append('update')
- expected_call = {
- 'args': apt_args,
- 'capture': False,
- 'env': {'DEBIAN_FRONTEND': 'noninteractive'},
- }
- assert m_subp.call_args == mock.call(**expected_call)
-
- @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
- side_effect=[False, False, True])
- @mock.patch("cloudinit.distros.debian.time.sleep")
- def test_wait_for_lock(self, m_sleep, m_apt_avail, m_subp, m_which):
- self.distro._wait_for_apt_command("stub", {"args": "stub2"})
- assert m_sleep.call_args_list == [mock.call(1), mock.call(1)]
- assert m_subp.call_args_list == [mock.call(args='stub2')]
-
- @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
- return_value=False)
- @mock.patch("cloudinit.distros.debian.time.sleep")
- @mock.patch("cloudinit.distros.debian.time.time", side_effect=count())
- def test_lock_wait_timeout(
- self, m_time, m_sleep, m_apt_avail, m_subp, m_which
- ):
- with pytest.raises(TimeoutError):
- self.distro._wait_for_apt_command("stub", "stub2", timeout=5)
- assert m_subp.call_args_list == []
-
- @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
- side_effect=cycle([True, False]))
- @mock.patch("cloudinit.distros.debian.time.sleep")
- def test_lock_exception_wait(self, m_sleep, m_apt_avail, m_subp, m_which):
- exception = subp.ProcessExecutionError(
- exit_code=100, stderr="Could not get apt lock"
- )
- m_subp.side_effect = [exception, exception, "return_thing"]
- ret = self.distro._wait_for_apt_command("stub", {"args": "stub2"})
- assert ret == "return_thing"
-
- @mock.patch("cloudinit.distros.debian.Distro._apt_lock_available",
- side_effect=cycle([True, False]))
- @mock.patch("cloudinit.distros.debian.time.sleep")
- @mock.patch("cloudinit.distros.debian.time.time", side_effect=count())
- def test_lock_exception_timeout(
- self, m_time, m_sleep, m_apt_avail, m_subp, m_which
- ):
- m_subp.side_effect = subp.ProcessExecutionError(
- exit_code=100, stderr="Could not get apt lock"
- )
- with pytest.raises(TimeoutError):
- self.distro._wait_for_apt_command(
- "stub", {"args": "stub2"}, timeout=5
- )
diff --git a/tests/unittests/test_distros/test_dragonflybsd.py b/tests/unittests/test_distros/test_dragonflybsd.py
deleted file mode 100644
index df2c00f4..00000000
--- a/tests/unittests/test_distros/test_dragonflybsd.py
+++ /dev/null
@@ -1,25 +0,0 @@
-#!/usr/bin/env python3
-
-
-import cloudinit.util
-from cloudinit.tests.helpers import mock
-
-
-def test_find_dragonflybsd_part():
- assert cloudinit.util.find_dragonflybsd_part("/dev/vbd0s3") == "vbd0s3"
-
-
-@mock.patch("cloudinit.util.is_DragonFlyBSD")
-@mock.patch("cloudinit.subp.subp")
-def test_parse_mount(mock_subp, m_is_DragonFlyBSD):
- mount_out = """
-vbd0s3 on / (hammer2, local)
-devfs on /dev (devfs, nosymfollow, local)
-/dev/vbd0s0a on /boot (ufs, local)
-procfs on /proc (procfs, local)
-tmpfs on /var/run/shm (tmpfs, local)
-"""
-
- mock_subp.return_value = (mount_out, "")
- m_is_DragonFlyBSD.return_value = True
- assert cloudinit.util.parse_mount("/") == ("vbd0s3", "hammer2", "/")
diff --git a/tests/unittests/test_distros/test_freebsd.py b/tests/unittests/test_distros/test_freebsd.py
deleted file mode 100644
index be565b04..00000000
--- a/tests/unittests/test_distros/test_freebsd.py
+++ /dev/null
@@ -1,45 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.util import (find_freebsd_part, get_path_dev_freebsd)
-from cloudinit.tests.helpers import (CiTestCase, mock)
-
-import os
-
-
-class TestDeviceLookUp(CiTestCase):
-
- @mock.patch('cloudinit.subp.subp')
- def test_find_freebsd_part_label(self, mock_subp):
- glabel_out = '''
-gptid/fa52d426-c337-11e6-8911-00155d4c5e47 N/A da0p1
- label/rootfs N/A da0p2
- label/swap N/A da0p3
-'''
- mock_subp.return_value = (glabel_out, "")
- res = find_freebsd_part("/dev/label/rootfs")
- self.assertEqual("da0p2", res)
-
- @mock.patch('cloudinit.subp.subp')
- def test_find_freebsd_part_gpt(self, mock_subp):
- glabel_out = '''
- gpt/bootfs N/A vtbd0p1
-gptid/3f4cbe26-75da-11e8-a8f2-002590ec6166 N/A vtbd0p1
- gpt/swapfs N/A vtbd0p2
- gpt/rootfs N/A vtbd0p3
- iso9660/cidata N/A vtbd2
-'''
- mock_subp.return_value = (glabel_out, "")
- res = find_freebsd_part("/dev/gpt/rootfs")
- self.assertEqual("vtbd0p3", res)
-
- def test_get_path_dev_freebsd_label(self):
- mnt_list = '''
-/dev/label/rootfs / ufs rw 1 1
-devfs /dev devfs rw,multilabel 0 0
-fdescfs /dev/fd fdescfs rw 0 0
-/dev/da1s1 /mnt/resource ufs rw 2 2
-'''
- with mock.patch.object(os.path, 'exists',
- return_value=True):
- res = get_path_dev_freebsd('/etc', mnt_list)
- self.assertIsNotNone(res)
diff --git a/tests/unittests/test_distros/test_generic.py b/tests/unittests/test_distros/test_generic.py
deleted file mode 100644
index 336150bc..00000000
--- a/tests/unittests/test_distros/test_generic.py
+++ /dev/null
@@ -1,315 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit import distros
-from cloudinit import util
-
-from cloudinit.tests import helpers
-
-import os
-import pytest
-import shutil
-import tempfile
-from unittest import mock
-
-unknown_arch_info = {
- 'arches': ['default'],
- 'failsafe': {'primary': 'http://fs-primary-default',
- 'security': 'http://fs-security-default'}
-}
-
-package_mirrors = [
- {'arches': ['i386', 'amd64'],
- 'failsafe': {'primary': 'http://fs-primary-intel',
- 'security': 'http://fs-security-intel'},
- 'search': {
- 'primary': ['http://%(ec2_region)s.ec2/',
- 'http://%(availability_zone)s.clouds/'],
- 'security': ['http://security-mirror1-intel',
- 'http://security-mirror2-intel']}},
- {'arches': ['armhf', 'armel'],
- 'failsafe': {'primary': 'http://fs-primary-arm',
- 'security': 'http://fs-security-arm'}},
- unknown_arch_info
-]
-
-gpmi = distros._get_package_mirror_info
-gapmi = distros._get_arch_package_mirror_info
-
-
-class TestGenericDistro(helpers.FilesystemMockingTestCase):
-
- def setUp(self):
- super(TestGenericDistro, self).setUp()
- # Make a temp directoy for tests to use.
- self.tmp = tempfile.mkdtemp()
- self.addCleanup(shutil.rmtree, self.tmp)
-
- def _write_load_sudoers(self, _user, rules):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- os.makedirs(os.path.join(self.tmp, "etc"))
- os.makedirs(os.path.join(self.tmp, "etc", 'sudoers.d'))
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- d.write_sudo_rules("harlowja", rules)
- contents = util.load_file(d.ci_sudoers_fn)
- return contents
-
- def _count_in(self, lines_look_for, text_content):
- found_amount = 0
- for e in lines_look_for:
- for line in text_content.splitlines():
- line = line.strip()
- if line == e:
- found_amount += 1
- return found_amount
-
- def test_sudoers_ensure_rules(self):
- rules = 'ALL=(ALL:ALL) ALL'
- contents = self._write_load_sudoers('harlowja', rules)
- expected = ['harlowja ALL=(ALL:ALL) ALL']
- self.assertEqual(len(expected), self._count_in(expected, contents))
- not_expected = [
- 'harlowja A',
- 'harlowja L',
- 'harlowja L',
- ]
- self.assertEqual(0, self._count_in(not_expected, contents))
-
- def test_sudoers_ensure_rules_list(self):
- rules = [
- 'ALL=(ALL:ALL) ALL',
- 'B-ALL=(ALL:ALL) ALL',
- 'C-ALL=(ALL:ALL) ALL',
- ]
- contents = self._write_load_sudoers('harlowja', rules)
- expected = [
- 'harlowja ALL=(ALL:ALL) ALL',
- 'harlowja B-ALL=(ALL:ALL) ALL',
- 'harlowja C-ALL=(ALL:ALL) ALL',
- ]
- self.assertEqual(len(expected), self._count_in(expected, contents))
- not_expected = [
- 'harlowja A',
- 'harlowja L',
- 'harlowja L',
- ]
- self.assertEqual(0, self._count_in(not_expected, contents))
-
- def test_sudoers_ensure_new(self):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- d.ensure_sudo_dir("/b")
- contents = util.load_file("/etc/sudoers")
- self.assertIn("includedir /b", contents)
- self.assertTrue(os.path.isdir("/b"))
-
- def test_sudoers_ensure_append(self):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- util.write_file("/etc/sudoers", "josh, josh\n")
- d.ensure_sudo_dir("/b")
- contents = util.load_file("/etc/sudoers")
- self.assertIn("includedir /b", contents)
- self.assertTrue(os.path.isdir("/b"))
- self.assertIn("josh", contents)
- self.assertEqual(2, contents.count("josh"))
-
- def test_sudoers_ensure_only_one_includedir(self):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- for char in ['#', '@']:
- util.write_file("/etc/sudoers", "{}includedir /b".format(char))
- d.ensure_sudo_dir("/b")
- contents = util.load_file("/etc/sudoers")
- self.assertIn("includedir /b", contents)
- self.assertTrue(os.path.isdir("/b"))
- self.assertEqual(1, contents.count("includedir /b"))
-
- def test_arch_package_mirror_info_unknown(self):
- """for an unknown arch, we should get back that with arch 'default'."""
- arch_mirrors = gapmi(package_mirrors, arch="unknown")
- self.assertEqual(unknown_arch_info, arch_mirrors)
-
- def test_arch_package_mirror_info_known(self):
- arch_mirrors = gapmi(package_mirrors, arch="amd64")
- self.assertEqual(package_mirrors[0], arch_mirrors)
-
- def test_systemd_in_use(self):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- os.makedirs('/run/systemd/system')
- self.assertTrue(d.uses_systemd())
-
- def test_systemd_not_in_use(self):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- self.assertFalse(d.uses_systemd())
-
- def test_systemd_symlink(self):
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- self.patchOS(self.tmp)
- self.patchUtils(self.tmp)
- os.makedirs('/run/systemd')
- os.symlink('/', '/run/systemd/system')
- self.assertFalse(d.uses_systemd())
-
- @mock.patch('cloudinit.distros.debian.read_system_locale')
- def test_get_locale_ubuntu(self, m_locale):
- """Test ubuntu distro returns locale set to C.UTF-8"""
- m_locale.return_value = 'C.UTF-8'
- cls = distros.fetch("ubuntu")
- d = cls("ubuntu", {}, None)
- locale = d.get_locale()
- self.assertEqual('C.UTF-8', locale)
-
- def test_get_locale_rhel(self):
- """Test rhel distro returns NotImplementedError exception"""
- cls = distros.fetch("rhel")
- d = cls("rhel", {}, None)
- with self.assertRaises(NotImplementedError):
- d.get_locale()
-
- def test_expire_passwd_uses_chpasswd(self):
- """Test ubuntu.expire_passwd uses the passwd command."""
- for d_name in ("ubuntu", "rhel"):
- cls = distros.fetch(d_name)
- d = cls(d_name, {}, None)
- with mock.patch("cloudinit.subp.subp") as m_subp:
- d.expire_passwd("myuser")
- m_subp.assert_called_once_with(["passwd", "--expire", "myuser"])
-
- def test_expire_passwd_freebsd_uses_pw_command(self):
- """Test FreeBSD.expire_passwd uses the pw command."""
- cls = distros.fetch("freebsd")
- d = cls("freebsd", {}, None)
- with mock.patch("cloudinit.subp.subp") as m_subp:
- d.expire_passwd("myuser")
- m_subp.assert_called_once_with(
- ["pw", "usermod", "myuser", "-p", "01-Jan-1970"])
-
-
-class TestGetPackageMirrors:
-
- def return_first(self, mlist):
- if not mlist:
- return None
- return mlist[0]
-
- def return_second(self, mlist):
- if not mlist:
- return None
-
- return mlist[1] if len(mlist) > 1 else None
-
- def return_none(self, _mlist):
- return None
-
- def return_last(self, mlist):
- if not mlist:
- return None
- return(mlist[-1])
-
- @pytest.mark.parametrize(
- "allow_ec2_mirror, platform_type, mirrors",
- [
- (True, "ec2", [
- {'primary': 'http://us-east-1.ec2/',
- 'security': 'http://security-mirror1-intel'},
- {'primary': 'http://us-east-1a.clouds/',
- 'security': 'http://security-mirror2-intel'}
- ]),
- (True, "other", [
- {'primary': 'http://us-east-1.ec2/',
- 'security': 'http://security-mirror1-intel'},
- {'primary': 'http://us-east-1a.clouds/',
- 'security': 'http://security-mirror2-intel'}
- ]),
- (False, "ec2", [
- {'primary': 'http://us-east-1.ec2/',
- 'security': 'http://security-mirror1-intel'},
- {'primary': 'http://us-east-1a.clouds/',
- 'security': 'http://security-mirror2-intel'}
- ]),
- (False, "other", [
- {'primary': 'http://us-east-1a.clouds/',
- 'security': 'http://security-mirror1-intel'},
- {'primary': 'http://fs-primary-intel',
- 'security': 'http://security-mirror2-intel'}
- ])
- ])
- def test_get_package_mirror_info_az_ec2(self,
- allow_ec2_mirror,
- platform_type,
- mirrors):
- flag_path = "cloudinit.distros." \
- "ALLOW_EC2_MIRRORS_ON_NON_AWS_INSTANCE_TYPES"
- with mock.patch(flag_path, allow_ec2_mirror):
- arch_mirrors = gapmi(package_mirrors, arch="amd64")
- data_source_mock = mock.Mock(
- availability_zone="us-east-1a",
- platform_type=platform_type)
-
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_first)
- assert(results == mirrors[0])
-
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_second)
- assert(results == mirrors[1])
-
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_none)
- assert(results == package_mirrors[0]['failsafe'])
-
- def test_get_package_mirror_info_az_non_ec2(self):
- arch_mirrors = gapmi(package_mirrors, arch="amd64")
- data_source_mock = mock.Mock(availability_zone="nova.cloudvendor")
-
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_first)
- assert(results == {
- 'primary': 'http://nova.cloudvendor.clouds/',
- 'security': 'http://security-mirror1-intel'}
- )
-
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_last)
- assert(results == {
- 'primary': 'http://nova.cloudvendor.clouds/',
- 'security': 'http://security-mirror2-intel'}
- )
-
- def test_get_package_mirror_info_none(self):
- arch_mirrors = gapmi(package_mirrors, arch="amd64")
- data_source_mock = mock.Mock(availability_zone=None)
-
- # because both search entries here replacement based on
- # availability-zone, the filter will be called with an empty list and
- # failsafe should be taken.
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_first)
- assert(results == {
- 'primary': 'http://fs-primary-intel',
- 'security': 'http://security-mirror1-intel'}
- )
-
- results = gpmi(arch_mirrors, data_source=data_source_mock,
- mirror_filter=self.return_last)
- assert(results == {
- 'primary': 'http://fs-primary-intel',
- 'security': 'http://security-mirror2-intel'}
- )
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_gentoo.py b/tests/unittests/test_distros/test_gentoo.py
deleted file mode 100644
index 37a4f51f..00000000
--- a/tests/unittests/test_distros/test_gentoo.py
+++ /dev/null
@@ -1,26 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit import util
-from cloudinit import atomic_helper
-from cloudinit.tests.helpers import CiTestCase
-from . import _get_distro
-
-
-class TestGentoo(CiTestCase):
-
- def test_write_hostname(self):
- distro = _get_distro("gentoo")
- hostname = "myhostname"
- hostfile = self.tmp_path("hostfile")
- distro._write_hostname(hostname, hostfile)
- self.assertEqual('hostname="myhostname"\n', util.load_file(hostfile))
-
- def test_write_existing_hostname_with_comments(self):
- distro = _get_distro("gentoo")
- hostname = "myhostname"
- contents = '#This is the hostname\nhostname="localhost"'
- hostfile = self.tmp_path("hostfile")
- atomic_helper.write_file(hostfile, contents, omode="w")
- distro._write_hostname(hostname, hostfile)
- self.assertEqual('#This is the hostname\nhostname="myhostname"\n',
- util.load_file(hostfile))
diff --git a/tests/unittests/test_distros/test_hostname.py b/tests/unittests/test_distros/test_hostname.py
deleted file mode 100644
index f6d4dbe5..00000000
--- a/tests/unittests/test_distros/test_hostname.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import unittest
-
-from cloudinit.distros.parsers import hostname
-
-
-BASE_HOSTNAME = '''
-# My super-duper-hostname
-
-blahblah
-
-'''
-BASE_HOSTNAME = BASE_HOSTNAME.strip()
-
-
-class TestHostnameHelper(unittest.TestCase):
- def test_parse_same(self):
- hn = hostname.HostnameConf(BASE_HOSTNAME)
- self.assertEqual(str(hn).strip(), BASE_HOSTNAME)
- self.assertEqual(hn.hostname, 'blahblah')
-
- def test_no_adjust_hostname(self):
- hn = hostname.HostnameConf(BASE_HOSTNAME)
- prev_name = hn.hostname
- hn.set_hostname("")
- self.assertEqual(hn.hostname, prev_name)
-
- def test_adjust_hostname(self):
- hn = hostname.HostnameConf(BASE_HOSTNAME)
- prev_name = hn.hostname
- self.assertEqual(prev_name, 'blahblah')
- hn.set_hostname("bbbbd")
- self.assertEqual(hn.hostname, 'bbbbd')
- expected_out = '''
-# My super-duper-hostname
-
-bbbbd
-'''
- self.assertEqual(str(hn).strip(), expected_out.strip())
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_hosts.py b/tests/unittests/test_distros/test_hosts.py
deleted file mode 100644
index 8aaa6e48..00000000
--- a/tests/unittests/test_distros/test_hosts.py
+++ /dev/null
@@ -1,45 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import unittest
-
-from cloudinit.distros.parsers import hosts
-
-
-BASE_ETC = '''
-# Example
-127.0.0.1 localhost
-192.168.1.10 foo.mydomain.org foo
-192.168.1.10 bar.mydomain.org bar
-146.82.138.7 master.debian.org master
-209.237.226.90 www.opensource.org
-'''
-BASE_ETC = BASE_ETC.strip()
-
-
-class TestHostsHelper(unittest.TestCase):
- def test_parse(self):
- eh = hosts.HostsConf(BASE_ETC)
- self.assertEqual(eh.get_entry('127.0.0.1'), [['localhost']])
- self.assertEqual(eh.get_entry('192.168.1.10'),
- [['foo.mydomain.org', 'foo'],
- ['bar.mydomain.org', 'bar']])
- eh = str(eh)
- self.assertTrue(eh.startswith('# Example'))
-
- def test_add(self):
- eh = hosts.HostsConf(BASE_ETC)
- eh.add_entry('127.0.0.0', 'blah')
- self.assertEqual(eh.get_entry('127.0.0.0'), [['blah']])
- eh.add_entry('127.0.0.3', 'blah', 'blah2', 'blah3')
- self.assertEqual(eh.get_entry('127.0.0.3'),
- [['blah', 'blah2', 'blah3']])
-
- def test_del(self):
- eh = hosts.HostsConf(BASE_ETC)
- eh.add_entry('127.0.0.0', 'blah')
- self.assertEqual(eh.get_entry('127.0.0.0'), [['blah']])
-
- eh.del_entries('127.0.0.0')
- self.assertEqual(eh.get_entry('127.0.0.0'), [])
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_manage_service.py b/tests/unittests/test_distros/test_manage_service.py
deleted file mode 100644
index 47e7cfb0..00000000
--- a/tests/unittests/test_distros/test_manage_service.py
+++ /dev/null
@@ -1,38 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.tests.helpers import (CiTestCase, mock)
-from tests.unittests.util import TestingDistro
-
-
-class TestManageService(CiTestCase):
-
- with_logs = True
-
- def setUp(self):
- super(TestManageService, self).setUp()
- self.dist = TestingDistro()
-
- @mock.patch.object(TestingDistro, 'uses_systemd', return_value=False)
- @mock.patch("cloudinit.distros.subp.subp")
- def test_manage_service_systemctl_initcmd(self, m_subp, m_sysd):
- self.dist.init_cmd = ['systemctl']
- self.dist.manage_service('start', 'myssh')
- m_subp.assert_called_with(['systemctl', 'start', 'myssh'],
- capture=True)
-
- @mock.patch.object(TestingDistro, 'uses_systemd', return_value=False)
- @mock.patch("cloudinit.distros.subp.subp")
- def test_manage_service_service_initcmd(self, m_subp, m_sysd):
- self.dist.init_cmd = ['service']
- self.dist.manage_service('start', 'myssh')
- m_subp.assert_called_with(['service', 'myssh', 'start'], capture=True)
-
- @mock.patch.object(TestingDistro, 'uses_systemd', return_value=True)
- @mock.patch("cloudinit.distros.subp.subp")
- def test_manage_service_systemctl(self, m_subp, m_sysd):
- self.dist.init_cmd = ['ignore']
- self.dist.manage_service('start', 'myssh')
- m_subp.assert_called_with(['systemctl', 'start', 'myssh'],
- capture=True)
-
-# vi: ts=4 sw=4 expandtab
diff --git a/tests/unittests/test_distros/test_netbsd.py b/tests/unittests/test_distros/test_netbsd.py
deleted file mode 100644
index 11a68d2a..00000000
--- a/tests/unittests/test_distros/test_netbsd.py
+++ /dev/null
@@ -1,17 +0,0 @@
-import cloudinit.distros.netbsd
-
-import pytest
-import unittest.mock as mock
-
-
-@pytest.mark.parametrize('with_pkgin', (True, False))
-@mock.patch("cloudinit.distros.netbsd.os")
-def test_init(m_os, with_pkgin):
- print(with_pkgin)
- m_os.path.exists.return_value = with_pkgin
- cfg = {}
-
- distro = cloudinit.distros.netbsd.NetBSD("netbsd", cfg, None)
- expectation = ['pkgin', '-y', 'full-upgrade'] if with_pkgin else None
- assert distro.pkg_cmd_upgrade_prefix == expectation
- assert [mock.call('/usr/pkg/bin/pkgin')] == m_os.path.exists.call_args_list
diff --git a/tests/unittests/test_distros/test_netconfig.py b/tests/unittests/test_distros/test_netconfig.py
deleted file mode 100644
index e4eba179..00000000
--- a/tests/unittests/test_distros/test_netconfig.py
+++ /dev/null
@@ -1,916 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import copy
-import os
-import re
-from io import StringIO
-from textwrap import dedent
-from unittest import mock
-
-from cloudinit import distros
-from cloudinit.distros.parsers.sys_conf import SysConf
-from cloudinit import helpers
-from cloudinit import settings
-from cloudinit.tests.helpers import (
- FilesystemMockingTestCase, dir2dict)
-from cloudinit import subp
-from cloudinit import util
-from cloudinit import safeyaml
-
-BASE_NET_CFG = '''
-auto lo
-iface lo inet loopback
-
-auto eth0
-iface eth0 inet static
- address 192.168.1.5
- broadcast 192.168.1.0
- gateway 192.168.1.254
- netmask 255.255.255.0
- network 192.168.0.0
-
-auto eth1
-iface eth1 inet dhcp
-'''
-
-BASE_NET_CFG_FROM_V2 = '''
-auto lo
-iface lo inet loopback
-
-auto eth0
-iface eth0 inet static
- address 192.168.1.5/24
- gateway 192.168.1.254
-
-auto eth1
-iface eth1 inet dhcp
-'''
-
-BASE_NET_CFG_IPV6 = '''
-auto lo
-iface lo inet loopback
-
-auto eth0
-iface eth0 inet static
- address 192.168.1.5
- netmask 255.255.255.0
- network 192.168.0.0
- broadcast 192.168.1.0
- gateway 192.168.1.254
-
-iface eth0 inet6 static
- address 2607:f0d0:1002:0011::2
- netmask 64
- gateway 2607:f0d0:1002:0011::1
-
-iface eth1 inet static
- address 192.168.1.6
- netmask 255.255.255.0
- network 192.168.0.0
- broadcast 192.168.1.0
- gateway 192.168.1.254
-
-iface eth1 inet6 static
- address 2607:f0d0:1002:0011::3
- netmask 64
- gateway 2607:f0d0:1002:0011::1
-'''
-
-V1_NET_CFG = {'config': [{'name': 'eth0',
-
- 'subnets': [{'address': '192.168.1.5',
- 'broadcast': '192.168.1.0',
- 'gateway': '192.168.1.254',
- 'netmask': '255.255.255.0',
- 'type': 'static'}],
- 'type': 'physical'},
- {'name': 'eth1',
- 'subnets': [{'control': 'auto', 'type': 'dhcp4'}],
- 'type': 'physical'}],
- 'version': 1}
-
-V1_NET_CFG_WITH_DUPS = """\
-# same value in interface specific dns and global dns
-# should produce single entry in network file
-version: 1
-config:
- - type: physical
- name: eth0
- subnets:
- - type: static
- address: 192.168.0.102/24
- dns_nameservers: [1.2.3.4]
- dns_search: [test.com]
- interface: eth0
- - type: nameserver
- address: [1.2.3.4]
- search: [test.com]
-"""
-
-V1_NET_CFG_OUTPUT = """\
-# This file is generated from information provided by the datasource. Changes
-# to it will not persist across an instance reboot. To disable cloud-init's
-# network configuration capabilities, write a file
-# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
-# network: {config: disabled}
-auto lo
-iface lo inet loopback
-
-auto eth0
-iface eth0 inet static
- address 192.168.1.5/24
- broadcast 192.168.1.0
- gateway 192.168.1.254
-
-auto eth1
-iface eth1 inet dhcp
-"""
-
-V1_NET_CFG_IPV6_OUTPUT = """\
-# This file is generated from information provided by the datasource. Changes
-# to it will not persist across an instance reboot. To disable cloud-init's
-# network configuration capabilities, write a file
-# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
-# network: {config: disabled}
-auto lo
-iface lo inet loopback
-
-auto eth0
-iface eth0 inet6 static
- address 2607:f0d0:1002:0011::2/64
- gateway 2607:f0d0:1002:0011::1
-
-auto eth1
-iface eth1 inet dhcp
-"""
-
-V1_NET_CFG_IPV6 = {'config': [{'name': 'eth0',
- 'subnets': [{'address':
- '2607:f0d0:1002:0011::2',
- 'gateway':
- '2607:f0d0:1002:0011::1',
- 'netmask': '64',
- 'type': 'static6'}],
- 'type': 'physical'},
- {'name': 'eth1',
- 'subnets': [{'control': 'auto',
- 'type': 'dhcp4'}],
- 'type': 'physical'}],
- 'version': 1}
-
-
-V1_TO_V2_NET_CFG_OUTPUT = """\
-# This file is generated from information provided by the datasource. Changes
-# to it will not persist across an instance reboot. To disable cloud-init's
-# network configuration capabilities, write a file
-# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
-# network: {config: disabled}
-network:
- version: 2
- ethernets:
- eth0:
- addresses:
- - 192.168.1.5/24
- gateway4: 192.168.1.254
- eth1:
- dhcp4: true
-"""
-
-V1_TO_V2_NET_CFG_IPV6_OUTPUT = """\
-# This file is generated from information provided by the datasource. Changes
-# to it will not persist across an instance reboot. To disable cloud-init's
-# network configuration capabilities, write a file
-# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
-# network: {config: disabled}
-network:
- version: 2
- ethernets:
- eth0:
- addresses:
- - 2607:f0d0:1002:0011::2/64
- gateway6: 2607:f0d0:1002:0011::1
- eth1:
- dhcp4: true
-"""
-
-V2_NET_CFG = {
- 'ethernets': {
- 'eth7': {
- 'addresses': ['192.168.1.5/24'],
- 'gateway4': '192.168.1.254'},
- 'eth9': {
- 'dhcp4': True}
- },
- 'version': 2
-}
-
-
-V2_TO_V2_NET_CFG_OUTPUT = """\
-# This file is generated from information provided by the datasource. Changes
-# to it will not persist across an instance reboot. To disable cloud-init's
-# network configuration capabilities, write a file
-# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
-# network: {config: disabled}
-network:
- ethernets:
- eth7:
- addresses:
- - 192.168.1.5/24
- gateway4: 192.168.1.254
- eth9:
- dhcp4: true
- version: 2
-"""
-
-
-class WriteBuffer(object):
- def __init__(self):
- self.buffer = StringIO()
- self.mode = None
- self.omode = None
-
- def write(self, text):
- self.buffer.write(text)
-
- def __str__(self):
- return self.buffer.getvalue()
-
-
-class TestNetCfgDistroBase(FilesystemMockingTestCase):
-
- def setUp(self):
- super(TestNetCfgDistroBase, self).setUp()
- self.add_patch('cloudinit.util.system_is_snappy', 'm_snappy')
-
- def _get_distro(self, dname, renderers=None):
- cls = distros.fetch(dname)
- cfg = settings.CFG_BUILTIN
- cfg['system_info']['distro'] = dname
- if renderers:
- cfg['system_info']['network'] = {'renderers': renderers}
- paths = helpers.Paths({})
- return cls(dname, cfg.get('system_info'), paths)
-
- def assertCfgEquals(self, blob1, blob2):
- b1 = dict(SysConf(blob1.strip().splitlines()))
- b2 = dict(SysConf(blob2.strip().splitlines()))
- self.assertEqual(b1, b2)
- for (k, v) in b1.items():
- self.assertIn(k, b2)
- for (k, v) in b2.items():
- self.assertIn(k, b1)
- for (k, v) in b1.items():
- self.assertEqual(v, b2[k])
-
-
-class TestNetCfgDistroFreeBSD(TestNetCfgDistroBase):
-
- def setUp(self):
- super(TestNetCfgDistroFreeBSD, self).setUp()
- self.distro = self._get_distro('freebsd', renderers=['freebsd'])
-
- def _apply_and_verify_freebsd(self, apply_fn, config, expected_cfgs=None,
- bringup=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.freebsd.available') as m_avail:
- m_avail.return_value = True
- with self.reRooted(tmpd) as tmpd:
- util.ensure_dir('/etc')
- util.ensure_file('/etc/rc.conf')
- util.ensure_file('/etc/resolv.conf')
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- print("----------")
- print(expected)
- print("^^^^ expected | rendered VVVVVVV")
- print(results[cfgpath])
- print("----------")
- self.assertEqual(
- set(expected.split('\n')),
- set(results[cfgpath].split('\n')))
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- @mock.patch('cloudinit.net.get_interfaces_by_mac')
- def test_apply_network_config_freebsd_standard(self, ifaces_mac):
- ifaces_mac.return_value = {
- '00:15:5d:4c:73:00': 'eth0',
- }
- rc_conf_expected = """\
-defaultrouter=192.168.1.254
-ifconfig_eth0='192.168.1.5 netmask 255.255.255.0'
-ifconfig_eth1=DHCP
-"""
-
- expected_cfgs = {
- '/etc/rc.conf': rc_conf_expected,
- '/etc/resolv.conf': ''
- }
- self._apply_and_verify_freebsd(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy())
-
- @mock.patch('cloudinit.net.get_interfaces_by_mac')
- def test_apply_network_config_freebsd_ifrename(self, ifaces_mac):
- ifaces_mac.return_value = {
- '00:15:5d:4c:73:00': 'vtnet0',
- }
- rc_conf_expected = """\
-ifconfig_vtnet0_name=eth0
-defaultrouter=192.168.1.254
-ifconfig_eth0='192.168.1.5 netmask 255.255.255.0'
-ifconfig_eth1=DHCP
-"""
-
- V1_NET_CFG_RENAME = copy.deepcopy(V1_NET_CFG)
- V1_NET_CFG_RENAME['config'][0]['mac_address'] = '00:15:5d:4c:73:00'
-
- expected_cfgs = {
- '/etc/rc.conf': rc_conf_expected,
- '/etc/resolv.conf': ''
- }
- self._apply_and_verify_freebsd(self.distro.apply_network_config,
- V1_NET_CFG_RENAME,
- expected_cfgs=expected_cfgs.copy())
-
- @mock.patch('cloudinit.net.get_interfaces_by_mac')
- def test_apply_network_config_freebsd_nameserver(self, ifaces_mac):
- ifaces_mac.return_value = {
- '00:15:5d:4c:73:00': 'eth0',
- }
-
- V1_NET_CFG_DNS = copy.deepcopy(V1_NET_CFG)
- ns = ['1.2.3.4']
- V1_NET_CFG_DNS['config'][0]['subnets'][0]['dns_nameservers'] = ns
- expected_cfgs = {
- '/etc/resolv.conf': 'nameserver 1.2.3.4\n'
- }
- self._apply_and_verify_freebsd(self.distro.apply_network_config,
- V1_NET_CFG_DNS,
- expected_cfgs=expected_cfgs.copy())
-
-
-class TestNetCfgDistroUbuntuEni(TestNetCfgDistroBase):
-
- def setUp(self):
- super(TestNetCfgDistroUbuntuEni, self).setUp()
- self.distro = self._get_distro('ubuntu', renderers=['eni'])
-
- def eni_path(self):
- return '/etc/network/interfaces.d/50-cloud-init.cfg'
-
- def _apply_and_verify_eni(self, apply_fn, config, expected_cfgs=None,
- bringup=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.eni.available') as m_avail:
- m_avail.return_value = True
- with self.reRooted(tmpd) as tmpd:
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- print("----------")
- print(expected)
- print("^^^^ expected | rendered VVVVVVV")
- print(results[cfgpath])
- print("----------")
- self.assertEqual(expected, results[cfgpath])
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- def test_apply_network_config_eni_ub(self):
- expected_cfgs = {
- self.eni_path(): V1_NET_CFG_OUTPUT,
- }
- # ub_distro.apply_network_config(V1_NET_CFG, False)
- self._apply_and_verify_eni(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy())
-
- def test_apply_network_config_ipv6_ub(self):
- expected_cfgs = {
- self.eni_path(): V1_NET_CFG_IPV6_OUTPUT
- }
- self._apply_and_verify_eni(self.distro.apply_network_config,
- V1_NET_CFG_IPV6,
- expected_cfgs=expected_cfgs.copy())
-
-
-class TestNetCfgDistroUbuntuNetplan(TestNetCfgDistroBase):
- def setUp(self):
- super(TestNetCfgDistroUbuntuNetplan, self).setUp()
- self.distro = self._get_distro('ubuntu', renderers=['netplan'])
- self.devlist = ['eth0', 'lo']
-
- def _apply_and_verify_netplan(self, apply_fn, config, expected_cfgs=None,
- bringup=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.netplan.available',
- return_value=True):
- with mock.patch("cloudinit.net.netplan.get_devicelist",
- return_value=self.devlist):
- with self.reRooted(tmpd) as tmpd:
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- print("----------")
- print(expected)
- print("^^^^ expected | rendered VVVVVVV")
- print(results[cfgpath])
- print("----------")
- self.assertEqual(expected, results[cfgpath])
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- def netplan_path(self):
- return '/etc/netplan/50-cloud-init.yaml'
-
- def test_apply_network_config_v1_to_netplan_ub(self):
- expected_cfgs = {
- self.netplan_path(): V1_TO_V2_NET_CFG_OUTPUT,
- }
-
- # ub_distro.apply_network_config(V1_NET_CFG, False)
- self._apply_and_verify_netplan(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy())
-
- def test_apply_network_config_v1_ipv6_to_netplan_ub(self):
- expected_cfgs = {
- self.netplan_path(): V1_TO_V2_NET_CFG_IPV6_OUTPUT,
- }
-
- # ub_distro.apply_network_config(V1_NET_CFG_IPV6, False)
- self._apply_and_verify_netplan(self.distro.apply_network_config,
- V1_NET_CFG_IPV6,
- expected_cfgs=expected_cfgs.copy())
-
- def test_apply_network_config_v2_passthrough_ub(self):
- expected_cfgs = {
- self.netplan_path(): V2_TO_V2_NET_CFG_OUTPUT,
- }
- # ub_distro.apply_network_config(V2_NET_CFG, False)
- self._apply_and_verify_netplan(self.distro.apply_network_config,
- V2_NET_CFG,
- expected_cfgs=expected_cfgs.copy())
-
-
-class TestNetCfgDistroRedhat(TestNetCfgDistroBase):
-
- def setUp(self):
- super(TestNetCfgDistroRedhat, self).setUp()
- self.distro = self._get_distro('rhel', renderers=['sysconfig'])
-
- def ifcfg_path(self, ifname):
- return '/etc/sysconfig/network-scripts/ifcfg-%s' % ifname
-
- def control_path(self):
- return '/etc/sysconfig/network'
-
- def _apply_and_verify(self, apply_fn, config, expected_cfgs=None,
- bringup=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.sysconfig.available') as m_avail:
- m_avail.return_value = True
- with self.reRooted(tmpd) as tmpd:
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- self.assertCfgEquals(expected, results[cfgpath])
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- def test_apply_network_config_rh(self):
- expected_cfgs = {
- self.ifcfg_path('eth0'): dedent("""\
- BOOTPROTO=none
- DEFROUTE=yes
- DEVICE=eth0
- GATEWAY=192.168.1.254
- IPADDR=192.168.1.5
- NETMASK=255.255.255.0
- NM_CONTROLLED=no
- ONBOOT=yes
- TYPE=Ethernet
- USERCTL=no
- """),
- self.ifcfg_path('eth1'): dedent("""\
- BOOTPROTO=dhcp
- DEVICE=eth1
- NM_CONTROLLED=no
- ONBOOT=yes
- TYPE=Ethernet
- USERCTL=no
- """),
- self.control_path(): dedent("""\
- NETWORKING=yes
- """),
- }
- # rh_distro.apply_network_config(V1_NET_CFG, False)
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy())
-
- def test_apply_network_config_ipv6_rh(self):
- expected_cfgs = {
- self.ifcfg_path('eth0'): dedent("""\
- BOOTPROTO=none
- DEFROUTE=yes
- DEVICE=eth0
- IPV6ADDR=2607:f0d0:1002:0011::2/64
- IPV6INIT=yes
- IPV6_AUTOCONF=no
- IPV6_DEFAULTGW=2607:f0d0:1002:0011::1
- IPV6_FORCE_ACCEPT_RA=no
- NM_CONTROLLED=no
- ONBOOT=yes
- TYPE=Ethernet
- USERCTL=no
- """),
- self.ifcfg_path('eth1'): dedent("""\
- BOOTPROTO=dhcp
- DEVICE=eth1
- NM_CONTROLLED=no
- ONBOOT=yes
- TYPE=Ethernet
- USERCTL=no
- """),
- self.control_path(): dedent("""\
- NETWORKING=yes
- NETWORKING_IPV6=yes
- IPV6_AUTOCONF=no
- """),
- }
- # rh_distro.apply_network_config(V1_NET_CFG_IPV6, False)
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG_IPV6,
- expected_cfgs=expected_cfgs.copy())
-
- def test_vlan_render_unsupported(self):
- """Render officially unsupported vlan names."""
- cfg = {
- 'version': 2,
- 'ethernets': {
- 'eth0': {'addresses': ["192.10.1.2/24"],
- 'match': {'macaddress': "00:16:3e:60:7c:df"}}},
- 'vlans': {
- 'infra0': {'addresses': ["10.0.1.2/16"],
- 'id': 1001, 'link': 'eth0'}},
- }
- expected_cfgs = {
- self.ifcfg_path('eth0'): dedent("""\
- BOOTPROTO=none
- DEVICE=eth0
- HWADDR=00:16:3e:60:7c:df
- IPADDR=192.10.1.2
- NETMASK=255.255.255.0
- NM_CONTROLLED=no
- ONBOOT=yes
- TYPE=Ethernet
- USERCTL=no
- """),
- self.ifcfg_path('infra0'): dedent("""\
- BOOTPROTO=none
- DEVICE=infra0
- IPADDR=10.0.1.2
- NETMASK=255.255.0.0
- NM_CONTROLLED=no
- ONBOOT=yes
- PHYSDEV=eth0
- USERCTL=no
- VLAN=yes
- """),
- self.control_path(): dedent("""\
- NETWORKING=yes
- """),
- }
- self._apply_and_verify(
- self.distro.apply_network_config, cfg,
- expected_cfgs=expected_cfgs)
-
- def test_vlan_render(self):
- cfg = {
- 'version': 2,
- 'ethernets': {
- 'eth0': {'addresses': ["192.10.1.2/24"]}},
- 'vlans': {
- 'eth0.1001': {'addresses': ["10.0.1.2/16"],
- 'id': 1001, 'link': 'eth0'}},
- }
- expected_cfgs = {
- self.ifcfg_path('eth0'): dedent("""\
- BOOTPROTO=none
- DEVICE=eth0
- IPADDR=192.10.1.2
- NETMASK=255.255.255.0
- NM_CONTROLLED=no
- ONBOOT=yes
- TYPE=Ethernet
- USERCTL=no
- """),
- self.ifcfg_path('eth0.1001'): dedent("""\
- BOOTPROTO=none
- DEVICE=eth0.1001
- IPADDR=10.0.1.2
- NETMASK=255.255.0.0
- NM_CONTROLLED=no
- ONBOOT=yes
- PHYSDEV=eth0
- USERCTL=no
- VLAN=yes
- """),
- self.control_path(): dedent("""\
- NETWORKING=yes
- """),
- }
- self._apply_and_verify(
- self.distro.apply_network_config, cfg,
- expected_cfgs=expected_cfgs)
-
-
-class TestNetCfgDistroOpensuse(TestNetCfgDistroBase):
-
- def setUp(self):
- super(TestNetCfgDistroOpensuse, self).setUp()
- self.distro = self._get_distro('opensuse', renderers=['sysconfig'])
-
- def ifcfg_path(self, ifname):
- return '/etc/sysconfig/network/ifcfg-%s' % ifname
-
- def _apply_and_verify(self, apply_fn, config, expected_cfgs=None,
- bringup=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.sysconfig.available') as m_avail:
- m_avail.return_value = True
- with self.reRooted(tmpd) as tmpd:
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- self.assertCfgEquals(expected, results[cfgpath])
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- def test_apply_network_config_opensuse(self):
- """Opensuse uses apply_network_config and renders sysconfig"""
- expected_cfgs = {
- self.ifcfg_path('eth0'): dedent("""\
- BOOTPROTO=static
- IPADDR=192.168.1.5
- NETMASK=255.255.255.0
- STARTMODE=auto
- """),
- self.ifcfg_path('eth1'): dedent("""\
- BOOTPROTO=dhcp4
- STARTMODE=auto
- """),
- }
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy())
-
- def test_apply_network_config_ipv6_opensuse(self):
- """Opensuse uses apply_network_config and renders sysconfig w/ipv6"""
- expected_cfgs = {
- self.ifcfg_path('eth0'): dedent("""\
- BOOTPROTO=static
- IPADDR6=2607:f0d0:1002:0011::2/64
- STARTMODE=auto
- """),
- self.ifcfg_path('eth1'): dedent("""\
- BOOTPROTO=dhcp4
- STARTMODE=auto
- """),
- }
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG_IPV6,
- expected_cfgs=expected_cfgs.copy())
-
-
-class TestNetCfgDistroArch(TestNetCfgDistroBase):
- def setUp(self):
- super(TestNetCfgDistroArch, self).setUp()
- self.distro = self._get_distro('arch', renderers=['netplan'])
-
- def _apply_and_verify(self, apply_fn, config, expected_cfgs=None,
- bringup=False, with_netplan=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.netplan.available',
- return_value=with_netplan):
- with self.reRooted(tmpd) as tmpd:
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- print("----------")
- print(expected)
- print("^^^^ expected | rendered VVVVVVV")
- print(results[cfgpath])
- print("----------")
- self.assertEqual(expected, results[cfgpath])
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- def netctl_path(self, iface):
- return '/etc/netctl/%s' % iface
-
- def netplan_path(self):
- return '/etc/netplan/50-cloud-init.yaml'
-
- def test_apply_network_config_v1_without_netplan(self):
- # Note that this is in fact an invalid netctl config:
- # "Address=None/None"
- # But this is what the renderer has been writing out for a long time,
- # and the test's purpose is to assert that the netctl renderer is
- # still being used in absence of netplan, not the correctness of the
- # rendered netctl config.
- expected_cfgs = {
- self.netctl_path('eth0'): dedent("""\
- Address=192.168.1.5/255.255.255.0
- Connection=ethernet
- DNS=()
- Gateway=192.168.1.254
- IP=static
- Interface=eth0
- """),
- self.netctl_path('eth1'): dedent("""\
- Address=None/None
- Connection=ethernet
- DNS=()
- Gateway=
- IP=dhcp
- Interface=eth1
- """),
- }
-
- # ub_distro.apply_network_config(V1_NET_CFG, False)
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy(),
- with_netplan=False)
-
- def test_apply_network_config_v1_with_netplan(self):
- expected_cfgs = {
- self.netplan_path(): dedent("""\
- # generated by cloud-init
- network:
- version: 2
- ethernets:
- eth0:
- addresses:
- - 192.168.1.5/24
- gateway4: 192.168.1.254
- eth1:
- dhcp4: true
- """),
- }
-
- with mock.patch(
- 'cloudinit.net.netplan.get_devicelist',
- return_value=[]
- ):
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs=expected_cfgs.copy(),
- with_netplan=True)
-
-
-class TestNetCfgDistroPhoton(TestNetCfgDistroBase):
-
- def setUp(self):
- super(TestNetCfgDistroPhoton, self).setUp()
- self.distro = self._get_distro('photon', renderers=['networkd'])
-
- def create_conf_dict(self, contents):
- content_dict = {}
- for line in contents:
- if line:
- line = line.strip()
- if line and re.search(r'^\[(.+)\]$', line):
- content_dict[line] = []
- key = line
- elif line:
- assert key
- content_dict[key].append(line)
-
- return content_dict
-
- def compare_dicts(self, actual, expected):
- for k, v in actual.items():
- self.assertEqual(sorted(expected[k]), sorted(v))
-
- def _apply_and_verify(self, apply_fn, config, expected_cfgs=None,
- bringup=False):
- if not expected_cfgs:
- raise ValueError('expected_cfg must not be None')
-
- tmpd = None
- with mock.patch('cloudinit.net.networkd.available') as m_avail:
- m_avail.return_value = True
- with self.reRooted(tmpd) as tmpd:
- apply_fn(config, bringup)
-
- results = dir2dict(tmpd)
- for cfgpath, expected in expected_cfgs.items():
- actual = self.create_conf_dict(results[cfgpath].splitlines())
- self.compare_dicts(actual, expected)
- self.assertEqual(0o644, get_mode(cfgpath, tmpd))
-
- def nwk_file_path(self, ifname):
- return '/etc/systemd/network/10-cloud-init-%s.network' % ifname
-
- def net_cfg_1(self, ifname):
- ret = """\
- [Match]
- Name=%s
- [Network]
- DHCP=no
- [Address]
- Address=192.168.1.5/24
- [Route]
- Gateway=192.168.1.254""" % ifname
- return ret
-
- def net_cfg_2(self, ifname):
- ret = """\
- [Match]
- Name=%s
- [Network]
- DHCP=ipv4""" % ifname
- return ret
-
- def test_photon_network_config_v1(self):
- tmp = self.net_cfg_1('eth0').splitlines()
- expected_eth0 = self.create_conf_dict(tmp)
-
- tmp = self.net_cfg_2('eth1').splitlines()
- expected_eth1 = self.create_conf_dict(tmp)
-
- expected_cfgs = {
- self.nwk_file_path('eth0'): expected_eth0,
- self.nwk_file_path('eth1'): expected_eth1,
- }
-
- self._apply_and_verify(self.distro.apply_network_config,
- V1_NET_CFG,
- expected_cfgs.copy())
-
- def test_photon_network_config_v2(self):
- tmp = self.net_cfg_1('eth7').splitlines()
- expected_eth7 = self.create_conf_dict(tmp)
-
- tmp = self.net_cfg_2('eth9').splitlines()
- expected_eth9 = self.create_conf_dict(tmp)
-
- expected_cfgs = {
- self.nwk_file_path('eth7'): expected_eth7,
- self.nwk_file_path('eth9'): expected_eth9,
- }
-
- self._apply_and_verify(self.distro.apply_network_config,
- V2_NET_CFG,
- expected_cfgs.copy())
-
- def test_photon_network_config_v1_with_duplicates(self):
- expected = """\
- [Match]
- Name=eth0
- [Network]
- DHCP=no
- DNS=1.2.3.4
- Domains=test.com
- [Address]
- Address=192.168.0.102/24"""
-
- net_cfg = safeyaml.load(V1_NET_CFG_WITH_DUPS)
-
- expected = self.create_conf_dict(expected.splitlines())
- expected_cfgs = {
- self.nwk_file_path('eth0'): expected,
- }
-
- self._apply_and_verify(self.distro.apply_network_config,
- net_cfg,
- expected_cfgs.copy())
-
-
-def get_mode(path, target=None):
- return os.stat(subp.target_path(target, path)).st_mode & 0o777
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_opensuse.py b/tests/unittests/test_distros/test_opensuse.py
deleted file mode 100644
index b9bb9b3e..00000000
--- a/tests/unittests/test_distros/test_opensuse.py
+++ /dev/null
@@ -1,12 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.tests.helpers import CiTestCase
-
-from . import _get_distro
-
-
-class TestopenSUSE(CiTestCase):
-
- def test_get_distro(self):
- distro = _get_distro("opensuse")
- self.assertEqual(distro.osfamily, 'suse')
diff --git a/tests/unittests/test_distros/test_photon.py b/tests/unittests/test_distros/test_photon.py
deleted file mode 100644
index 1c3145ca..00000000
--- a/tests/unittests/test_distros/test_photon.py
+++ /dev/null
@@ -1,68 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from . import _get_distro
-from cloudinit import util
-from cloudinit.tests.helpers import mock
-from cloudinit.tests.helpers import CiTestCase
-
-SYSTEM_INFO = {
- 'paths': {
- 'cloud_dir': '/var/lib/cloud/',
- 'templates_dir': '/etc/cloud/templates/',
- },
- 'network': {'renderers': 'networkd'},
-}
-
-
-class TestPhoton(CiTestCase):
- with_logs = True
- distro = _get_distro('photon', SYSTEM_INFO)
- expected_log_line = 'Rely on PhotonOS default network config'
-
- def test_network_renderer(self):
- self.assertEqual(self.distro._cfg['network']['renderers'], 'networkd')
-
- def test_get_distro(self):
- self.assertEqual(self.distro.osfamily, 'photon')
-
- @mock.patch("cloudinit.distros.photon.subp.subp")
- def test_write_hostname(self, m_subp):
- hostname = 'myhostname'
- hostfile = self.tmp_path('previous-hostname')
- self.distro._write_hostname(hostname, hostfile)
- self.assertEqual(hostname, util.load_file(hostfile))
-
- ret = self.distro._read_hostname(hostfile)
- self.assertEqual(ret, hostname)
-
- m_subp.return_value = (None, None)
- hostfile += 'hostfile'
- self.distro._write_hostname(hostname, hostfile)
-
- m_subp.return_value = (hostname, None)
- ret = self.distro._read_hostname(hostfile)
- self.assertEqual(ret, hostname)
-
- self.logs.truncate(0)
- m_subp.return_value = (None, 'bla')
- self.distro._write_hostname(hostname, None)
- self.assertIn('Error while setting hostname', self.logs.getvalue())
-
- @mock.patch('cloudinit.net.generate_fallback_config')
- def test_fallback_netcfg(self, m_fallback_cfg):
-
- key = 'disable_fallback_netcfg'
- # Don't use fallback if no setting given
- self.logs.truncate(0)
- assert(self.distro.generate_fallback_config() is None)
- self.assertIn(self.expected_log_line, self.logs.getvalue())
-
- self.logs.truncate(0)
- self.distro._cfg[key] = True
- assert(self.distro.generate_fallback_config() is None)
- self.assertIn(self.expected_log_line, self.logs.getvalue())
-
- self.logs.truncate(0)
- self.distro._cfg[key] = False
- assert(self.distro.generate_fallback_config() is not None)
- self.assertNotIn(self.expected_log_line, self.logs.getvalue())
diff --git a/tests/unittests/test_distros/test_resolv.py b/tests/unittests/test_distros/test_resolv.py
deleted file mode 100644
index 7d940750..00000000
--- a/tests/unittests/test_distros/test_resolv.py
+++ /dev/null
@@ -1,65 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.distros.parsers import resolv_conf
-
-from cloudinit.tests.helpers import TestCase
-
-import re
-
-
-BASE_RESOLVE = '''
-; generated by /sbin/dhclient-script
-search blah.yahoo.com yahoo.com
-nameserver 10.15.44.14
-nameserver 10.15.30.92
-'''
-BASE_RESOLVE = BASE_RESOLVE.strip()
-
-
-class TestResolvHelper(TestCase):
- def test_parse_same(self):
- rp = resolv_conf.ResolvConf(BASE_RESOLVE)
- rp_r = str(rp).strip()
- self.assertEqual(BASE_RESOLVE, rp_r)
-
- def test_local_domain(self):
- rp = resolv_conf.ResolvConf(BASE_RESOLVE)
- self.assertIsNone(rp.local_domain)
-
- rp.local_domain = "bob"
- self.assertEqual('bob', rp.local_domain)
- self.assertIn('domain bob', str(rp))
-
- def test_nameservers(self):
- rp = resolv_conf.ResolvConf(BASE_RESOLVE)
- self.assertIn('10.15.44.14', rp.nameservers)
- self.assertIn('10.15.30.92', rp.nameservers)
- rp.add_nameserver('10.2')
- self.assertIn('10.2', rp.nameservers)
- self.assertIn('nameserver 10.2', str(rp))
- self.assertNotIn('10.3', rp.nameservers)
- self.assertEqual(len(rp.nameservers), 3)
- rp.add_nameserver('10.2')
- rp.add_nameserver('10.3')
- self.assertNotIn('10.3', rp.nameservers)
-
- def test_search_domains(self):
- rp = resolv_conf.ResolvConf(BASE_RESOLVE)
- self.assertIn('yahoo.com', rp.search_domains)
- self.assertIn('blah.yahoo.com', rp.search_domains)
- rp.add_search_domain('bbb.y.com')
- self.assertIn('bbb.y.com', rp.search_domains)
- self.assertTrue(re.search(r'search(.*)bbb.y.com(.*)', str(rp)))
- self.assertIn('bbb.y.com', rp.search_domains)
- rp.add_search_domain('bbb.y.com')
- self.assertEqual(len(rp.search_domains), 3)
- rp.add_search_domain('bbb2.y.com')
- self.assertEqual(len(rp.search_domains), 4)
- rp.add_search_domain('bbb3.y.com')
- self.assertEqual(len(rp.search_domains), 5)
- rp.add_search_domain('bbb4.y.com')
- self.assertEqual(len(rp.search_domains), 6)
- self.assertRaises(ValueError, rp.add_search_domain, 'bbb5.y.com')
- self.assertEqual(len(rp.search_domains), 6)
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_sles.py b/tests/unittests/test_distros/test_sles.py
deleted file mode 100644
index 33e3c457..00000000
--- a/tests/unittests/test_distros/test_sles.py
+++ /dev/null
@@ -1,12 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from cloudinit.tests.helpers import CiTestCase
-
-from . import _get_distro
-
-
-class TestSLES(CiTestCase):
-
- def test_get_distro(self):
- distro = _get_distro("sles")
- self.assertEqual(distro.osfamily, 'suse')
diff --git a/tests/unittests/test_distros/test_sysconfig.py b/tests/unittests/test_distros/test_sysconfig.py
deleted file mode 100644
index c1d5b693..00000000
--- a/tests/unittests/test_distros/test_sysconfig.py
+++ /dev/null
@@ -1,86 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-import re
-
-from cloudinit.distros.parsers.sys_conf import SysConf
-
-from cloudinit.tests.helpers import TestCase
-
-
-# Lots of good examples @
-# http://content.hccfl.edu/pollock/AUnix1/SysconfigFilesDesc.txt
-
-class TestSysConfHelper(TestCase):
- # This function was added in 2.7, make it work for 2.6
- def assertRegMatches(self, text, regexp):
- regexp = re.compile(regexp)
- self.assertTrue(regexp.search(text),
- msg="%s must match %s!" % (text, regexp.pattern))
-
- def test_parse_no_change(self):
- contents = '''# A comment
-USESMBAUTH=no
-KEYTABLE=/usr/lib/kbd/keytables/us.map
-SHORTDATE=$(date +%y:%m:%d:%H:%M)
-HOSTNAME=blahblah
-NETMASK0=255.255.255.0
-# Inline comment
-LIST=$LOGROOT/incremental-list
-IPV6TO4_ROUTING='eth0-:0004::1/64 eth1-:0005::1/64'
-ETHTOOL_OPTS="-K ${DEVICE} tso on; -G ${DEVICE} rx 256 tx 256"
-USEMD5=no'''
- conf = SysConf(contents.splitlines())
- self.assertEqual(conf['HOSTNAME'], 'blahblah')
- self.assertEqual(conf['SHORTDATE'], '$(date +%y:%m:%d:%H:%M)')
- # Should be unquoted
- self.assertEqual(conf['ETHTOOL_OPTS'], ('-K ${DEVICE} tso on; '
- '-G ${DEVICE} rx 256 tx 256'))
- self.assertEqual(contents, str(conf))
-
- def test_parse_shell_vars(self):
- contents = 'USESMBAUTH=$XYZ'
- conf = SysConf(contents.splitlines())
- self.assertEqual(contents, str(conf))
- conf = SysConf('')
- conf['B'] = '${ZZ}d apples'
- # Should be quoted
- self.assertEqual('B="${ZZ}d apples"', str(conf))
- conf = SysConf('')
- conf['B'] = '$? d apples'
- self.assertEqual('B="$? d apples"', str(conf))
- contents = 'IPMI_WATCHDOG_OPTIONS="timeout=60"'
- conf = SysConf(contents.splitlines())
- self.assertEqual('IPMI_WATCHDOG_OPTIONS=timeout=60', str(conf))
-
- def test_parse_adjust(self):
- contents = 'IPV6TO4_ROUTING="eth0-:0004::1/64 eth1-:0005::1/64"'
- conf = SysConf(contents.splitlines())
- # Should be unquoted
- self.assertEqual('eth0-:0004::1/64 eth1-:0005::1/64',
- conf['IPV6TO4_ROUTING'])
- conf['IPV6TO4_ROUTING'] = "blah \tblah"
- contents2 = str(conf).strip()
- # Should be requoted due to whitespace
- self.assertRegMatches(contents2,
- r'IPV6TO4_ROUTING=[\']blah\s+blah[\']')
-
- def test_parse_no_adjust_shell(self):
- conf = SysConf(''.splitlines())
- conf['B'] = ' $(time)'
- contents = str(conf)
- self.assertEqual('B= $(time)', contents)
-
- def test_parse_empty(self):
- contents = ''
- conf = SysConf(contents.splitlines())
- self.assertEqual('', str(conf).strip())
-
- def test_parse_add_new(self):
- contents = 'BLAH=b'
- conf = SysConf(contents.splitlines())
- conf['Z'] = 'd'
- contents = str(conf)
- self.assertIn("Z=d", contents)
- self.assertIn("BLAH=b", contents)
-
-# vi: ts=4 expandtab
diff --git a/tests/unittests/test_distros/test_user_data_normalize.py b/tests/unittests/test_distros/test_user_data_normalize.py
deleted file mode 100644
index 50c86942..00000000
--- a/tests/unittests/test_distros/test_user_data_normalize.py
+++ /dev/null
@@ -1,372 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-
-from unittest import mock
-
-from cloudinit import distros
-from cloudinit.distros import ug_util
-from cloudinit import helpers
-from cloudinit import settings
-
-from cloudinit.tests.helpers import TestCase
-
-
-bcfg = {
- 'name': 'bob',
- 'plain_text_passwd': 'ubuntu',
- 'home': "/home/ubuntu",
- 'shell': "/bin/bash",
- 'lock_passwd': True,
- 'gecos': "Ubuntu",
- 'groups': ["foo"]
-}
-
-
-class TestUGNormalize(TestCase):
-
- def setUp(self):
- super(TestUGNormalize, self).setUp()
- self.add_patch('cloudinit.util.system_is_snappy', 'm_snappy')
-
- def _make_distro(self, dtype, def_user=None):
- cfg = dict(settings.CFG_BUILTIN)
- cfg['system_info']['distro'] = dtype
- paths = helpers.Paths(cfg['system_info']['paths'])
- distro_cls = distros.fetch(dtype)
- if def_user:
- cfg['system_info']['default_user'] = def_user.copy()
- distro = distro_cls(dtype, cfg['system_info'], paths)
- return distro
-
- def _norm(self, cfg, distro):
- return ug_util.normalize_users_groups(cfg, distro)
-
- def test_group_dict(self):
- distro = self._make_distro('ubuntu')
- g = {'groups':
- [{'ubuntu': ['foo', 'bar'],
- 'bob': 'users'},
- 'cloud-users',
- {'bob': 'users2'}]}
- (_users, groups) = self._norm(g, distro)
- self.assertIn('ubuntu', groups)
- ub_members = groups['ubuntu']
- self.assertEqual(sorted(['foo', 'bar']), sorted(ub_members))
- self.assertIn('bob', groups)
- b_members = groups['bob']
- self.assertEqual(sorted(['users', 'users2']),
- sorted(b_members))
-
- def test_basic_groups(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'groups': ['bob'],
- }
- (users, groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', groups)
- self.assertEqual({}, users)
-
- def test_csv_groups(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'groups': 'bob,joe,steve',
- }
- (users, groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', groups)
- self.assertIn('joe', groups)
- self.assertIn('steve', groups)
- self.assertEqual({}, users)
-
- def test_more_groups(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'groups': ['bob', 'joe', 'steve']
- }
- (users, groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', groups)
- self.assertIn('joe', groups)
- self.assertIn('steve', groups)
- self.assertEqual({}, users)
-
- def test_member_groups(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'groups': {
- 'bob': ['s'],
- 'joe': [],
- 'steve': [],
- }
- }
- (users, groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', groups)
- self.assertEqual(['s'], groups['bob'])
- self.assertEqual([], groups['joe'])
- self.assertIn('joe', groups)
- self.assertIn('steve', groups)
- self.assertEqual({}, users)
-
- def test_users_simple_dict(self):
- distro = self._make_distro('ubuntu', bcfg)
- ug_cfg = {
- 'users': {
- 'default': True,
- }
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- ug_cfg = {
- 'users': {
- 'default': 'yes',
- }
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- ug_cfg = {
- 'users': {
- 'default': '1',
- }
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
-
- def test_users_simple_dict_no(self):
- distro = self._make_distro('ubuntu', bcfg)
- ug_cfg = {
- 'users': {
- 'default': False,
- }
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertEqual({}, users)
- ug_cfg = {
- 'users': {
- 'default': 'no',
- }
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertEqual({}, users)
-
- def test_users_simple_csv(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': 'joe,bob',
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('joe', users)
- self.assertIn('bob', users)
- self.assertEqual({'default': False}, users['joe'])
- self.assertEqual({'default': False}, users['bob'])
-
- def test_users_simple(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': [
- 'joe',
- 'bob'
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('joe', users)
- self.assertIn('bob', users)
- self.assertEqual({'default': False}, users['joe'])
- self.assertEqual({'default': False}, users['bob'])
-
- def test_users_old_user(self):
- distro = self._make_distro('ubuntu', bcfg)
- ug_cfg = {
- 'user': 'zetta',
- 'users': 'default'
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertNotIn('bob', users) # Bob is not the default now, zetta is
- self.assertIn('zetta', users)
- self.assertTrue(users['zetta']['default'])
- self.assertNotIn('default', users)
- ug_cfg = {
- 'user': 'zetta',
- 'users': 'default, joe'
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertNotIn('bob', users) # Bob is not the default now, zetta is
- self.assertIn('joe', users)
- self.assertIn('zetta', users)
- self.assertTrue(users['zetta']['default'])
- self.assertNotIn('default', users)
- ug_cfg = {
- 'user': 'zetta',
- 'users': ['bob', 'joe']
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- self.assertIn('joe', users)
- self.assertIn('zetta', users)
- self.assertTrue(users['zetta']['default'])
- ug_cfg = {
- 'user': 'zetta',
- 'users': {
- 'bob': True,
- 'joe': True,
- }
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- self.assertIn('joe', users)
- self.assertIn('zetta', users)
- self.assertTrue(users['zetta']['default'])
- ug_cfg = {
- 'user': 'zetta',
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('zetta', users)
- ug_cfg = {}
- (users, groups) = self._norm(ug_cfg, distro)
- self.assertEqual({}, users)
- self.assertEqual({}, groups)
-
- def test_users_dict_default_additional(self):
- distro = self._make_distro('ubuntu', bcfg)
- ug_cfg = {
- 'users': [
- {'name': 'default', 'blah': True}
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- self.assertEqual(",".join(distro.get_default_user()['groups']),
- users['bob']['groups'])
- self.assertEqual(True, users['bob']['blah'])
- self.assertEqual(True, users['bob']['default'])
-
- def test_users_dict_extract(self):
- distro = self._make_distro('ubuntu', bcfg)
- ug_cfg = {
- 'users': [
- 'default',
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- (name, config) = ug_util.extract_default(users)
- self.assertEqual(name, 'bob')
- expected_config = {}
- def_config = None
- try:
- def_config = distro.get_default_user()
- except NotImplementedError:
- pass
- if not def_config:
- def_config = {}
- expected_config.update(def_config)
-
- # Ignore these for now
- expected_config.pop('name', None)
- expected_config.pop('groups', None)
- config.pop('groups', None)
- self.assertEqual(config, expected_config)
-
- def test_users_dict_default(self):
- distro = self._make_distro('ubuntu', bcfg)
- ug_cfg = {
- 'users': [
- 'default',
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('bob', users)
- self.assertEqual(",".join(distro.get_default_user()['groups']),
- users['bob']['groups'])
- self.assertEqual(True, users['bob']['default'])
-
- def test_users_dict_trans(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': [
- {'name': 'joe',
- 'tr-me': True},
- {'name': 'bob'},
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('joe', users)
- self.assertIn('bob', users)
- self.assertEqual({'tr_me': True, 'default': False}, users['joe'])
- self.assertEqual({'default': False}, users['bob'])
-
- def test_users_dict(self):
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': [
- {'name': 'joe'},
- {'name': 'bob'},
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- self.assertIn('joe', users)
- self.assertIn('bob', users)
- self.assertEqual({'default': False}, users['joe'])
- self.assertEqual({'default': False}, users['bob'])
-
- @mock.patch('cloudinit.subp.subp')
- def test_create_snap_user(self, mock_subp):
- mock_subp.side_effect = [('{"username": "joe", "ssh-key-count": 1}\n',
- '')]
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': [
- {'name': 'joe', 'snapuser': 'joe@joe.com'},
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- for (user, config) in users.items():
- print('user=%s config=%s' % (user, config))
- username = distro.create_user(user, **config)
-
- snapcmd = ['snap', 'create-user', '--sudoer', '--json', 'joe@joe.com']
- mock_subp.assert_called_with(snapcmd, capture=True, logstring=snapcmd)
- self.assertEqual(username, 'joe')
-
- @mock.patch('cloudinit.subp.subp')
- def test_create_snap_user_known(self, mock_subp):
- mock_subp.side_effect = [('{"username": "joe", "ssh-key-count": 1}\n',
- '')]
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': [
- {'name': 'joe', 'snapuser': 'joe@joe.com', 'known': True},
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- for (user, config) in users.items():
- print('user=%s config=%s' % (user, config))
- username = distro.create_user(user, **config)
-
- snapcmd = ['snap', 'create-user', '--sudoer', '--json', '--known',
- 'joe@joe.com']
- mock_subp.assert_called_with(snapcmd, capture=True, logstring=snapcmd)
- self.assertEqual(username, 'joe')
-
- @mock.patch('cloudinit.util.system_is_snappy')
- @mock.patch('cloudinit.util.is_group')
- @mock.patch('cloudinit.subp.subp')
- def test_add_user_on_snappy_system(self, mock_subp, mock_isgrp,
- mock_snappy):
- mock_isgrp.return_value = False
- mock_subp.return_value = True
- mock_snappy.return_value = True
- distro = self._make_distro('ubuntu')
- ug_cfg = {
- 'users': [
- {'name': 'joe', 'groups': 'users', 'create_groups': True},
- ],
- }
- (users, _groups) = self._norm(ug_cfg, distro)
- for (user, config) in users.items():
- print('user=%s config=%s' % (user, config))
- distro.add_user(user, **config)
-
- groupcmd = ['groupadd', 'users', '--extrausers']
- addcmd = ['useradd', 'joe', '--extrausers', '--groups', 'users', '-m']
-
- mock_subp.assert_any_call(groupcmd)
- mock_subp.assert_any_call(addcmd, logstring=addcmd)
-
-# vi: ts=4 expandtab
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-10 05:50:01 +0000