1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
|
# 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)
from cloudinit.util import write_file
from ..helpers import CiTestCase, mock
from copy import copy
from six import StringIO
from textwrap import dedent
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
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())
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())
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))
def test_validateconfig_schema_honors_formats(self):
"""When strict is True validate_cloudconfig_schema raises errors."""
schema = {
'properties': {'p1': {'type': 'string', 'format': 'hostname'}}}
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 'hostname'",
str(context_mgr.exception))
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: header: 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_format(self):
"""On non-yaml format, 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: File {0} is not valid yaml.'.format(
self.config_file),
str(context_mgr.exception))
def test_validateconfig_file_sctricty_validates_schema(self):
"""validate_cloudconfig_file raises errors on invalid schema."""
schema = {
'properties': {'p1': {'type': 'string', 'format': 'hostname'}}}
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 a 'hostname'",
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': 'int'}}}})
self.assertEqual(
dedent("""
name
---
**Summary:** title
description
**Internal name:** ``id``
**Module frequency:** frequency
**Supported distros:** debian, rhel
**Config schema**:
**prop1:** (array of int) prop-description\n\n"""),
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]},
'properties': {
'prop1': {'type': 'array', 'description': 'prop-description',
'items': {'type': 'int'}}}})
self.assertIn(
dedent("""
**Config schema**:
**prop1:** (array of int) prop-description
**Examples**::
ex1"""),
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.assertEqual("'{0}'".format(key), str(context_mgr.exception))
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')
self.assertEqual(
'Expected either --config-file argument or --doc\n',
m_stderr.getvalue())
def test_main_prints_docs(self):
"""When --doc parameter is provided, main generates documentation."""
myargs = ['mycmd', '--doc']
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())
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
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())
# vi: ts=4 expandtab syntax=python
|
