Add schema-based config data validation

This patch adds a jsonschema for os-net-config's configuration data
and a library function to validate configuration data based on this
schema. Adding schema-based validation allows catching a larger class of
errors (typos, missing required parameters, etc.) for all devices
configurable through os-net-config.

The validation is run in the os-net-config CLI after loading the config
file. If the config file fails to validate, the current default is to
just log a warning and try to continue. By providing the new CLI option
'--exit-on-validation-errors', this can be changed to log an error and
exist instead.

This validation is meant to be reusable, for example for pre-deployment
validation of network environments (see change
Ic16ee0bc353c46f8fe512454176a07ee95347346). Packaging with os-net-config
makes it easier to keep object model and schema in sync.

Change-Id: Ie4a905863b2d46c88d9cd6c3afc50e7d0a877090
Signed-off-by: Frank A. Zdarsky <fzdarsky@redhat.com>
This commit is contained in:
Frank A. Zdarsky 2017-01-20 12:53:52 +01:00
parent 03d3a5dea1
commit 9ef9a7a92d
7 changed files with 1632 additions and 12 deletions

View File

@ -25,6 +25,7 @@ from os_net_config import impl_eni
from os_net_config import impl_ifcfg
from os_net_config import impl_iproute
from os_net_config import objects
from os_net_config import validator
from os_net_config import version
@ -55,6 +56,14 @@ def parse_opts(argv):
"""that files were modified."""
"""Disabled by default.""",
default=False)
parser.add_argument(
'--exit-on-validation-errors',
action='store_true',
help="Exit with an error if configuration file validation fails. "
"Without this option, just log a warning and continue.",
default=False)
parser.add_argument(
'-d', '--debug',
dest="debug",
@ -181,6 +190,16 @@ def main(argv=sys.argv):
for iface_json in iface_array:
iface_json.update({'nic_mapping': iface_mapping})
iface_json.update({'persist_mapping': persist_mapping})
validation_errors = validator.validate_config(iface_array)
if validation_errors:
if opts.exit_on_validation_errors:
logger.error('\n'.join(validation_errors))
return 1
else:
logger.warning('\n'.join(validation_errors))
for iface_json in iface_array:
obj = objects.object_from_json(iface_json)
provider.add_object(obj)
files_changed = provider.apply(cleanup=opts.cleanup,

View File

@ -14,6 +14,11 @@
# License for the specific language governing permissions and limitations
# under the License.
#
# NOTE: When making changes to the object model, remember to also update
# schema.yaml to reflect changes to the schema of config files!
#
import logging
import netaddr
from oslo_utils import strutils

1003
os_net_config/schema.yaml Normal file

File diff suppressed because it is too large Load Diff

View File

@ -52,9 +52,11 @@ class TestCli(base.TestCase):
bond_yaml = os.path.join(SAMPLE_BASE, 'bond.yaml')
bond_json = os.path.join(SAMPLE_BASE, 'bond.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % bond_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % bond_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=br-ctlplane',
@ -70,9 +72,11 @@ class TestCli(base.TestCase):
ivs_yaml = os.path.join(SAMPLE_BASE, 'ivs.yaml')
ivs_json = os.path.join(SAMPLE_BASE, 'ivs.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % ivs_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % ivs_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=nic2',
@ -87,11 +91,13 @@ class TestCli(base.TestCase):
def test_bridge_noop_output(self):
bridge_yaml = os.path.join(SAMPLE_BASE, 'bridge_dhcp.yaml')
bridge_json = os.path.join(SAMPLE_BASE, 'bridge_dhcp.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=eni --noop -c %s' %
bridge_yaml)
stdout_yaml, stderr = self.run_cli('ARG0 --provider=eni --noop '
'--exit-on-validation-errors '
'-c %s' % bridge_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=eni --noop -c %s' %
bridge_json)
stdout_json, stderr = self.run_cli('ARG0 --provider=eni --noop '
'--exit-on-validation-errors '
'-c %s' % bridge_json)
self.assertEqual('', stderr)
sanity_devices = ['iface br-ctlplane inet dhcp',
'iface em1',
@ -103,11 +109,13 @@ class TestCli(base.TestCase):
def test_vlan_noop_output(self):
vlan_yaml = os.path.join(SAMPLE_BASE, 'bridge_vlan.yaml')
vlan_json = os.path.join(SAMPLE_BASE, 'bridge_vlan.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop -c %s'
% vlan_yaml)
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % vlan_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop -c %s'
% vlan_json)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % vlan_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=br-ctlplane',
'DEVICE=em1',
@ -120,11 +128,13 @@ class TestCli(base.TestCase):
def test_interface_noop_output(self):
interface_yaml = os.path.join(SAMPLE_BASE, 'interface.yaml')
interface_json = os.path.join(SAMPLE_BASE, 'interface.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop -c %s'
% interface_yaml)
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % interface_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop -c %s'
% interface_json)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % interface_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=em1',
'BOOTPROTO=static',
@ -137,6 +147,7 @@ class TestCli(base.TestCase):
for provider in ('ifcfg', 'eni'):
bond_yaml = os.path.join(SAMPLE_BASE, 'bridge_dhcp.yaml')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=%s --noop '
'--exit-on-validation-errors '
'--root-dir=/rootfs '
'-c %s' % (provider, bond_yaml))
self.assertEqual('', stderr)
@ -145,6 +156,7 @@ class TestCli(base.TestCase):
def test_interface_noop_detailed_exit_codes(self):
interface_yaml = os.path.join(SAMPLE_BASE, 'interface.yaml')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s --detailed-exit-codes'
% interface_yaml, exitcodes=(2,))
@ -162,6 +174,7 @@ class TestCli(base.TestCase):
self.stubs.Set(impl_ifcfg, 'IfcfgNetConfig', TestImpl)
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s --detailed-exit-codes'
% interface_yaml, exitcodes=(0,))
@ -169,9 +182,11 @@ class TestCli(base.TestCase):
ivs_yaml = os.path.join(SAMPLE_BASE, 'ovs_dpdk_bond.yaml')
ivs_json = os.path.join(SAMPLE_BASE, 'ovs_dpdk_bond.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % ivs_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % ivs_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=br-link',
@ -186,9 +201,11 @@ class TestCli(base.TestCase):
nfvswitch_yaml = os.path.join(SAMPLE_BASE, 'nfvswitch.yaml')
nfvswitch_json = os.path.join(SAMPLE_BASE, 'nfvswitch.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % nfvswitch_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % nfvswitch_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=nic2',
@ -204,9 +221,11 @@ class TestCli(base.TestCase):
ivs_yaml = os.path.join(SAMPLE_BASE, 'ovs_dpdk.yaml')
ivs_json = os.path.join(SAMPLE_BASE, 'ovs_dpdk.json')
stdout_yaml, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % ivs_yaml)
self.assertEqual('', stderr)
stdout_json, stderr = self.run_cli('ARG0 --provider=ifcfg --noop '
'--exit-on-validation-errors '
'-c %s' % ivs_json)
self.assertEqual('', stderr)
sanity_devices = ['DEVICE=br-link',

View File

@ -0,0 +1,392 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import glob
import jsonschema
import os.path
import yaml
from os_net_config.tests import base
from os_net_config import validator
REALPATH = os.path.dirname(os.path.realpath(__file__))
SAMPLE_BASE = os.path.join(REALPATH, '../../', 'etc',
'os-net-config', 'samples')
class TestSchemaValidation(base.TestCase):
def test_schema_is_valid(self):
schema = validator.get_os_net_config_schema()
jsonschema.Draft4Validator.check_schema(schema)
def test__validate_config(self):
schema = {"type": "string"}
errors = validator._validate_config(42, "foo", schema, False)
self.assertEqual(len(errors), 1)
errors = validator._validate_config("42", "foo", schema, False)
self.assertEqual(len(errors), 0)
def test_consistent_error_messages_type(self):
error = jsonschema.ValidationError(
"%r is not of type %r" % (u'name', u'string'), validator=u'type',
validator_value=u'string', instance=u'name')
msg = validator._get_consistent_error_message(error)
self.assertEqual(msg, "'name' is not of type 'string'")
def test_consistent_error_messages_oneOf(self):
error = jsonschema.ValidationError(
"%r is not one of %r" % (u'type', [u'vlan', u'interface']),
validator=u'enum', validator_value=[u'vlan', u'interface'],
instance=u'type')
msg = validator._get_consistent_error_message(error)
self.assertEqual(msg, "'type' is not one of ['vlan','interface']")
def test_consistent_error_messages_required(self):
error = jsonschema.ValidationError(
"%r is a required property" % u'name', validator=u'required')
msg = validator._get_consistent_error_message(error)
self.assertEqual(msg, "'name' is a required property")
error = jsonschema.ValidationError(
"u'name' is a required property", validator=u'required')
msg = validator._get_consistent_error_message(error)
self.assertEqual(msg, "'name' is a required property")
def test_pretty_print_schema_path(self):
schema = validator.get_os_net_config_schema()
path = ['items', 'oneOf', 0, 'properties', 'name']
path_string = validator._pretty_print_schema_path(path, schema)
self.assertEqual(path_string, "items/oneOf/interface/name")
def test_find_type_in_list_of_references(self):
schemas = [
{'$ref': '#/definitions/vlan'},
{'properties': {'type': 'interface'}},
None
]
result = validator._find_type_in_schema_list(schemas, 'vlan')
self.assertEqual(result, (True, 0))
result = validator._find_type_in_schema_list(schemas, 'interface')
self.assertEqual(result, (True, 1))
result = validator._find_type_in_schema_list(schemas, 'ovs_bridge')
self.assertEqual(result, (False, 0))
def test_missing_required_property(self):
ifaces = [{"type": "interface"}]
errors = validator.validate_config(ifaces)
self.assertEqual(len(errors), 1)
self.assertIn("'name' is a required property", errors[0])
class TestBaseTypes(base.TestCase):
def test_param(self):
schema = validator.get_schema_for_defined_type("bool_or_param")
v = jsonschema.Draft4Validator(schema)
self.assertTrue(v.is_valid({"get_param": "foo"}))
self.assertTrue(v.is_valid({"get_input": "bar"}))
self.assertFalse(v.is_valid([]))
self.assertFalse(v.is_valid({}))
self.assertFalse(v.is_valid(None))
self.assertFalse(v.is_valid("foo"))
def test_bool_or_param(self):
schema = validator.get_schema_for_defined_type("bool_or_param")
v = jsonschema.Draft4Validator(schema)
self.assertTrue(v.is_valid(True))
self.assertTrue(v.is_valid(False))
self.assertTrue(v.is_valid("TRUE"))
self.assertTrue(v.is_valid("true"))
self.assertTrue(v.is_valid("yes"))
self.assertTrue(v.is_valid("1"))
self.assertTrue(v.is_valid("on"))
self.assertTrue(v.is_valid("false"))
self.assertTrue(v.is_valid("FALSE"))
self.assertTrue(v.is_valid("off"))
self.assertTrue(v.is_valid("no"))
self.assertTrue(v.is_valid("0"))
self.assertFalse(v.is_valid([]))
self.assertFalse(v.is_valid({}))
self.assertFalse(v.is_valid(None))
self.assertFalse(v.is_valid("falsch"))
def test_ip_address_string(self):
schema = validator.get_schema_for_defined_type("ip_address_string")
v = jsonschema.Draft4Validator(schema)
self.assertTrue(v.is_valid("0.0.0.0"))
self.assertTrue(v.is_valid("192.168.0.1"))
self.assertTrue(v.is_valid("::"))
self.assertTrue(v.is_valid("fe80::"))
self.assertTrue(v.is_valid("1:1:1::"))
self.assertFalse(v.is_valid("192.168.0.1/24"))
def test_ip_cidr_string(self):
schema = validator.get_schema_for_defined_type("ip_cidr_string")
v = jsonschema.Draft4Validator(schema)
self.assertTrue(v.is_valid("0.0.0.0/0"))
self.assertTrue(v.is_valid("192.168.0.1/24"))
self.assertTrue(v.is_valid("::/0"))
self.assertTrue(v.is_valid("::1/128"))
self.assertTrue(v.is_valid("fe80::1/64"))
self.assertFalse(v.is_valid("193.168.0.1"))
class TestDerivedTypes(base.TestCase):
def test_address(self):
schema = validator.get_schema_for_defined_type("address")
v = jsonschema.Draft4Validator(schema)
data = {"ip_netmask": "127.0.0.1/32"}
self.assertTrue(v.is_valid(data))
data = {"ip_netmask": "127.0.0.1"}
self.assertFalse(v.is_valid(data))
data = {"ip_netmask": None}
self.assertFalse(v.is_valid(data))
data = {"ip_netmask": "127.0.0.1/32", "unkown_property": "value"}
self.assertFalse(v.is_valid(data))
self.assertFalse(v.is_valid([]))
self.assertFalse(v.is_valid(None))
def test_list_of_address(self):
schema = validator.get_schema_for_defined_type("list_of_address")
v = jsonschema.Draft4Validator(schema)
data = {"ip_netmask": "127.0.0.1/32"}
self.assertTrue(v.is_valid([data]))
self.assertFalse(v.is_valid(data))
self.assertFalse(v.is_valid([]))
self.assertFalse(v.is_valid(None))
def test_route(self):
schema = validator.get_schema_for_defined_type("route")
v = jsonschema.Draft4Validator(schema)
data = {"next_hop": "172.19.0.1", "ip_netmask": "172.19.0.0/24",
"default": True, "route_options": "metric 10"}
self.assertTrue(v.is_valid(data))
data["unkown_property"] = "value"
self.assertFalse(v.is_valid(data))
self.assertFalse(v.is_valid({}))
self.assertFalse(v.is_valid([]))
self.assertFalse(v.is_valid(None))
class TestDeviceTypes(base.TestCase):
def test_interface(self):
schema = validator.get_schema_for_defined_type("interface")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "interface",
"name": "em1",
"use_dhcp": False,
"addresses": [{
"ip_netmask": "192.0.2.1/24"
}],
"defroute": False,
"dhclient_args": "--foobar",
"dns_servers": ["1.2.3.4"],
"mtu": 1501,
"ethtool_opts": "speed 1000 duplex full",
"hotplug": True,
"routes": [{
"next_hop": "192.0.2.1",
"ip_netmask": "192.0.2.1/24",
"route_options": "metric 10"
}]
}
self.assertTrue(v.is_valid(data))
def test_vlan(self):
schema = validator.get_schema_for_defined_type("vlan")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "vlan",
"vlan_id": 101,
"addresses": [{
"ip_netmask": "192.0.2.1/24"
}]
}
self.assertTrue(v.is_valid(data))
def test_ovs_bridge(self):
schema = validator.get_schema_for_defined_type("ovs_bridge")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "ovs_bridge",
"name": "br-ctlplane",
"ovs_options": "lacp=active",
"ovs_extra": [
"br-set-external-id br-ctlplane bridge-id br-ctlplane",
"set bridge {name} stp_enable=true"
],
"ovs_fail_mode": "secure",
"members": [
{"type": "interface", "name": "em1"}
]
}
self.assertTrue(v.is_valid(data))
def test_ovs_bond(self):
schema = validator.get_schema_for_defined_type("ovs_bond")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "ovs_bond",
"name": "bond1",
"use_dhcp": "true",
"members": [
{"type": "interface", "name": "em1"},
{"type": "interface", "name": "em2"}
]
}
self.assertTrue(v.is_valid(data))
def test_ovs_user_bridge(self):
schema = validator.get_schema_for_defined_type("ovs_user_bridge")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "ovs_user_bridge",
"name": "br-link",
"members": [{
"type": "ovs_dpdk_bond",
"name": "dpdkbond0",
"mtu": 9000,
"rx_queue": 4,
"members": [{
"type": "ovs_dpdk_port",
"name": "dpdk0",
"members": [{
"type": "interface",
"name": "nic2"
}]
}, {
"type": "ovs_dpdk_port",
"name": "dpdk1",
"members": [{
"type": "interface",
"name": "nic3"
}]
}]
}]
}
self.assertTrue(v.is_valid(data))
def test_ovs_patch_port(self):
schema = validator.get_schema_for_defined_type("ovs_patch_port")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "ovs_patch_port",
"name": "br_pub-patch",
"bridge_name": "br-ctlplane",
"peer": "br-ctlplane-patch"
}
self.assertTrue(v.is_valid(data))
def test_ovs_tunnel(self):
schema = validator.get_schema_for_defined_type("ovs_tunnel")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "ovs_tunnel",
"name": "tun0",
"tunnel_type": "vxlan",
"ovs_options": ["lacp=active"]
}
self.assertTrue(v.is_valid(data))
def test_vpp_interface(self):
schema = validator.get_schema_for_defined_type("vpp_interface")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "vpp_interface",
"name": "nic2",
"addresses": [
{"ip_netmask": "192.0.2.1/24"}
],
"uio_driver": "uio_pci_generic",
"options": "vlan-strip-offload off"
}
self.assertTrue(v.is_valid(data))
def test_linux_bridge(self):
schema = validator.get_schema_for_defined_type("linux_bridge")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "linux_bridge",
"name": "br-ctlplane",
"use_dhcp": True,
"members": [
{"type": "interface", "name": "em1"}
]
}
self.assertTrue(v.is_valid(data))
def test_linux_bond(self):
schema = validator.get_schema_for_defined_type("linux_bond")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "linux_bond",
"name": "bond1",
"use_dhcp": True,
"bonding_options": "mode=active-backup",
"members": [
{"type": "interface", "name": "em1"},
{"type": "interface", "name": "em2"}
]
}
self.assertTrue(v.is_valid(data))
def test_nfvswitch_bridge(self):
schema = validator.get_schema_for_defined_type("nfvswitch_bridge")
v = jsonschema.Draft4Validator(schema)
data = {
"type": "nfvswitch_bridge",
"options": "-c 2,3,4,5",
"members": [{
"type": "nfvswitch_internal",
"name": "api",
"addresses": [
{"ip_netmask": "172.16.2.7/24"}
],
"vlan_id": 201
}, {
"type": "nfvswitch_internal",
"name": "storage",
"addresses": [
{"ip_netmask": "172.16.1.6/24"}
],
"vlan_id": 202
}]
}
self.assertTrue(v.is_valid(data))
class TestSampleFiles(base.TestCase):
def test_sample_files(self):
sample_files = (glob.glob(os.path.join(SAMPLE_BASE, '*.json')) +
glob.glob(os.path.join(SAMPLE_BASE, '*.yaml')))
for sample_file in sample_files:
with open(sample_file, 'r') as f:
try:
config = yaml.load(f.read()).get("network_config")
except Exception:
continue
if not config:
continue
errors = validator.validate_config(config, sample_file)
if os.path.basename(sample_file).startswith("invalid_"):
self.assertTrue(errors)
else:
self.assertFalse(errors)

