summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unittests/test_cli.py2
-rw-r--r--tests/unittests/test_handler/test_handler_apk_configure.py299
-rw-r--r--tests/unittests/test_handler/test_handler_ca_certs.py11
-rw-r--r--tests/unittests/test_handler/test_handler_ntp.py129
-rw-r--r--tests/unittests/test_handler/test_handler_power_state.py29
-rw-r--r--tests/unittests/test_handler/test_schema.py1
6 files changed, 419 insertions, 52 deletions
diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py
index 43d996b9..dcf0fe5a 100644
--- a/tests/unittests/test_cli.py
+++ b/tests/unittests/test_cli.py
@@ -224,7 +224,7 @@ class TestCLI(test_helpers.FilesystemMockingTestCase):
self._call_main(['cloud-init', 'devel', 'schema', '--docs', 'all'])
expected_doc_sections = [
'**Supported distros:** all',
- '**Supported distros:** centos, debian, fedora',
+ '**Supported distros:** alpine, centos, debian, fedora',
'**Config schema**:\n **resize_rootfs:** (true/false/noblock)',
'**Examples**::\n\n runcmd:\n - [ ls, -l, / ]\n'
]
diff --git a/tests/unittests/test_handler/test_handler_apk_configure.py b/tests/unittests/test_handler/test_handler_apk_configure.py
new file mode 100644
index 00000000..8acc0b33
--- /dev/null
+++ b/tests/unittests/test_handler/test_handler_apk_configure.py
@@ -0,0 +1,299 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+""" test_apk_configure
+Test creation of repositories file
+"""
+
+import logging
+import os
+import textwrap
+
+from cloudinit import (cloud, helpers, util)
+
+from cloudinit.config import cc_apk_configure
+from cloudinit.tests.helpers import (FilesystemMockingTestCase, mock)
+
+REPO_FILE = "/etc/apk/repositories"
+DEFAULT_MIRROR_URL = "https://alpine.global.ssl.fastly.net/alpine"
+CC_APK = 'cloudinit.config.cc_apk_configure'
+
+
+class TestNoConfig(FilesystemMockingTestCase):
+ def setUp(self):
+ super(TestNoConfig, self).setUp()
+ self.add_patch(CC_APK + '._write_repositories_file', 'm_write_repos')
+ self.name = "apk-configure"
+ self.cloud_init = None
+ self.log = logging.getLogger("TestNoConfig")
+ self.args = []
+
+ def test_no_config(self):
+ """
+ Test that nothing is done if no apk-configure
+ configuration is provided.
+ """
+ config = util.get_builtin_cfg()
+
+ cc_apk_configure.handle(self.name, config, self.cloud_init,
+ self.log, self.args)
+
+ self.assertEqual(0, self.m_write_repos.call_count)
+
+
+class TestConfig(FilesystemMockingTestCase):
+ def setUp(self):
+ super(TestConfig, self).setUp()
+ self.new_root = self.tmp_dir()
+ self.new_root = self.reRoot(root=self.new_root)
+ for dirname in ['tmp', 'etc/apk']:
+ util.ensure_dir(os.path.join(self.new_root, dirname))
+ self.paths = helpers.Paths({'templates_dir': self.new_root})
+ self.name = "apk-configure"
+ self.cloud = cloud.Cloud(None, self.paths, None, None, None)
+ self.log = logging.getLogger("TestNoConfig")
+ self.args = []
+
+ @mock.patch(CC_APK + '._write_repositories_file')
+ def test_no_repo_settings(self, m_write_repos):
+ """
+ Test that nothing is written if the 'alpine-repo' key
+ is not present.
+ """
+ config = {"apk_repos": {}}
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ self.assertEqual(0, m_write_repos.call_count)
+
+ @mock.patch(CC_APK + '._write_repositories_file')
+ def test_empty_repo_settings(self, m_write_repos):
+ """
+ Test that nothing is written if 'alpine_repo' list is empty.
+ """
+ config = {"apk_repos": {"alpine_repo": []}}
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ self.assertEqual(0, m_write_repos.call_count)
+
+ def test_only_main_repo(self):
+ """
+ Test when only details of main repo is written to file.
+ """
+ alpine_version = 'v3.12'
+ config = {
+ "apk_repos": {
+ "alpine_repo": {
+ "version": alpine_version
+ }
+ }
+ }
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ expected_content = textwrap.dedent("""\
+ #
+ # Created by cloud-init
+ #
+ # This file is written on first boot of an instance
+ #
+
+ {0}/{1}/main
+
+ """.format(DEFAULT_MIRROR_URL, alpine_version))
+
+ self.assertEqual(expected_content, util.load_file(REPO_FILE))
+
+ def test_main_and_community_repos(self):
+ """
+ Test when only details of main and community repos are
+ written to file.
+ """
+ alpine_version = 'edge'
+ config = {
+ "apk_repos": {
+ "alpine_repo": {
+ "version": alpine_version,
+ "community_enabled": True
+ }
+ }
+ }
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ expected_content = textwrap.dedent("""\
+ #
+ # Created by cloud-init
+ #
+ # This file is written on first boot of an instance
+ #
+
+ {0}/{1}/main
+ {0}/{1}/community
+
+ """.format(DEFAULT_MIRROR_URL, alpine_version))
+
+ self.assertEqual(expected_content, util.load_file(REPO_FILE))
+
+ def test_main_community_testing_repos(self):
+ """
+ Test when details of main, community and testing repos
+ are written to file.
+ """
+ alpine_version = 'v3.12'
+ config = {
+ "apk_repos": {
+ "alpine_repo": {
+ "version": alpine_version,
+ "community_enabled": True,
+ "testing_enabled": True
+ }
+ }
+ }
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ expected_content = textwrap.dedent("""\
+ #
+ # Created by cloud-init
+ #
+ # This file is written on first boot of an instance
+ #
+
+ {0}/{1}/main
+ {0}/{1}/community
+ #
+ # Testing - using with non-Edge installation may cause problems!
+ #
+ {0}/edge/testing
+
+ """.format(DEFAULT_MIRROR_URL, alpine_version))
+
+ self.assertEqual(expected_content, util.load_file(REPO_FILE))
+
+ def test_edge_main_community_testing_repos(self):
+ """
+ Test when details of main, community and testing repos
+ for Edge version of Alpine are written to file.
+ """
+ alpine_version = 'edge'
+ config = {
+ "apk_repos": {
+ "alpine_repo": {
+ "version": alpine_version,
+ "community_enabled": True,
+ "testing_enabled": True
+ }
+ }
+ }
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ expected_content = textwrap.dedent("""\
+ #
+ # Created by cloud-init
+ #
+ # This file is written on first boot of an instance
+ #
+
+ {0}/{1}/main
+ {0}/{1}/community
+ {0}/{1}/testing
+
+ """.format(DEFAULT_MIRROR_URL, alpine_version))
+
+ self.assertEqual(expected_content, util.load_file(REPO_FILE))
+
+ def test_main_community_testing_local_repos(self):
+ """
+ Test when details of main, community, testing and
+ local repos are written to file.
+ """
+ alpine_version = 'v3.12'
+ local_repo_url = 'http://some.mirror/whereever'
+ config = {
+ "apk_repos": {
+ "alpine_repo": {
+ "version": alpine_version,
+ "community_enabled": True,
+ "testing_enabled": True
+ },
+ "local_repo_base_url": local_repo_url
+ }
+ }
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ expected_content = textwrap.dedent("""\
+ #
+ # Created by cloud-init
+ #
+ # This file is written on first boot of an instance
+ #
+
+ {0}/{1}/main
+ {0}/{1}/community
+ #
+ # Testing - using with non-Edge installation may cause problems!
+ #
+ {0}/edge/testing
+
+ #
+ # Local repo
+ #
+ {2}/{1}
+
+ """.format(DEFAULT_MIRROR_URL, alpine_version, local_repo_url))
+
+ self.assertEqual(expected_content, util.load_file(REPO_FILE))
+
+ def test_edge_main_community_testing_local_repos(self):
+ """
+ Test when details of main, community, testing and local repos
+ for Edge version of Alpine are written to file.
+ """
+ alpine_version = 'edge'
+ local_repo_url = 'http://some.mirror/whereever'
+ config = {
+ "apk_repos": {
+ "alpine_repo": {
+ "version": alpine_version,
+ "community_enabled": True,
+ "testing_enabled": True
+ },
+ "local_repo_base_url": local_repo_url
+ }
+ }
+
+ cc_apk_configure.handle(self.name, config, self.cloud, self.log,
+ self.args)
+
+ expected_content = textwrap.dedent("""\
+ #
+ # Created by cloud-init
+ #
+ # This file is written on first boot of an instance
+ #
+
+ {0}/{1}/main
+ {0}/{1}/community
+ {0}/edge/testing
+
+ #
+ # Local repo
+ #
+ {2}/{1}
+
+ """.format(DEFAULT_MIRROR_URL, alpine_version, local_repo_url))
+
+ self.assertEqual(expected_content, util.load_file(REPO_FILE))
+
+
+# vi: ts=4 expandtab
diff --git a/tests/unittests/test_handler/test_handler_ca_certs.py b/tests/unittests/test_handler/test_handler_ca_certs.py
index c1aff181..e74a0a08 100644
--- a/tests/unittests/test_handler/test_handler_ca_certs.py
+++ b/tests/unittests/test_handler/test_handler_ca_certs.py
@@ -1,6 +1,7 @@
# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit import cloud
+from cloudinit import distros
from cloudinit.config import cc_ca_certs
from cloudinit import helpers
from cloudinit import subp
@@ -46,8 +47,9 @@ class TestConfig(TestCase):
def setUp(self):
super(TestConfig, self).setUp()
self.name = "ca-certs"
+ distro = self._fetch_distro('ubuntu')
self.paths = None
- self.cloud = cloud.Cloud(None, self.paths, None, None, None)
+ self.cloud = cloud.Cloud(None, self.paths, None, distro, None)
self.log = logging.getLogger("TestNoConfig")
self.args = []
@@ -62,6 +64,11 @@ class TestConfig(TestCase):
self.mock_remove = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'remove_default_ca_certs'))
+ def _fetch_distro(self, kind):
+ cls = distros.fetch(kind)
+ paths = helpers.Paths({})
+ return cls(kind, {}, paths)
+
def test_no_trusted_list(self):
"""
Test that no certificates are written if the 'trusted' key is not
@@ -275,7 +282,7 @@ class TestRemoveDefaultCaCerts(TestCase):
mock.patch.object(util, 'write_file'))
mock_subp = mocks.enter_context(mock.patch.object(subp, 'subp'))
- cc_ca_certs.remove_default_ca_certs()
+ cc_ca_certs.remove_default_ca_certs('ubuntu')
mock_delete.assert_has_calls([
mock.call("/usr/share/ca-certificates/"),
diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py
index 92a33ec1..6b9c8377 100644
--- a/tests/unittests/test_handler/test_handler_ntp.py
+++ b/tests/unittests/test_handler/test_handler_ntp.py
@@ -239,6 +239,35 @@ class TestNtp(FilesystemMockingTestCase):
self.assertEqual(delta[distro][client][key],
result[client][key])
+ def _get_expected_pools(self, pools, distro, client):
+ if client in ['ntp', 'chrony']:
+ if client == 'ntp' and distro == 'alpine':
+ # NTP for Alpine Linux is Busybox's ntp which does not
+ # support 'pool' lines in its configuration file.
+ expected_pools = []
+ else:
+ expected_pools = [
+ 'pool {0} iburst'.format(pool) for pool in pools]
+ elif client == 'systemd-timesyncd':
+ expected_pools = " ".join(pools)
+
+ return expected_pools
+
+ def _get_expected_servers(self, servers, distro, client):
+ if client in ['ntp', 'chrony']:
+ if client == 'ntp' and distro == 'alpine':
+ # NTP for Alpine Linux is Busybox's ntp which only supports
+ # 'server' lines without iburst option.
+ expected_servers = [
+ 'server {0}'.format(srv) for srv in servers]
+ else:
+ expected_servers = [
+ 'server {0} iburst'.format(srv) for srv in servers]
+ elif client == 'systemd-timesyncd':
+ expected_servers = " ".join(servers)
+
+ return expected_servers
+
def test_ntp_handler_real_distro_ntp_templates(self):
"""Test ntp handler renders the shipped distro ntp client templates."""
pools = ['0.mycompany.pool.ntp.org', '3.mycompany.pool.ntp.org']
@@ -269,27 +298,35 @@ class TestNtp(FilesystemMockingTestCase):
content = util.load_file(confpath)
if client in ['ntp', 'chrony']:
content_lines = content.splitlines()
- expected_servers = [
- 'server {0} iburst'.format(srv) for srv in servers]
+ expected_servers = self._get_expected_servers(servers,
+ distro,
+ client)
print('distro=%s client=%s' % (distro, client))
for sline in expected_servers:
self.assertIn(sline, content_lines,
('failed to render {0} conf'
' for distro:{1}'.format(client,
distro)))
- expected_pools = [
- 'pool {0} iburst'.format(pool) for pool in pools]
- for pline in expected_pools:
- self.assertIn(pline, content_lines,
- ('failed to render {0} conf'
- ' for distro:{1}'.format(client,
- distro)))
+ expected_pools = self._get_expected_pools(pools, distro,
+ client)
+ if expected_pools != []:
+ for pline in expected_pools:
+ self.assertIn(pline, content_lines,
+ ('failed to render {0} conf'
+ ' for distro:{1}'.format(client,
+ distro)))
elif client == 'systemd-timesyncd':
+ expected_servers = self._get_expected_servers(servers,
+ distro,
+ client)
+ expected_pools = self._get_expected_pools(pools,
+ distro,
+ client)
expected_content = (
"# cloud-init generated file\n" +
"# See timesyncd.conf(5) for details.\n\n" +
- "[Time]\nNTP=%s %s \n" % (" ".join(servers),
- " ".join(pools)))
+ "[Time]\nNTP=%s %s \n" % (expected_servers,
+ expected_pools))
self.assertEqual(expected_content, content)
def test_no_ntpcfg_does_nothing(self):
@@ -312,10 +349,20 @@ class TestNtp(FilesystemMockingTestCase):
confpath = ntpconfig['confpath']
m_select.return_value = ntpconfig
cc_ntp.handle('cc_ntp', valid_empty_config, mycloud, None, [])
- pools = cc_ntp.generate_server_names(mycloud.distro.name)
- self.assertEqual(
- "servers []\npools {0}\n".format(pools),
- util.load_file(confpath))
+ if distro == 'alpine':
+ # _mock_ntp_client_config call above did not specify a
+ # client value and so it defaults to "ntp" which on
+ # Alpine Linux only supports servers and not pools.
+
+ servers = cc_ntp.generate_server_names(mycloud.distro.name)
+ self.assertEqual(
+ "servers {0}\npools []\n".format(servers),
+ util.load_file(confpath))
+ else:
+ pools = cc_ntp.generate_server_names(mycloud.distro.name)
+ self.assertEqual(
+ "servers []\npools {0}\n".format(pools),
+ util.load_file(confpath))
self.assertNotIn('Invalid config:', self.logs.getvalue())
@skipUnlessJsonSchema()
@@ -374,18 +421,19 @@ class TestNtp(FilesystemMockingTestCase):
invalid_config = {
'ntp': {'invalidkey': 1, 'pools': ['0.mycompany.pool.ntp.org']}}
for distro in cc_ntp.distros:
- mycloud = self._get_cloud(distro)
- ntpconfig = self._mock_ntp_client_config(distro=distro)
- confpath = ntpconfig['confpath']
- m_select.return_value = ntpconfig
- cc_ntp.handle('cc_ntp', invalid_config, mycloud, None, [])
- self.assertIn(
- "Invalid config:\nntp: Additional properties are not allowed "
- "('invalidkey' was unexpected)",
- self.logs.getvalue())
- self.assertEqual(
- "servers []\npools ['0.mycompany.pool.ntp.org']\n",
- util.load_file(confpath))
+ if distro != 'alpine':
+ mycloud = self._get_cloud(distro)
+ ntpconfig = self._mock_ntp_client_config(distro=distro)
+ confpath = ntpconfig['confpath']
+ m_select.return_value = ntpconfig
+ cc_ntp.handle('cc_ntp', invalid_config, mycloud, None, [])
+ self.assertIn(
+ "Invalid config:\nntp: Additional properties are not "
+ "allowed ('invalidkey' was unexpected)",
+ self.logs.getvalue())
+ self.assertEqual(
+ "servers []\npools ['0.mycompany.pool.ntp.org']\n",
+ util.load_file(confpath))
@skipUnlessJsonSchema()
@mock.patch('cloudinit.config.cc_ntp.select_ntp_client')
@@ -452,9 +500,22 @@ class TestNtp(FilesystemMockingTestCase):
confpath = ntpconfig['confpath']
service_name = ntpconfig['service_name']
m_select.return_value = ntpconfig
- pools = cc_ntp.generate_server_names(mycloud.distro.name)
- # force uses systemd path
- m_sysd.return_value = True
+
+ hosts = cc_ntp.generate_server_names(mycloud.distro.name)
+ uses_systemd = True
+ expected_service_call = ['systemctl', 'reload-or-restart',
+ service_name]
+ expected_content = "servers []\npools {0}\n".format(hosts)
+
+ if distro == 'alpine':
+ uses_systemd = False
+ expected_service_call = ['service', service_name, 'restart']
+ # _mock_ntp_client_config call above did not specify a client
+ # value and so it defaults to "ntp" which on Alpine Linux only
+ # supports servers and not pools.
+ expected_content = "servers {0}\npools []\n".format(hosts)
+
+ m_sysd.return_value = uses_systemd
with mock.patch('cloudinit.config.cc_ntp.util') as m_util:
# allow use of util.mergemanydict
m_util.mergemanydict.side_effect = util.mergemanydict
@@ -465,11 +526,9 @@ class TestNtp(FilesystemMockingTestCase):
cfg['ntp']['enabled'])
cc_ntp.handle('notimportant', cfg, mycloud, None, None)
m_subp.subp.assert_called_with(
- ['systemctl', 'reload-or-restart',
- service_name], capture=True)
- self.assertEqual(
- "servers []\npools {0}\n".format(pools),
- util.load_file(confpath))
+ expected_service_call, capture=True)
+
+ self.assertEqual(expected_content, util.load_file(confpath))
def test_opensuse_picks_chrony(self):
"""Test opensuse picks chrony or ntp on certain distro versions"""
diff --git a/tests/unittests/test_handler/test_handler_power_state.py b/tests/unittests/test_handler/test_handler_power_state.py
index 0d8d17b9..93b24fdc 100644
--- a/tests/unittests/test_handler/test_handler_power_state.py
+++ b/tests/unittests/test_handler/test_handler_power_state.py
@@ -11,62 +11,63 @@ from cloudinit.tests.helpers import mock
class TestLoadPowerState(t_help.TestCase):
def test_no_config(self):
# completely empty config should mean do nothing
- (cmd, _timeout, _condition) = psc.load_power_state({})
+ (cmd, _timeout, _condition) = psc.load_power_state({}, 'ubuntu')
self.assertIsNone(cmd)
def test_irrelevant_config(self):
# no power_state field in config should return None for cmd
- (cmd, _timeout, _condition) = psc.load_power_state({'foo': 'bar'})
+ (cmd, _timeout, _condition) = psc.load_power_state({'foo': 'bar'},
+ 'ubuntu')
self.assertIsNone(cmd)
def test_invalid_mode(self):
cfg = {'power_state': {'mode': 'gibberish'}}
- self.assertRaises(TypeError, psc.load_power_state, cfg)
+ self.assertRaises(TypeError, psc.load_power_state, cfg, 'ubuntu')
cfg = {'power_state': {'mode': ''}}
- self.assertRaises(TypeError, psc.load_power_state, cfg)
+ self.assertRaises(TypeError, psc.load_power_state, cfg, 'ubuntu')
def test_empty_mode(self):
cfg = {'power_state': {'message': 'goodbye'}}
- self.assertRaises(TypeError, psc.load_power_state, cfg)
+ self.assertRaises(TypeError, psc.load_power_state, cfg, 'ubuntu')
def test_valid_modes(self):
cfg = {'power_state': {}}
for mode in ('halt', 'poweroff', 'reboot'):
cfg['power_state']['mode'] = mode
- check_lps_ret(psc.load_power_state(cfg), mode=mode)
+ check_lps_ret(psc.load_power_state(cfg, 'ubuntu'), mode=mode)
def test_invalid_delay(self):
cfg = {'power_state': {'mode': 'poweroff', 'delay': 'goodbye'}}
- self.assertRaises(TypeError, psc.load_power_state, cfg)
+ self.assertRaises(TypeError, psc.load_power_state, cfg, 'ubuntu')
def test_valid_delay(self):
cfg = {'power_state': {'mode': 'poweroff', 'delay': ''}}
for delay in ("now", "+1", "+30"):
cfg['power_state']['delay'] = delay
- check_lps_ret(psc.load_power_state(cfg))
+ check_lps_ret(psc.load_power_state(cfg, 'ubuntu'))
def test_message_present(self):
cfg = {'power_state': {'mode': 'poweroff', 'message': 'GOODBYE'}}
- ret = psc.load_power_state(cfg)
- check_lps_ret(psc.load_power_state(cfg))
+ ret = psc.load_power_state(cfg, 'ubuntu')
+ check_lps_ret(psc.load_power_state(cfg, 'ubuntu'))
self.assertIn(cfg['power_state']['message'], ret[0])
def test_no_message(self):
# if message is not present, then no argument should be passed for it
cfg = {'power_state': {'mode': 'poweroff'}}
- (cmd, _timeout, _condition) = psc.load_power_state(cfg)
+ (cmd, _timeout, _condition) = psc.load_power_state(cfg, 'ubuntu')
self.assertNotIn("", cmd)
- check_lps_ret(psc.load_power_state(cfg))
+ check_lps_ret(psc.load_power_state(cfg, 'ubuntu'))
self.assertTrue(len(cmd) == 3)
def test_condition_null_raises(self):
cfg = {'power_state': {'mode': 'poweroff', 'condition': None}}
- self.assertRaises(TypeError, psc.load_power_state, cfg)
+ self.assertRaises(TypeError, psc.load_power_state, cfg, 'ubuntu')
def test_condition_default_is_true(self):
cfg = {'power_state': {'mode': 'poweroff'}}
- _cmd, _timeout, cond = psc.load_power_state(cfg)
+ _cmd, _timeout, cond = psc.load_power_state(cfg, 'ubuntu')
self.assertEqual(cond, True)
diff --git a/tests/unittests/test_handler/test_schema.py b/tests/unittests/test_handler/test_schema.py
index 99f0b06c..44292571 100644
--- a/tests/unittests/test_handler/test_schema.py
+++ b/tests/unittests/test_handler/test_schema.py
@@ -24,6 +24,7 @@ class GetSchemaTest(CiTestCase):
schema = get_schema()
self.assertCountEqual(
[
+ 'cc_apk_configure',
'cc_apt_configure',
'cc_bootcmd',
'cc_locale',
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-28 23:26:01 +0000