diff options
| -rw-r--r-- | cloudinit/config/schema.py | 41 | ||||
| -rw-r--r-- | doc/rtd/topics/faq.rst | 6 | ||||
| -rw-r--r-- | tests/unittests/test_cli.py | 2 | ||||
| -rw-r--r-- | tests/unittests/test_handler/test_schema.py | 109 |
4 files changed, 114 insertions, 44 deletions
diff --git a/cloudinit/config/schema.py b/cloudinit/config/schema.py index 8a966aee..456bab2c 100644 --- a/cloudinit/config/schema.py +++ b/cloudinit/config/schema.py @@ -1,6 +1,7 @@ # This file is part of cloud-init. See LICENSE file for license information. """schema.py: Set of module functions for processing cloud-config schema.""" +from cloudinit.cmd.devel import read_cfg_paths from cloudinit import importer from cloudinit.util import find_modules, load_file @@ -173,7 +174,8 @@ def annotated_cloudconfig_file(cloudconfig, original_content, schema_errors): def validate_cloudconfig_file(config_path, schema, annotate=False): """Validate cloudconfig file adheres to a specific jsonschema. - @param config_path: Path to the yaml cloud-config file to parse. + @param config_path: Path to the yaml cloud-config file to parse, or None + to default to system userdata from Paths object. @param schema: Dict describing a valid jsonschema to validate against. @param annotate: Boolean set True to print original config file with error annotations on the offending lines. @@ -181,9 +183,24 @@ def validate_cloudconfig_file(config_path, schema, annotate=False): @raises SchemaValidationError containing any of schema_errors encountered. @raises RuntimeError when config_path does not exist. """ - if not os.path.exists(config_path): - raise RuntimeError('Configfile {0} does not exist'.format(config_path)) - content = load_file(config_path, decode=False) + if config_path is None: + # Use system's raw userdata path + if os.getuid() != 0: + raise RuntimeError( + "Unable to read system userdata as non-root user." + " Try using sudo" + ) + paths = read_cfg_paths() + user_data_file = paths.get_ipath_cur("userdata_raw") + content = load_file(user_data_file, decode=False) + else: + if not os.path.exists(config_path): + raise RuntimeError( + 'Configfile {0} does not exist'.format( + config_path + ) + ) + content = load_file(config_path, decode=False) if not content.startswith(CLOUD_CONFIG_HEADER): errors = ( ('format-l1.c1', 'File {0} needs to begin with "{1}"'.format( @@ -425,6 +442,8 @@ def get_parser(parser=None): description='Validate cloud-config files or document schema') parser.add_argument('-c', '--config-file', help='Path of the cloud-config yaml file to validate') + parser.add_argument('--system', action='store_true', default=False, + help='Validate the system cloud-config userdata') parser.add_argument('-d', '--docs', nargs='+', help=('Print schema module docs. Choices: all or' ' space-delimited cc_names.')) @@ -435,11 +454,11 @@ def get_parser(parser=None): def handle_schema_args(name, args): """Handle provided schema args and perform the appropriate actions.""" - exclusive_args = [args.config_file, args.docs] - if not any(exclusive_args) or all(exclusive_args): - error('Expected either --config-file argument or --docs') + exclusive_args = [args.config_file, args.docs, args.system] + if len([arg for arg in exclusive_args if arg]) != 1: + error('Expected one of --config-file, --system or --docs arguments') full_schema = get_schema() - if args.config_file: + if args.config_file or args.system: try: validate_cloudconfig_file( args.config_file, full_schema, args.annotate) @@ -449,7 +468,11 @@ def handle_schema_args(name, args): except RuntimeError as e: error(str(e)) else: - print("Valid cloud-config file {0}".format(args.config_file)) + if args.config_file is None: + cfg_name = "system userdata" + else: + cfg_name = args.config_file + print("Valid cloud-config:", cfg_name) elif args.docs: schema_ids = [subschema['id'] for subschema in full_schema['allOf']] schema_ids += ['all'] diff --git a/doc/rtd/topics/faq.rst b/doc/rtd/topics/faq.rst index d08914b5..27fabf15 100644 --- a/doc/rtd/topics/faq.rst +++ b/doc/rtd/topics/faq.rst @@ -141,12 +141,12 @@ that can validate your user data offline. .. _validate-yaml.py: https://github.com/canonical/cloud-init/blob/master/tools/validate-yaml.py -Another option is to run the following on an instance when debugging: +Another option is to run the following on an instance to debug userdata +provided to the system: .. code-block:: shell-session - $ sudo cloud-init query userdata > user-data.yaml - $ cloud-init devel schema -c user-data.yaml --annotate + $ cloud-init devel schema --system --annotate As launching instances in the cloud can cost money and take a bit longer, sometimes it is easier to launch instances locally using Multipass or LXD: diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py index dcf0fe5a..74f85959 100644 --- a/tests/unittests/test_cli.py +++ b/tests/unittests/test_cli.py @@ -214,7 +214,7 @@ class TestCLI(test_helpers.FilesystemMockingTestCase): self.assertEqual(1, exit_code) # Known whitebox output from schema subcommand self.assertEqual( - 'Expected either --config-file argument or --docs\n', + 'Expected one of --config-file, --system or --docs arguments\n', self.stderr.getvalue()) def test_wb_devel_schema_subcommand_doc_content(self): diff --git a/tests/unittests/test_handler/test_schema.py b/tests/unittests/test_handler/test_schema.py index 44292571..15aa77bb 100644 --- a/tests/unittests/test_handler/test_schema.py +++ b/tests/unittests/test_handler/test_schema.py @@ -9,9 +9,9 @@ from cloudinit.util import write_file from cloudinit.tests.helpers import CiTestCase, mock, skipUnlessJsonSchema from copy import copy +import itertools import os import pytest -from io import StringIO from pathlib import Path from textwrap import dedent from yaml import safe_load @@ -400,50 +400,97 @@ class AnnotatedCloudconfigFileTest(CiTestCase): annotated_cloudconfig_file(parsed_config, content, schema_errors)) -class MainTest(CiTestCase): +class TestMain: - def test_main_missing_args(self): + exclusive_combinations = itertools.combinations( + ["--system", "--docs all", "--config-file something"], 2 + ) + + @pytest.mark.parametrize("params", exclusive_combinations) + def test_main_exclusive_args(self, params, capsys): + """Main exits non-zero and error on required exclusive args.""" + params = list(itertools.chain(*[a.split() for a in params])) + with mock.patch('sys.argv', ['mycmd'] + params): + with pytest.raises(SystemExit) as context_manager: + main() + assert 1 == context_manager.value.code + + _out, err = capsys.readouterr() + expected = ( + 'Expected one of --config-file, --system or --docs arguments\n' + ) + assert expected == err + + def test_main_missing_args(self, capsys): """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: - with self.assertRaises(SystemExit) as context_manager: - main() - self.assertEqual(1, context_manager.exception.code) - self.assertEqual( - 'Expected either --config-file argument or --docs\n', - m_stderr.getvalue()) + with pytest.raises(SystemExit) as context_manager: + main() + assert 1 == context_manager.value.code + + _out, err = capsys.readouterr() + expected = ( + 'Expected one of --config-file, --system or --docs arguments\n' + ) + assert expected == err - def test_main_absent_config_file(self): + def test_main_absent_config_file(self, capsys): """Main exits non-zero when config file is absent.""" myargs = ['mycmd', '--annotate', '--config-file', 'NOT_A_FILE'] with mock.patch('sys.argv', myargs): - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - with self.assertRaises(SystemExit) as context_manager: - main() - self.assertEqual(1, context_manager.exception.code) - self.assertEqual( - 'Configfile NOT_A_FILE does not exist\n', - m_stderr.getvalue()) + with pytest.raises(SystemExit) as context_manager: + main() + assert 1 == context_manager.value.code + _out, err = capsys.readouterr() + assert 'Configfile NOT_A_FILE does not exist\n' == err - def test_main_prints_docs(self): + def test_main_prints_docs(self, capsys): """When --docs parameter is provided, main generates documentation.""" myargs = ['mycmd', '--docs', 'all'] 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') - self.assertIn('\nNTP\n---\n', m_stdout.getvalue()) - self.assertIn('\nRuncmd\n------\n', m_stdout.getvalue()) + assert 0 == main(), 'Expected 0 exit code' + out, _err = capsys.readouterr() + assert '\nNTP\n---\n' in out + assert '\nRuncmd\n------\n' in out - def test_main_validates_config_file(self): + def test_main_validates_config_file(self, tmpdir, capsys): """When --config-file parameter is provided, main validates schema.""" - myyaml = self.tmp_path('my.yaml') - myargs = ['mycmd', '--config-file', myyaml] - write_file(myyaml, b'#cloud-config\nntp:') # shortest ntp schema + myyaml = tmpdir.join('my.yaml') + myargs = ['mycmd', '--config-file', myyaml.strpath] + myyaml.write(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') - self.assertIn( - 'Valid cloud-config file {0}'.format(myyaml), m_stdout.getvalue()) + assert 0 == main(), 'Expected 0 exit code' + out, _err = capsys.readouterr() + assert 'Valid cloud-config: {0}\n'.format(myyaml) == out + + @mock.patch('cloudinit.config.schema.read_cfg_paths') + @mock.patch('cloudinit.config.schema.os.getuid', return_value=0) + def test_main_validates_system_userdata( + self, m_getuid, m_read_cfg_paths, capsys, paths + ): + """When --system is provided, main validates system userdata.""" + m_read_cfg_paths.return_value = paths + ud_file = paths.get_ipath_cur("userdata_raw") + write_file(ud_file, b'#cloud-config\nntp:') + myargs = ['mycmd', '--system'] + with mock.patch('sys.argv', myargs): + assert 0 == main(), 'Expected 0 exit code' + out, _err = capsys.readouterr() + assert 'Valid cloud-config: system userdata\n' == out + + @mock.patch('cloudinit.config.schema.os.getuid', return_value=1000) + def test_main_system_userdata_requires_root(self, m_getuid, capsys, paths): + """Non-root user can't use --system param""" + myargs = ['mycmd', '--system'] + with mock.patch('sys.argv', myargs): + with pytest.raises(SystemExit) as context_manager: + main() + assert 1 == context_manager.value.code + _out, err = capsys.readouterr() + expected = ( + 'Unable to read system userdata as non-root user. Try using sudo\n' + ) + assert expected == err class CloudTestsIntegrationTest(CiTestCase): |