summaryrefslogtreecommitdiff
path: root/tests/unittests
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests')
-rw-r--r--tests/unittests/test_cli.py21
-rw-r--r--tests/unittests/test_handler/test_handler_runcmd.py108
-rw-r--r--tests/unittests/test_handler/test_schema.py157
3 files changed, 273 insertions, 13 deletions
diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py
index 7780f164..24498802 100644
--- a/tests/unittests/test_cli.py
+++ b/tests/unittests/test_cli.py
@@ -46,7 +46,7 @@ class TestCLI(test_helpers.FilesystemMockingTestCase):
self._call_main()
error = self.stderr.getvalue()
expected_subcommands = ['analyze', 'init', 'modules', 'single',
- 'dhclient-hook', 'features']
+ 'dhclient-hook', 'features', 'devel']
for subcommand in expected_subcommands:
self.assertIn(subcommand, error)
@@ -79,6 +79,25 @@ class TestCLI(test_helpers.FilesystemMockingTestCase):
for subcommand in expected_subcommands:
self.assertIn(subcommand, error)
+ def test_devel_subcommand_parser(self):
+ """The subcommand cloud-init devel calls the correct subparser."""
+ self._call_main(['cloud-init', 'devel'])
+ # These subcommands only valid for cloud-init schema script
+ expected_subcommands = ['schema']
+ error = self.stderr.getvalue()
+ for subcommand in expected_subcommands:
+ self.assertIn(subcommand, error)
+
+ @mock.patch('cloudinit.config.schema.handle_schema_args')
+ def test_wb_devel_schema_subcommand_parser(self, m_schema):
+ """The subcommand cloud-init schema calls the correct subparser."""
+ exit_code = self._call_main(['cloud-init', 'devel', 'schema'])
+ self.assertEqual(1, exit_code)
+ # Known whitebox output from schema subcommand
+ self.assertEqual(
+ 'Expected either --config-file argument or --doc\n',
+ self.stderr.getvalue())
+
@mock.patch('cloudinit.cmd.main.main_single')
def test_single_subcommand(self, m_main_single):
"""The subcommand 'single' calls main_single with valid args."""
diff --git a/tests/unittests/test_handler/test_handler_runcmd.py b/tests/unittests/test_handler/test_handler_runcmd.py
new file mode 100644
index 00000000..7880ee72
--- /dev/null
+++ b/tests/unittests/test_handler/test_handler_runcmd.py
@@ -0,0 +1,108 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+from cloudinit.config import cc_runcmd
+from cloudinit.sources import DataSourceNone
+from cloudinit import (distros, helpers, cloud, util)
+from ..helpers import FilesystemMockingTestCase, skipIf
+
+import logging
+import os
+import stat
+
+try:
+ import jsonschema
+ assert jsonschema # avoid pyflakes error F401: import unused
+ _missing_jsonschema_dep = False
+except ImportError:
+ _missing_jsonschema_dep = True
+
+LOG = logging.getLogger(__name__)
+
+
+class TestRuncmd(FilesystemMockingTestCase):
+
+ with_logs = True
+
+ def setUp(self):
+ super(TestRuncmd, self).setUp()
+ self.subp = util.subp
+ self.new_root = self.tmp_dir()
+
+ def _get_cloud(self, distro):
+ self.patchUtils(self.new_root)
+ paths = helpers.Paths({'scripts': self.new_root})
+ cls = distros.fetch(distro)
+ mydist = cls(distro, {}, paths)
+ myds = DataSourceNone.DataSourceNone({}, mydist, paths)
+ paths.datasource = myds
+ return cloud.Cloud(myds, paths, {}, mydist, None)
+
+ def test_handler_skip_if_no_runcmd(self):
+ """When the provided config doesn't contain runcmd, skip it."""
+ cfg = {}
+ mycloud = self._get_cloud('ubuntu')
+ cc_runcmd.handle('notimportant', cfg, mycloud, LOG, None)
+ self.assertIn(
+ "Skipping module named notimportant, no 'runcmd' key",
+ self.logs.getvalue())
+
+ def test_handler_invalid_command_set(self):
+ """Commands which can't be converted to shell will raise errors."""
+ invalid_config = {'runcmd': 1}
+ cc = self._get_cloud('ubuntu')
+ cc_runcmd.handle('cc_runcmd', invalid_config, cc, LOG, [])
+ self.assertIn(
+ 'Failed to shellify 1 into file'
+ ' /var/lib/cloud/instances/iid-datasource-none/scripts/runcmd',
+ self.logs.getvalue())
+
+ @skipIf(_missing_jsonschema_dep, "No python-jsonschema dependency")
+ def test_handler_schema_validation_warns_non_array_type(self):
+ """Schema validation warns of non-array type for runcmd key.
+
+ Schema validation is not strict, so runcmd attempts to shellify the
+ invalid content.
+ """
+ invalid_config = {'runcmd': 1}
+ cc = self._get_cloud('ubuntu')
+ cc_runcmd.handle('cc_runcmd', invalid_config, cc, LOG, [])
+ self.assertIn(
+ 'Invalid config:\nruncmd: 1 is not of type \'array\'',
+ self.logs.getvalue())
+ self.assertIn('Failed to shellify', self.logs.getvalue())
+
+ @skipIf(_missing_jsonschema_dep, 'No python-jsonschema dependency')
+ def test_handler_schema_validation_warns_non_array_item_type(self):
+ """Schema validation warns of non-array or string runcmd items.
+
+ Schema validation is not strict, so runcmd attempts to shellify the
+ invalid content.
+ """
+ invalid_config = {
+ 'runcmd': ['ls /', 20, ['wget', 'http://stuff/blah'], {'a': 'n'}]}
+ cc = self._get_cloud('ubuntu')
+ cc_runcmd.handle('cc_runcmd', invalid_config, cc, LOG, [])
+ expected_warnings = [
+ 'runcmd.1: 20 is not valid under any of the given schemas',
+ 'runcmd.3: {\'a\': \'n\'} is not valid under any of the given'
+ ' schema'
+ ]
+ logs = self.logs.getvalue()
+ for warning in expected_warnings:
+ self.assertIn(warning, logs)
+ self.assertIn('Failed to shellify', logs)
+
+ def test_handler_write_valid_runcmd_schema_to_file(self):
+ """Valid runcmd schema is written to a runcmd shell script."""
+ valid_config = {'runcmd': [['ls', '/']]}
+ cc = self._get_cloud('ubuntu')
+ cc_runcmd.handle('cc_runcmd', valid_config, cc, LOG, [])
+ runcmd_file = os.path.join(
+ self.new_root,
+ 'var/lib/cloud/instances/iid-datasource-none/scripts/runcmd')
+ self.assertEqual("#!/bin/sh\n'ls' '/'\n", util.load_file(runcmd_file))
+ file_stat = os.stat(runcmd_file)
+ self.assertEqual(0o700, stat.S_IMODE(file_stat.st_mode))
+
+
+# vi: ts=4 expandtab
diff --git a/tests/unittests/test_handler/test_schema.py b/tests/unittests/test_handler/test_schema.py
index eda4802a..640f11d4 100644
--- a/tests/unittests/test_handler/test_schema.py
+++ b/tests/unittests/test_handler/test_schema.py
@@ -1,9 +1,9 @@
# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit.config.schema import (
- CLOUD_CONFIG_HEADER, SchemaValidationError, get_schema_doc,
- validate_cloudconfig_file, validate_cloudconfig_schema,
- main)
+ 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 ..helpers import CiTestCase, mock, skipIf
@@ -11,6 +11,7 @@ from ..helpers import CiTestCase, mock, skipIf
from copy import copy
from six import StringIO
from textwrap import dedent
+from yaml import safe_load
try:
import jsonschema
@@ -20,6 +21,29 @@ except ImportError:
_missing_jsonschema_dep = True
+class GetSchemaTest(CiTestCase):
+
+ def test_get_schema_coalesces_known_schema(self):
+ """Every cloudconfig module with schema is listed in allOf keyword."""
+ schema = get_schema()
+ self.assertItemsEqual(
+ ['cc_ntp', 'cc_runcmd'],
+ [subschema['id'] for subschema in schema['allOf']])
+ self.assertEqual('cloud-config-schema', schema['id'])
+ self.assertEqual(
+ 'http://json-schema.org/draft-04/schema#',
+ schema['$schema'])
+ # FULL_SCHEMA is updated by the get_schema call
+ from cloudinit.config.schema import FULL_SCHEMA
+ self.assertItemsEqual(['id', '$schema', 'allOf'], FULL_SCHEMA.keys())
+
+ def test_get_schema_returns_global_when_set(self):
+ """When FULL_SCHEMA global is already set, get_schema returns it."""
+ m_schema_path = 'cloudinit.config.schema.FULL_SCHEMA'
+ with mock.patch(m_schema_path, {'here': 'iam'}):
+ self.assertEqual({'here': 'iam'}, get_schema())
+
+
class SchemaValidationErrorTest(CiTestCase):
"""Test validate_cloudconfig_schema"""
@@ -151,11 +175,11 @@ class GetSchemaDocTest(CiTestCase):
full_schema.update(
{'properties': {
'prop1': {'type': 'array', 'description': 'prop-description',
- 'items': {'type': 'int'}}}})
+ 'items': {'type': 'integer'}}}})
self.assertEqual(
dedent("""
name
- ---
+ ----
**Summary:** title
description
@@ -167,27 +191,71 @@ class GetSchemaDocTest(CiTestCase):
**Supported distros:** debian, rhel
**Config schema**:
- **prop1:** (array of int) prop-description\n\n"""),
+ **prop1:** (array of integer) prop-description\n\n"""),
+ get_schema_doc(full_schema))
+
+ def test_get_schema_doc_handles_multiple_types(self):
+ """get_schema_doc delimits multiple property types with a '/'."""
+ full_schema = copy(self.required_schema)
+ full_schema.update(
+ {'properties': {
+ 'prop1': {'type': ['string', 'integer'],
+ 'description': 'prop-description'}}})
+ self.assertIn(
+ '**prop1:** (string/integer) prop-description',
+ get_schema_doc(full_schema))
+
+ def test_get_schema_doc_handles_nested_oneof_property_types(self):
+ """get_schema_doc describes array items oneOf declarations in type."""
+ full_schema = copy(self.required_schema)
+ full_schema.update(
+ {'properties': {
+ 'prop1': {'type': 'array',
+ 'items': {
+ 'oneOf': [{'type': 'string'},
+ {'type': 'integer'}]},
+ 'description': 'prop-description'}}})
+ self.assertIn(
+ '**prop1:** (array of (string)/(integer)) prop-description',
get_schema_doc(full_schema))
def test_get_schema_doc_returns_restructured_text_with_examples(self):
"""get_schema_doc returns indented examples when present in schema."""
full_schema = copy(self.required_schema)
full_schema.update(
- {'examples': {'ex1': [1, 2, 3]},
+ {'examples': [{'ex1': [1, 2, 3]}],
'properties': {
'prop1': {'type': 'array', 'description': 'prop-description',
- 'items': {'type': 'int'}}}})
+ 'items': {'type': 'integer'}}}})
self.assertIn(
dedent("""
**Config schema**:
- **prop1:** (array of int) prop-description
+ **prop1:** (array of integer) prop-description
**Examples**::
ex1"""),
get_schema_doc(full_schema))
+ def test_get_schema_doc_handles_unstructured_examples(self):
+ """get_schema_doc properly indented examples which as just strings."""
+ full_schema = copy(self.required_schema)
+ full_schema.update(
+ {'examples': ['My example:\n [don\'t, expand, "this"]'],
+ 'properties': {
+ 'prop1': {'type': 'array', 'description': 'prop-description',
+ 'items': {'type': 'integer'}}}})
+ self.assertIn(
+ dedent("""
+ **Config schema**:
+ **prop1:** (array of integer) prop-description
+
+ **Examples**::
+
+ My example:
+ [don't, expand, "this"]"""),
+ get_schema_doc(full_schema))
+
def test_get_schema_doc_raises_key_errors(self):
"""get_schema_doc raises KeyErrors on missing keys."""
for key in self.required_schema:
@@ -198,13 +266,78 @@ class GetSchemaDocTest(CiTestCase):
self.assertIn(key, str(context_mgr.exception))
+class AnnotatedCloudconfigFileTest(CiTestCase):
+ maxDiff = None
+
+ def test_annotated_cloudconfig_file_no_schema_errors(self):
+ """With no schema_errors, print the original content."""
+ content = b'ntp:\n pools: [ntp1.pools.com]\n'
+ self.assertEqual(
+ content,
+ annotated_cloudconfig_file({}, content, schema_errors=[]))
+
+ def test_annotated_cloudconfig_file_schema_annotates_and_adds_footer(self):
+ """With schema_errors, error lines are annotated and a footer added."""
+ content = dedent("""\
+ #cloud-config
+ # comment
+ ntp:
+ pools: [-99, 75]
+ """).encode()
+ expected = dedent("""\
+ #cloud-config
+ # comment
+ ntp: # E1
+ pools: [-99, 75] # E2,E3
+
+ # Errors: -------------
+ # E1: Some type error
+ # E2: -99 is not a string
+ # E3: 75 is not a string
+
+ """)
+ parsed_config = safe_load(content[13:])
+ schema_errors = [
+ ('ntp', 'Some type error'), ('ntp.pools.0', '-99 is not a string'),
+ ('ntp.pools.1', '75 is not a string')]
+ self.assertEqual(
+ expected,
+ annotated_cloudconfig_file(parsed_config, content, schema_errors))
+
+ def test_annotated_cloudconfig_file_annotates_separate_line_items(self):
+ """Errors are annotated for lists with items on separate lines."""
+ content = dedent("""\
+ #cloud-config
+ # comment
+ ntp:
+ pools:
+ - -99
+ - 75
+ """).encode()
+ expected = dedent("""\
+ ntp:
+ pools:
+ - -99 # E1
+ - 75 # E2
+ """)
+ parsed_config = safe_load(content[13:])
+ schema_errors = [
+ ('ntp.pools.0', '-99 is not a string'),
+ ('ntp.pools.1', '75 is not a string')]
+ self.assertIn(
+ expected,
+ annotated_cloudconfig_file(parsed_config, content, schema_errors))
+
+
class MainTest(CiTestCase):
def test_main_missing_args(self):
"""Main exits non-zero and reports an error on missing parameters."""
with mock.patch('sys.argv', ['mycmd']):
with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr:
- self.assertEqual(1, main(), 'Expected non-zero exit code')
+ with self.assertRaises(SystemExit) as context_manager:
+ main()
+ self.assertEqual('1', str(context_manager.exception))
self.assertEqual(
'Expected either --config-file argument or --doc\n',
m_stderr.getvalue())
@@ -216,13 +349,13 @@ class MainTest(CiTestCase):
with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout:
self.assertEqual(0, main(), 'Expected 0 exit code')
self.assertIn('\nNTP\n---\n', m_stdout.getvalue())
+ self.assertIn('\nRuncmd\n------\n', m_stdout.getvalue())
def test_main_validates_config_file(self):
"""When --config-file parameter is provided, main validates schema."""
myyaml = self.tmp_path('my.yaml')
myargs = ['mycmd', '--config-file', myyaml]
- with open(myyaml, 'wb') as stream:
- stream.write(b'#cloud-config\nntp:') # shortest ntp schema
+ write_file(myyaml, b'#cloud-config\nntp:') # shortest ntp schema
with mock.patch('sys.argv', myargs):
with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout:
self.assertEqual(0, main(), 'Expected 0 exit code')
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-23 20:18:10 +0000