summaryrefslogtreecommitdiff
path: root/tests/unittests/test_handler/test_schema.py
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_handler/test_schema.py
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_handler/test_schema.py')
-rw-r--r--tests/unittests/test_handler/test_schema.py515
1 files changed, 0 insertions, 515 deletions
diff --git a/tests/unittests/test_handler/test_schema.py b/tests/unittests/test_handler/test_schema.py
deleted file mode 100644
index 1dae223d..00000000
--- a/tests/unittests/test_handler/test_schema.py
+++ /dev/null
@@ -1,515 +0,0 @@
-# This file is part of cloud-init. See LICENSE file for license information.
-import cloudinit
-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.tests.helpers import CiTestCase, mock, skipUnlessJsonSchema
-
-from copy import copy
-import itertools
-import pytest
-from pathlib import Path
-from textwrap import dedent
-from yaml import safe_load
-
-
-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.assertCountEqual(
- [
- 'cc_apk_configure',
- 'cc_apt_configure',
- 'cc_bootcmd',
- 'cc_locale',
- 'cc_ntp',
- 'cc_resizefs',
- 'cc_runcmd',
- 'cc_snap',
- 'cc_ubuntu_advantage',
- 'cc_ubuntu_drivers',
- 'cc_write_files',
- 'cc_write_files_deferred',
- 'cc_zypper_add_repo',
- 'cc_chef',
- 'cc_install_hotplug',
- ],
- [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.assertCountEqual(['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"""
-
- def test_schema_validation_error_expects_schema_errors(self):
- """SchemaValidationError is initialized from schema_errors."""
- errors = (('key.path', 'unexpected key "junk"'),
- ('key2.path', '"-123" is not a valid "hostname" format'))
- exception = SchemaValidationError(schema_errors=errors)
- self.assertIsInstance(exception, Exception)
- self.assertEqual(exception.schema_errors, errors)
- self.assertEqual(
- 'Cloud config schema errors: key.path: unexpected key "junk", '
- 'key2.path: "-123" is not a valid "hostname" format',
- str(exception))
- self.assertTrue(isinstance(exception, ValueError))
-
-
-class ValidateCloudConfigSchemaTest(CiTestCase):
- """Tests for validate_cloudconfig_schema."""
-
- with_logs = True
-
- @skipUnlessJsonSchema()
- def test_validateconfig_schema_non_strict_emits_warnings(self):
- """When strict is False validate_cloudconfig_schema emits warnings."""
- schema = {'properties': {'p1': {'type': 'string'}}}
- validate_cloudconfig_schema({'p1': -1}, schema, strict=False)
- self.assertIn(
- "Invalid config:\np1: -1 is not of type 'string'\n",
- self.logs.getvalue())
-
- @skipUnlessJsonSchema()
- def test_validateconfig_schema_emits_warning_on_missing_jsonschema(self):
- """Warning from validate_cloudconfig_schema when missing jsonschema."""
- schema = {'properties': {'p1': {'type': 'string'}}}
- with mock.patch.dict('sys.modules', **{'jsonschema': ImportError()}):
- validate_cloudconfig_schema({'p1': -1}, schema, strict=True)
- self.assertIn(
- 'Ignoring schema validation. python-jsonschema is not present',
- self.logs.getvalue())
-
- @skipUnlessJsonSchema()
- def test_validateconfig_schema_strict_raises_errors(self):
- """When strict is True validate_cloudconfig_schema raises errors."""
- schema = {'properties': {'p1': {'type': 'string'}}}
- with self.assertRaises(SchemaValidationError) as context_mgr:
- validate_cloudconfig_schema({'p1': -1}, schema, strict=True)
- self.assertEqual(
- "Cloud config schema errors: p1: -1 is not of type 'string'",
- str(context_mgr.exception))
-
- @skipUnlessJsonSchema()
- def test_validateconfig_schema_honors_formats(self):
- """With strict True, validate_cloudconfig_schema errors on format."""
- schema = {
- 'properties': {'p1': {'type': 'string', 'format': 'email'}}}
- with self.assertRaises(SchemaValidationError) as context_mgr:
- validate_cloudconfig_schema({'p1': '-1'}, schema, strict=True)
- self.assertEqual(
- "Cloud config schema errors: p1: '-1' is not a 'email'",
- str(context_mgr.exception))
-
-
-class TestCloudConfigExamples:
- schema = get_schema()
- params = [
- (schema["id"], example)
- for schema in schema["allOf"] for example in schema["examples"]]
-
- @pytest.mark.parametrize("schema_id,example", params)
- @skipUnlessJsonSchema()
- def test_validateconfig_schema_of_example(self, schema_id, example):
- """ For a given example in a config module we test if it is valid
- according to the unified schema of all config modules
- """
- config_load = safe_load(example)
- validate_cloudconfig_schema(
- config_load, self.schema, strict=True)
-
-
-class ValidateCloudConfigFileTest(CiTestCase):
- """Tests for validate_cloudconfig_file."""
-
- def setUp(self):
- super(ValidateCloudConfigFileTest, self).setUp()
- self.config_file = self.tmp_path('cloudcfg.yaml')
-
- def test_validateconfig_file_error_on_absent_file(self):
- """On absent config_path, validate_cloudconfig_file errors."""
- with self.assertRaises(RuntimeError) as context_mgr:
- validate_cloudconfig_file('/not/here', {})
- self.assertEqual(
- 'Configfile /not/here does not exist',
- str(context_mgr.exception))
-
- def test_validateconfig_file_error_on_invalid_header(self):
- """On invalid header, validate_cloudconfig_file errors.
-
- A SchemaValidationError is raised when the file doesn't begin with
- CLOUD_CONFIG_HEADER.
- """
- write_file(self.config_file, '#junk')
- with self.assertRaises(SchemaValidationError) as context_mgr:
- validate_cloudconfig_file(self.config_file, {})
- self.assertEqual(
- 'Cloud config schema errors: format-l1.c1: File {0} needs to begin'
- ' with "{1}"'.format(
- self.config_file, CLOUD_CONFIG_HEADER.decode()),
- str(context_mgr.exception))
-
- def test_validateconfig_file_error_on_non_yaml_scanner_error(self):
- """On non-yaml scan issues, validate_cloudconfig_file errors."""
- # Generate a scanner error by providing text on a single line with
- # improper indent.
- write_file(self.config_file, '#cloud-config\nasdf:\nasdf')
- with self.assertRaises(SchemaValidationError) as context_mgr:
- validate_cloudconfig_file(self.config_file, {})
- self.assertIn(
- 'schema errors: format-l3.c1: File {0} is not valid yaml.'.format(
- self.config_file),
- str(context_mgr.exception))
-
- def test_validateconfig_file_error_on_non_yaml_parser_error(self):
- """On non-yaml parser issues, validate_cloudconfig_file errors."""
- write_file(self.config_file, '#cloud-config\n{}}')
- with self.assertRaises(SchemaValidationError) as context_mgr:
- validate_cloudconfig_file(self.config_file, {})
- self.assertIn(
- 'schema errors: format-l2.c3: File {0} is not valid yaml.'.format(
- self.config_file),
- str(context_mgr.exception))
-
- @skipUnlessJsonSchema()
- def test_validateconfig_file_sctrictly_validates_schema(self):
- """validate_cloudconfig_file raises errors on invalid schema."""
- schema = {
- 'properties': {'p1': {'type': 'string', 'format': 'string'}}}
- write_file(self.config_file, '#cloud-config\np1: -1')
- with self.assertRaises(SchemaValidationError) as context_mgr:
- validate_cloudconfig_file(self.config_file, schema)
- self.assertEqual(
- "Cloud config schema errors: p1: -1 is not of type 'string'",
- str(context_mgr.exception))
-
-
-class GetSchemaDocTest(CiTestCase):
- """Tests for get_schema_doc."""
-
- def setUp(self):
- super(GetSchemaDocTest, self).setUp()
- self.required_schema = {
- 'title': 'title', 'description': 'description', 'id': 'id',
- 'name': 'name', 'frequency': 'frequency',
- 'distros': ['debian', 'rhel']}
-
- def test_get_schema_doc_returns_restructured_text(self):
- """get_schema_doc returns restructured text for a cloudinit schema."""
- full_schema = copy(self.required_schema)
- full_schema.update(
- {'properties': {
- 'prop1': {'type': 'array', 'description': 'prop-description',
- 'items': {'type': 'integer'}}}})
- self.assertEqual(
- dedent("""
- name
- ----
- **Summary:** title
-
- description
-
- **Internal name:** ``id``
-
- **Module frequency:** frequency
-
- **Supported distros:** debian, rhel
-
- **Config schema**:
- **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_enum_types(self):
- """get_schema_doc converts enum types to yaml and delimits with '/'."""
- full_schema = copy(self.required_schema)
- full_schema.update(
- {'properties': {
- 'prop1': {'enum': [True, False, 'stuff'],
- 'description': 'prop-description'}}})
- self.assertIn(
- '**prop1:** (true/false/stuff) 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_handles_string_examples(self):
- """get_schema_doc properly indented examples as a list of strings."""
- full_schema = copy(self.required_schema)
- full_schema.update(
- {'examples': ['ex1:\n [don\'t, expand, "this"]', 'ex2: true'],
- 'properties': {
- 'prop1': {'type': 'array', 'description': 'prop-description',
- 'items': {'type': 'integer'}}}})
- self.assertIn(
- dedent("""
- **Config schema**:
- **prop1:** (array of integer) prop-description
-
- **Examples**::
-
- ex1:
- [don't, expand, "this"]
- # --- Example2 ---
- ex2: true
- """),
- get_schema_doc(full_schema))
-
- def test_get_schema_doc_properly_parse_description(self):
- """get_schema_doc description properly formatted"""
- full_schema = copy(self.required_schema)
- full_schema.update(
- {'properties': {
- 'p1': {
- 'type': 'string',
- 'description': dedent("""\
- This item
- has the
- following options:
-
- - option1
- - option2
- - option3
-
- The default value is
- option1""")
- }
- }}
- )
-
- self.assertIn(
- dedent("""
- **Config schema**:
- **p1:** (string) This item has the following options:
-
- - option1
- - option2
- - option3
-
- The default value is option1
- """),
- 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:
- invalid_schema = copy(self.required_schema)
- invalid_schema.pop(key)
- with self.assertRaises(KeyError) as context_mgr:
- get_schema_doc(invalid_schema)
- 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 TestMain:
-
- 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 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, 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 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, capsys):
- """When --docs parameter is provided, main generates documentation."""
- myargs = ['mycmd', '--docs', 'all']
- with mock.patch('sys.argv', myargs):
- 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, tmpdir, capsys):
- """When --config-file parameter is provided, main validates 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):
- 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
-
-
-def _get_schema_doc_examples():
- examples_dir = Path(
- cloudinit.__file__).parent.parent / 'doc' / 'examples'
- assert examples_dir.is_dir()
-
- all_text_files = (f for f in examples_dir.glob('cloud-config*.txt')
- if not f.name.startswith('cloud-config-archive'))
- return all_text_files
-
-
-class TestSchemaDocExamples:
- schema = get_schema()
-
- @pytest.mark.parametrize("example_path", _get_schema_doc_examples())
- @skipUnlessJsonSchema()
- def test_schema_doc_examples(self, example_path):
- validate_cloudconfig_file(str(example_path), self.schema)
-
-# vi: ts=4 expandtab syntax=python