aboutsummaryrefslogtreecommitdiffstats
path: root/os_net_config/tests/test_validator.py
diff options
context:
space:
mode:
Diffstat (limited to 'os_net_config/tests/test_validator.py')
-rw-r--r--os_net_config/tests/test_validator.py392
1 files changed, 392 insertions, 0 deletions
diff --git a/os_net_config/tests/test_validator.py b/os_net_config/tests/test_validator.py
new file mode 100644
index 0000000..7991d9f
--- /dev/null
+++ b/os_net_config/tests/test_validator.py
@@ -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)