summaryrefslogtreecommitdiff
path: root/tests/unittests/test_handler
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/test_handler')
-rw-r--r--tests/unittests/test_handler/test_handler_lxd.py16
-rw-r--r--tests/unittests/test_handler/test_handler_ntp.py23
-rw-r--r--tests/unittests/test_handler/test_handler_resizefs.py91
-rw-r--r--tests/unittests/test_handler/test_schema.py37
4 files changed, 113 insertions, 54 deletions
diff --git a/tests/unittests/test_handler/test_handler_lxd.py b/tests/unittests/test_handler/test_handler_lxd.py
index f132a778..e0d9ab6c 100644
--- a/tests/unittests/test_handler/test_handler_lxd.py
+++ b/tests/unittests/test_handler/test_handler_lxd.py
@@ -5,17 +5,16 @@ from cloudinit.sources import DataSourceNoCloud
from cloudinit import (distros, helpers, cloud)
from cloudinit.tests import helpers as t_help
-import logging
-
try:
from unittest import mock
except ImportError:
import mock
-LOG = logging.getLogger(__name__)
+class TestLxd(t_help.CiTestCase):
+
+ with_logs = True
-class TestLxd(t_help.TestCase):
lxd_cfg = {
'lxd': {
'init': {
@@ -41,7 +40,7 @@ class TestLxd(t_help.TestCase):
def test_lxd_init(self, mock_util):
cc = self._get_cloud('ubuntu')
mock_util.which.return_value = True
- cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, LOG, [])
+ cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, self.logger, [])
self.assertTrue(mock_util.which.called)
init_call = mock_util.subp.call_args_list[0][0][0]
self.assertEqual(init_call,
@@ -55,7 +54,8 @@ class TestLxd(t_help.TestCase):
cc = self._get_cloud('ubuntu')
cc.distro = mock.MagicMock()
mock_util.which.return_value = None
- cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, LOG, [])
+ cc_lxd.handle('cc_lxd', self.lxd_cfg, cc, self.logger, [])
+ self.assertNotIn('WARN', self.logs.getvalue())
self.assertTrue(cc.distro.install_packages.called)
install_pkg = cc.distro.install_packages.call_args_list[0][0][0]
self.assertEqual(sorted(install_pkg), ['lxd', 'zfs'])
@@ -64,7 +64,7 @@ class TestLxd(t_help.TestCase):
def test_no_init_does_nothing(self, mock_util):
cc = self._get_cloud('ubuntu')
cc.distro = mock.MagicMock()
- cc_lxd.handle('cc_lxd', {'lxd': {}}, cc, LOG, [])
+ cc_lxd.handle('cc_lxd', {'lxd': {}}, cc, self.logger, [])
self.assertFalse(cc.distro.install_packages.called)
self.assertFalse(mock_util.subp.called)
@@ -72,7 +72,7 @@ class TestLxd(t_help.TestCase):
def test_no_lxd_does_nothing(self, mock_util):
cc = self._get_cloud('ubuntu')
cc.distro = mock.MagicMock()
- cc_lxd.handle('cc_lxd', {'package_update': True}, cc, LOG, [])
+ cc_lxd.handle('cc_lxd', {'package_update': True}, cc, self.logger, [])
self.assertFalse(cc.distro.install_packages.called)
self.assertFalse(mock_util.subp.called)
diff --git a/tests/unittests/test_handler/test_handler_ntp.py b/tests/unittests/test_handler/test_handler_ntp.py
index 4f291248..3abe5786 100644
--- a/tests/unittests/test_handler/test_handler_ntp.py
+++ b/tests/unittests/test_handler/test_handler_ntp.py
@@ -293,23 +293,24 @@ class TestNtp(FilesystemMockingTestCase):
def test_ntp_handler_schema_validation_allows_empty_ntp_config(self):
"""Ntp schema validation allows for an empty ntp: configuration."""
- invalid_config = {'ntp': {}}
+ valid_empty_configs = [{'ntp': {}}, {'ntp': None}]
distro = 'ubuntu'
cc = self._get_cloud(distro)
ntp_conf = os.path.join(self.new_root, 'ntp.conf')
with open('{0}.tmpl'.format(ntp_conf), 'wb') as stream:
stream.write(NTP_TEMPLATE)
- with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf):
- cc_ntp.handle('cc_ntp', invalid_config, cc, None, [])
+ for valid_empty_config in valid_empty_configs:
+ with mock.patch('cloudinit.config.cc_ntp.NTP_CONF', ntp_conf):
+ cc_ntp.handle('cc_ntp', valid_empty_config, cc, None, [])
+ with open(ntp_conf) as stream:
+ content = stream.read()
+ default_pools = [
+ "{0}.{1}.pool.ntp.org".format(x, distro)
+ for x in range(0, cc_ntp.NR_POOL_SERVERS)]
+ self.assertEqual(
+ "servers []\npools {0}\n".format(default_pools),
+ content)
self.assertNotIn('Invalid config:', self.logs.getvalue())
- with open(ntp_conf) as stream:
- content = stream.read()
- default_pools = [
- "{0}.{1}.pool.ntp.org".format(x, distro)
- for x in range(0, cc_ntp.NR_POOL_SERVERS)]
- self.assertEqual(
- "servers []\npools {0}\n".format(default_pools),
- content)
@skipIf(_missing_jsonschema_dep, "No python-jsonschema dependency")
def test_ntp_handler_schema_validation_warns_non_string_item_type(self):
diff --git a/tests/unittests/test_handler/test_handler_resizefs.py b/tests/unittests/test_handler/test_handler_resizefs.py
index 3e5d436c..29d5574d 100644
--- a/tests/unittests/test_handler/test_handler_resizefs.py
+++ b/tests/unittests/test_handler/test_handler_resizefs.py
@@ -1,9 +1,9 @@
# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit.config.cc_resizefs import (
- can_skip_resize, handle, is_device_path_writable_block,
- rootdev_from_cmdline)
+ can_skip_resize, handle, maybe_get_writable_device_path)
+from collections import namedtuple
import logging
import textwrap
@@ -138,47 +138,48 @@ class TestRootDevFromCmdline(CiTestCase):
invalid_cases = [
'BOOT_IMAGE=/adsf asdfa werasef root adf', 'BOOT_IMAGE=/adsf', '']
for case in invalid_cases:
- self.assertIsNone(rootdev_from_cmdline(case))
+ self.assertIsNone(util.rootdev_from_cmdline(case))
def test_rootdev_from_cmdline_with_root_startswith_dev(self):
"""Return the cmdline root when the path starts with /dev."""
self.assertEqual(
- '/dev/this', rootdev_from_cmdline('asdf root=/dev/this'))
+ '/dev/this', util.rootdev_from_cmdline('asdf root=/dev/this'))
def test_rootdev_from_cmdline_with_root_without_dev_prefix(self):
"""Add /dev prefix to cmdline root when the path lacks the prefix."""
- self.assertEqual('/dev/this', rootdev_from_cmdline('asdf root=this'))
+ self.assertEqual(
+ '/dev/this', util.rootdev_from_cmdline('asdf root=this'))
def test_rootdev_from_cmdline_with_root_with_label(self):
"""When cmdline root contains a LABEL, our root is disk/by-label."""
self.assertEqual(
'/dev/disk/by-label/unique',
- rootdev_from_cmdline('asdf root=LABEL=unique'))
+ util.rootdev_from_cmdline('asdf root=LABEL=unique'))
def test_rootdev_from_cmdline_with_root_with_uuid(self):
"""When cmdline root contains a UUID, our root is disk/by-uuid."""
self.assertEqual(
'/dev/disk/by-uuid/adsfdsaf-adsf',
- rootdev_from_cmdline('asdf root=UUID=adsfdsaf-adsf'))
+ util.rootdev_from_cmdline('asdf root=UUID=adsfdsaf-adsf'))
-class TestIsDevicePathWritableBlock(CiTestCase):
+class TestMaybeGetDevicePathAsWritableBlock(CiTestCase):
with_logs = True
- def test_is_device_path_writable_block_false_on_overlayroot(self):
+ def test_maybe_get_writable_device_path_none_on_overlayroot(self):
"""When devpath is overlayroot (on MAAS), is_dev_writable is False."""
info = 'does not matter'
- is_writable = wrap_and_call(
+ devpath = wrap_and_call(
'cloudinit.config.cc_resizefs.util',
{'is_container': {'return_value': False}},
- is_device_path_writable_block, 'overlayroot', info, LOG)
- self.assertFalse(is_writable)
+ maybe_get_writable_device_path, 'overlayroot', info, LOG)
+ self.assertIsNone(devpath)
self.assertIn(
"Not attempting to resize devpath 'overlayroot'",
self.logs.getvalue())
- def test_is_device_path_writable_block_warns_missing_cmdline_root(self):
+ def test_maybe_get_writable_device_path_warns_missing_cmdline_root(self):
"""When root does not exist isn't in the cmdline, log warning."""
info = 'does not matter'
@@ -190,43 +191,43 @@ class TestIsDevicePathWritableBlock(CiTestCase):
exists_mock_path = 'cloudinit.config.cc_resizefs.os.path.exists'
with mock.patch(exists_mock_path) as m_exists:
m_exists.return_value = False
- is_writable = wrap_and_call(
+ devpath = wrap_and_call(
'cloudinit.config.cc_resizefs.util',
{'is_container': {'return_value': False},
'get_mount_info': {'side_effect': fake_mount_info},
'get_cmdline': {'return_value': 'BOOT_IMAGE=/vmlinuz.efi'}},
- is_device_path_writable_block, '/dev/root', info, LOG)
- self.assertFalse(is_writable)
+ maybe_get_writable_device_path, '/dev/root', info, LOG)
+ self.assertIsNone(devpath)
logs = self.logs.getvalue()
self.assertIn("WARNING: Unable to find device '/dev/root'", logs)
- def test_is_device_path_writable_block_does_not_exist(self):
+ def test_maybe_get_writable_device_path_does_not_exist(self):
"""When devpath does not exist, a warning is logged."""
info = 'dev=/I/dont/exist mnt_point=/ path=/dev/none'
- is_writable = wrap_and_call(
+ devpath = wrap_and_call(
'cloudinit.config.cc_resizefs.util',
{'is_container': {'return_value': False}},
- is_device_path_writable_block, '/I/dont/exist', info, LOG)
- self.assertFalse(is_writable)
+ maybe_get_writable_device_path, '/I/dont/exist', info, LOG)
+ self.assertIsNone(devpath)
self.assertIn(
"WARNING: Device '/I/dont/exist' did not exist."
' cannot resize: %s' % info,
self.logs.getvalue())
- def test_is_device_path_writable_block_does_not_exist_in_container(self):
+ def test_maybe_get_writable_device_path_does_not_exist_in_container(self):
"""When devpath does not exist in a container, log a debug message."""
info = 'dev=/I/dont/exist mnt_point=/ path=/dev/none'
- is_writable = wrap_and_call(
+ devpath = wrap_and_call(
'cloudinit.config.cc_resizefs.util',
{'is_container': {'return_value': True}},
- is_device_path_writable_block, '/I/dont/exist', info, LOG)
- self.assertFalse(is_writable)
+ maybe_get_writable_device_path, '/I/dont/exist', info, LOG)
+ self.assertIsNone(devpath)
self.assertIn(
"DEBUG: Device '/I/dont/exist' did not exist in container."
' cannot resize: %s' % info,
self.logs.getvalue())
- def test_is_device_path_writable_block_raises_oserror(self):
+ def test_maybe_get_writable_device_path_raises_oserror(self):
"""When unexpected OSError is raises by os.stat it is reraised."""
info = 'dev=/I/dont/exist mnt_point=/ path=/dev/none'
with self.assertRaises(OSError) as context_manager:
@@ -234,41 +235,63 @@ class TestIsDevicePathWritableBlock(CiTestCase):
'cloudinit.config.cc_resizefs',
{'util.is_container': {'return_value': True},
'os.stat': {'side_effect': OSError('Something unexpected')}},
- is_device_path_writable_block, '/I/dont/exist', info, LOG)
+ maybe_get_writable_device_path, '/I/dont/exist', info, LOG)
self.assertEqual(
'Something unexpected', str(context_manager.exception))
- def test_is_device_path_writable_block_non_block(self):
+ def test_maybe_get_writable_device_path_non_block(self):
"""When device is not a block device, emit warning return False."""
fake_devpath = self.tmp_path('dev/readwrite')
util.write_file(fake_devpath, '', mode=0o600) # read-write
info = 'dev=/dev/root mnt_point=/ path={0}'.format(fake_devpath)
- is_writable = wrap_and_call(
+ devpath = wrap_and_call(
'cloudinit.config.cc_resizefs.util',
{'is_container': {'return_value': False}},
- is_device_path_writable_block, fake_devpath, info, LOG)
- self.assertFalse(is_writable)
+ maybe_get_writable_device_path, fake_devpath, info, LOG)
+ self.assertIsNone(devpath)
self.assertIn(
"WARNING: device '{0}' not a block device. cannot resize".format(
fake_devpath),
self.logs.getvalue())
- def test_is_device_path_writable_block_non_block_on_container(self):
+ def test_maybe_get_writable_device_path_non_block_on_container(self):
"""When device is non-block device in container, emit debug log."""
fake_devpath = self.tmp_path('dev/readwrite')
util.write_file(fake_devpath, '', mode=0o600) # read-write
info = 'dev=/dev/root mnt_point=/ path={0}'.format(fake_devpath)
- is_writable = wrap_and_call(
+ devpath = wrap_and_call(
'cloudinit.config.cc_resizefs.util',
{'is_container': {'return_value': True}},
- is_device_path_writable_block, fake_devpath, info, LOG)
- self.assertFalse(is_writable)
+ maybe_get_writable_device_path, fake_devpath, info, LOG)
+ self.assertIsNone(devpath)
self.assertIn(
"DEBUG: device '{0}' not a block device in container."
' cannot resize'.format(fake_devpath),
self.logs.getvalue())
+ def test_maybe_get_writable_device_path_returns_cmdline_root(self):
+ """When root device is UUID in kernel commandline, update devpath."""
+ # XXX Long-term we want to use FilesystemMocking test to avoid
+ # touching os.stat.
+ FakeStat = namedtuple(
+ 'FakeStat', ['st_mode', 'st_size', 'st_mtime']) # minimal def.
+ info = 'dev=/dev/root mnt_point=/ path=/does/not/matter'
+ devpath = wrap_and_call(
+ 'cloudinit.config.cc_resizefs',
+ {'util.get_cmdline': {'return_value': 'asdf root=UUID=my-uuid'},
+ 'util.is_container': False,
+ 'os.path.exists': False, # /dev/root doesn't exist
+ 'os.stat': {
+ 'return_value': FakeStat(25008, 0, 1)} # char block device
+ },
+ maybe_get_writable_device_path, '/dev/root', info, LOG)
+ self.assertEqual('/dev/disk/by-uuid/my-uuid', devpath)
+ self.assertIn(
+ "DEBUG: Converted /dev/root to '/dev/disk/by-uuid/my-uuid'"
+ " per kernel cmdline",
+ self.logs.getvalue())
+
# vi: ts=4 expandtab
diff --git a/tests/unittests/test_handler/test_schema.py b/tests/unittests/test_handler/test_schema.py
index b8fc8930..648573f6 100644
--- a/tests/unittests/test_handler/test_schema.py
+++ b/tests/unittests/test_handler/test_schema.py
@@ -4,11 +4,12 @@ from cloudinit.config.schema import (
CLOUD_CONFIG_HEADER, SchemaValidationError, annotated_cloudconfig_file,
get_schema_doc, get_schema, validate_cloudconfig_file,
validate_cloudconfig_schema, main)
-from cloudinit.util import write_file
+from cloudinit.util import subp, write_file
from cloudinit.tests.helpers import CiTestCase, mock, skipIf
from copy import copy
+import os
from six import StringIO
from textwrap import dedent
from yaml import safe_load
@@ -364,4 +365,38 @@ class MainTest(CiTestCase):
self.assertIn(
'Valid cloud-config file {0}'.format(myyaml), m_stdout.getvalue())
+
+class CloudTestsIntegrationTest(CiTestCase):
+ """Validate all cloud-config yaml schema provided in integration tests.
+
+ It is less expensive to have unittests validate schema of all cloud-config
+ yaml provided to integration tests, than to run an integration test which
+ raises Warnings or errors on invalid cloud-config schema.
+ """
+
+ @skipIf(_missing_jsonschema_dep, "No python-jsonschema dependency")
+ def test_all_integration_test_cloud_config_schema(self):
+ """Validate schema of cloud_tests yaml files looking for warnings."""
+ schema = get_schema()
+ testsdir = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+ integration_testdir = os.path.sep.join(
+ [testsdir, 'cloud_tests', 'testcases'])
+ errors = []
+ out, _ = subp(['find', integration_testdir, '-name', '*yaml'])
+ for filename in out.splitlines():
+ test_cfg = safe_load(open(filename))
+ cloud_config = test_cfg.get('cloud_config')
+ if cloud_config:
+ cloud_config = safe_load(
+ cloud_config.replace("#cloud-config\n", ""))
+ try:
+ validate_cloudconfig_schema(
+ cloud_config, schema, strict=True)
+ except SchemaValidationError as e:
+ errors.append(
+ '{0}: {1}'.format(
+ filename, e))
+ if errors:
+ raise AssertionError(', '.join(errors))
+
# vi: ts=4 expandtab syntax=python
generated by cgit v1.2.3 (git 2.39.1) at 2025-06-04 18:59:47 +0000