181
os_net_config/validator.py Normal file
View File

@ -0,0 +1,181 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import collections
import copy
import jsonschema
import pkg_resources
import yaml
def get_os_net_config_schema():
"""Returns the schema for os_net_config's config files."""
schema_string = pkg_resources.resource_string(__name__, "schema.yaml")
return yaml.load(schema_string)
def get_schema_for_defined_type(defined_type):
"""Returns the schema for a given defined type of the full schema."""
full_schema = get_os_net_config_schema()
type_schema = copy.deepcopy(full_schema["definitions"][defined_type])
type_schema["$schema"] = full_schema["$schema"]
type_schema["definitions"] = full_schema["definitions"]
return type_schema
def validate_config(config, config_name="Config file"):
"""Validates a list of interface/bridge configurations against the schema.
If validation fails, returns a list of validation error message strings.
If validation succeeds, returns an empty list.
`config_name` can be used to prefix errors with a more specific name.
"""
return _validate_config(config, config_name,
get_os_net_config_schema(), True)
def _validate_config(config, config_name, schema, filter_errors):
error_messages = []
validator = jsonschema.Draft4Validator(schema)
v_errors = validator.iter_errors(config)
v_errors = sorted(v_errors, key=lambda e: e.path)
for v_error in v_errors:
error_message = _get_consistent_error_message(v_error)
details = _get_detailed_errors(v_error, 1, v_error.schema_path,
schema, filter_errors=filter_errors)
config_path = '/'.join([str(x) for x in v_error.path])
if details:
error_messages.append(
"{} failed schema validation at network_config/{}:\n"
" {}\n"
" Sub-schemas tested and not matching:\n"
" {}"
.format(config_name, config_path, error_message,
'\n '.join(details)))
else:
error_messages.append(
"{} failed schema validation at network_config/{}:\n"
" {}"
.format(config_name, config_path, error_message))
return error_messages
def _get_consistent_error_message(error):
"""Returns error messages consistent across Python 2 and 3.
jsonschema uses repr() to print its error messages, which means strings
will render as "u'...'" in Python 2 and "'...'" in Python 3, making
testing for error messages unnecessarily difficult.
"""
if error.validator == 'type':
return "'{}' is not of type '{}'".format(
error.instance, error.validator_value)
elif error.validator == 'enum':
return "'{}' is not one of ['{}']".format(
error.instance, "','".join(error.validator_value))
elif error.validator == 'required':
if error.message[0:2] == "u'":
return error.message[1:]
return error.message
def _get_detailed_errors(error, depth, absolute_schema_path, absolute_schema,
filter_errors=True):
"""Returns a list of error messages from all subschema validations.
Recurses the error tree and adds one message per sub error. That list can
get long, because jsonschema also tests the hypothesis that the provided
network element type is wrong (e.g. "ovs_bridge" instead of "ovs_bond").
Setting `filter_errors=True` assumes the type, if specified, is correct and
therefore produces a much shorter list of more relevant results.
"""
if not error.context:
return []
sub_errors = error.context
if filter_errors:
if (absolute_schema_path[-1] in ['oneOf', 'anyOf'] and
isinstance(error.instance, collections.Mapping) and
'type' in error.instance):
found, index = _find_type_in_schema_list(
error.validator_value, error.instance['type'])
if found:
sub_errors = [i for i in sub_errors if (
i.schema_path[0] == index)]
details = []
sub_errors = sorted(sub_errors, key=lambda e: e.schema_path)
for sub_error in sub_errors:
schema_path = collections.deque(absolute_schema_path)
schema_path.extend(sub_error.schema_path)
details.append("{} {}: {}".format(
'-' * depth,
_pretty_print_schema_path(schema_path, absolute_schema),
_get_consistent_error_message(sub_error)))
details.extend(_get_detailed_errors(
sub_error, depth + 1, schema_path, absolute_schema,
filter_errors))
return details
def _find_type_in_schema_list(schemas, type_to_find):
"""Finds an object of a given type in an anyOf/oneOf array.
Returns a tuple (`found`, `index`), where `found` indicates whether
on object of type `type_to_find` was found in the `schemas` array.
If so, `index` contains the object's position in the array.
"""
for index, schema in enumerate(schemas):
if not isinstance(schema, collections.Mapping):
continue
if ('$ref' in schema and
schema['$ref'].split('/')[-1] == type_to_find):
return (True, index)
if ('properties' in schema and 'type' in schema['properties'] and
schema['properties']['type'] == type_to_find):
return (True, index)
return (False, 0)
def _pretty_print_schema_path(absolute_schema_path, absolute_schema):
"""Returns a representation of the schema path that's easier to read.
For example:
>>> _pretty_print_schema_path("items/oneOf/0/properties/use_dhcp/oneOf/2")
"items/oneOf/interface/use_dhcp/oneOf/param"
"""
pretty_path = []
current_path = []
current_schema = absolute_schema
for item in absolute_schema_path:
if item not in ["properties"]:
pretty_path.append(item)
current_path.append(item)
current_schema = current_schema[item]
if (isinstance(current_schema, collections.Mapping) and
'$ref' in current_schema):
if (isinstance(pretty_path[-1], int) and
pretty_path[-2] in ['oneOf', 'anyOf']):
pretty_path[-1] = current_schema['$ref'].split('/')[-1]
current_path = current_schema['$ref'].split('/')
current_schema = absolute_schema
for i in current_path[1:]:
current_schema = current_schema[i]
return '/'.join([str(x) for x in pretty_path])

View File

@ -10,3 +10,4 @@ netaddr!=0.7.16,>=0.7.13 # BSD
oslo.concurrency>=3.8.0 # Apache-2.0
oslo.utils>=3.20.0 # Apache-2.0
PyYAML>=3.10.0 # MIT
jsonschema>=2.0.0,<3.0.0,!=2.5.0 # MIT