Python源码示例：jsonschema.Draft4Validator()
示例1
def validate_drydock_document(self, doc):
"""Validate a parsed document via jsonschema.
If a schema for a document Kind is not available, the document is
considered valid. Schema is chosen by the doc['kind'] field.
Returns a empty list for valid documents, otherwise returns a list
of all found errors
:param doc: dictionary of the parsed document.
"""
schemaname = doc.get('schema', '')
(schema_ns, doc_kind, doc_version) = schemaname.split('/')
errors_found = []
if doc_version == 'v1':
if schemaname in self.v1_doc_schemas:
validator = jsonschema.Draft4Validator(
self.v1_doc_schemas.get(schemaname))
for error in validator.iter_errors(doc.get('data', [])):
errors_found.append(error.message)
return errors_found
示例2
def validate_drydock_document(self, doc):
"""Validate a parsed document via jsonschema.
If a schema for a document Kind is not available, the document is
considered valid. Schema is chosen by the doc['kind'] field.
Returns a empty list for valid documents, otherwise returns a list
of all found errors
:param doc: dictionary of the parsed document.
"""
doc_kind = doc.get('kind')
if doc_kind in self.v1_doc_schemas:
validator = jsonschema.Draft4Validator(
self.v1_doc_schemas.get(doc_kind))
errors_found = []
for error in validator.iter_errors(doc):
errors_found.append(error.message)
return errors_found
else:
return []
示例3
def validate(ctx):
"""
Validate the paradrop.yaml file.
A note about versions: this command validates the chute configuration
against the current rules for the installed version of pdtools. If
the chute is to be installed on a Paradrop node running a different
version, then this command may not be reliable for determining
compatibility.
"""
with open('paradrop.yaml', 'r') as source:
chute = yaml.safe_load(source)
schema_path = pkg_resources.resource_filename('pdtools', 'schemas/chute.json')
with open(schema_path, 'r') as source:
schema = json.load(source)
validator = jsonschema.Draft4Validator(schema)
for error in sorted(validator.iter_errors(chute), key=str):
click.echo(error.message)
示例4
def add(self, message):
if isinstance(message, singer.RecordMessage):
stream = self.ensure_stream(message.stream)
if stream.latest_schema:
validator_fn = extend_with_default(Draft4Validator)
validator = validator_fn(
stream.latest_schema, format_checker=FormatChecker())
validator.validate(copy.deepcopy(message.record))
else:
print('I saw a record for stream {} before the schema'.format(
message.stream))
exit(1)
stream.num_records += 1
elif isinstance(message, singer.SchemaMessage):
stream = self.ensure_stream(message.stream)
stream.num_schemas += 1
stream.latest_schema = message.schema
elif isinstance(message, singer.StateMessage):
self.latest_state = message.value
self.num_states += 1
示例5
def test_draft3_schema_draft4_validator(self):
stdout, stderr = StringIO(), StringIO()
with self.assertRaises(SchemaError):
cli.run(
{
"validator": Draft4Validator,
"schema": {
"anyOf": [
{"minimum": 20},
{"type": "string"},
{"required": True},
],
},
"instances": [1],
"error_format": "{error.message}",
},
stdout=stdout,
stderr=stderr,
)
示例6
def test_oneOf_and_anyOf_are_weak_matches(self):
"""
A property you *must* match is probably better than one you have to
match a part of.
"""
validator = Draft4Validator(
{
"minProperties": 2,
"anyOf": [{"type": "string"}, {"type": "number"}],
"oneOf": [{"type": "string"}, {"type": "number"}],
}
)
best = self.best_match(validator.iter_errors({}))
self.assertEqual(best.validator, "minProperties")
示例7
def test_if_the_most_relevant_error_is_anyOf_it_is_traversed(self):
"""
If the most relevant error is an anyOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"anyOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
示例8
def test_if_the_most_relevant_error_is_allOf_it_is_traversed(self):
"""
Now, if the error is allOf, we traverse but select the *most* relevant
error from the context, because all schemas here must match anyways.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"allOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "string")
示例9
def test_nested_context_for_oneOf(self):
validator = Draft4Validator(
{
"properties": {
"foo": {
"oneOf": [
{"type": "string"},
{
"oneOf": [
{"type": "string"},
{
"properties": {
"bar": {"type": "array"},
},
},
],
},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
示例10
def test_oneOf_and_anyOf_are_weak_matches(self):
"""
A property you *must* match is probably better than one you have to
match a part of.
"""
validator = Draft4Validator(
{
"minProperties": 2,
"anyOf": [{"type": "string"}, {"type": "number"}],
"oneOf": [{"type": "string"}, {"type": "number"}],
}
)
best = self.best_match(validator.iter_errors({}))
self.assertEqual(best.validator, "minProperties")
示例11
def test_if_the_most_relevant_error_is_anyOf_it_is_traversed(self):
"""
If the most relevant error is an anyOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"anyOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
示例12
def test_if_the_most_relevant_error_is_oneOf_it_is_traversed(self):
"""
If the most relevant error is an oneOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"oneOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
示例13
def test_if_the_most_relevant_error_is_allOf_it_is_traversed(self):
"""
Now, if the error is allOf, we traverse but select the *most* relevant
error from the context, because all schemas here must match anyways.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"allOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "string")
示例14
def test_oneOf_and_anyOf_are_weak_matches(self):
"""
A property you *must* match is probably better than one you have to
match a part of.
"""
validator = Draft4Validator(
{
"minProperties" : 2,
"anyOf" : [{"type" : "string"}, {"type" : "number"}],
"oneOf" : [{"type" : "string"}, {"type" : "number"}],
}
)
best = self.best_match(validator.iter_errors({}))
self.assertEqual(best.validator, "minProperties")
示例15
def test_if_the_most_relevant_error_is_anyOf_it_is_traversed(self):
"""
If the most relevant error is an anyOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties" : {
"foo" : {
"anyOf" : [
{"type" : "string"},
{"properties" : {"bar" : {"type" : "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo" : {"bar" : 12}}))
self.assertEqual(best.validator_value, "array")
示例16
def test_if_the_most_relevant_error_is_oneOf_it_is_traversed(self):
"""
If the most relevant error is an oneOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties" : {
"foo" : {
"oneOf" : [
{"type" : "string"},
{"properties" : {"bar" : {"type" : "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo" : {"bar" : 12}}))
self.assertEqual(best.validator_value, "array")
示例17
def test_nested_context_for_oneOf(self):
validator = Draft4Validator(
{
"properties" : {
"foo" : {
"oneOf" : [
{"type" : "string"},
{
"oneOf" : [
{"type" : "string"},
{
"properties" : {
"bar" : {"type" : "array"}
},
},
],
},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo" : {"bar" : 12}}))
self.assertEqual(best.validator_value, "array")
示例18
def setup(self, level=None, log_file=None, json=None):
'''
Load everything up. Note that any arg here will override both
default and custom settings
@param level: the log level
@param log_file: boolean t/f whether to log to a file, else stdout
@param json: boolean t/f whether to write the logs in json
'''
self.settings = self.wrapper.load(self.settings_name)
my_level = level if level else self.settings['LOG_LEVEL']
# negate because logger wants True for std out
my_output = not log_file if log_file else self.settings['LOG_STDOUT']
my_json = json if json else self.settings['LOG_JSON']
self.logger = LogFactory.get_instance(json=my_json, stdout=my_output,
level=my_level,
name=self.settings['LOGGER_NAME'],
dir=self.settings['LOG_DIR'],
file=self.settings['LOG_FILE'],
bytes=self.settings['LOG_MAX_BYTES'],
backups=self.settings['LOG_BACKUPS'])
self.validator = self.extend_with_default(Draft4Validator)
示例19
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
示例20
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"))
示例21
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"],
"domain": "openstack.local",
"mtu": 1501,
"ethtool_opts": "speed 1000 duplex full",
"hotplug": True,
"onboot": 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))
示例22
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))
示例23
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))
示例24
def is_datafile_valid(datafile):
""" Given a datafile determine if it is valid or not.
Args:
datafile: JSON string representing the project.
Returns:
Boolean depending upon whether datafile is valid or not.
"""
try:
datafile_json = json.loads(datafile)
except:
return False
try:
jsonschema.Draft4Validator(constants.JSON_SCHEMA).validate(datafile_json)
except:
return False
return True
示例25
def validate_json(self, data):
try:
validator = jsonschema.Draft4Validator(self.schema)
validation_errors = [e for e in validator.iter_errors(data)]
# Iteratre over all errors and raise as single exception
if validation_errors:
exception_msgs = {}
for err in validation_errors:
if err.path:
field = '-'.join([str(e) for e in err.path])
elif err.schema_path:
field = '-'.join([str(e) for e in err.schema_path])
else:
field = 'error'
if field in exception_msgs:
exception_msgs[field].append(err.message)
else:
exception_msgs[field] = [err.message]
raise serializers.ValidationError(exception_msgs)
except (JSONSchemaValidationError, JSONSchemaError) as e:
raise serializers.ValidationError(e.message)
return self.to_internal_value(json.dumps(data))
示例26
def test_draft3_schema_draft4_validator(self):
stdout, stderr = StringIO(), StringIO()
with self.assertRaises(SchemaError):
cli.run(
{
"validator": Draft4Validator,
"schema": {
"anyOf": [
{"minimum": 20},
{"type": "string"},
{"required": True},
],
},
"instances": [1],
"error_format": "{error.message}",
},
stdout=stdout,
stderr=stderr,
)
示例27
def test_if_the_most_relevant_error_is_anyOf_it_is_traversed(self):
"""
If the most relevant error is an anyOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"anyOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
示例28
def test_if_the_most_relevant_error_is_oneOf_it_is_traversed(self):
"""
If the most relevant error is an oneOf, then we traverse its context
and select the otherwise *least* relevant error, since in this case
that means the most specific, deep, error inside the instance.
I.e. since only one of the schemas must match, we look for the most
relevant one.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"oneOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
示例29
def test_if_the_most_relevant_error_is_allOf_it_is_traversed(self):
"""
Now, if the error is allOf, we traverse but select the *most* relevant
error from the context, because all schemas here must match anyways.
"""
validator = Draft4Validator(
{
"properties": {
"foo": {
"allOf": [
{"type": "string"},
{"properties": {"bar": {"type": "array"}}},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "string")
示例30
def test_nested_context_for_oneOf(self):
validator = Draft4Validator(
{
"properties": {
"foo": {
"oneOf": [
{"type": "string"},
{
"oneOf": [
{"type": "string"},
{
"properties": {
"bar": {"type": "array"},
},
},
],
},
],
},
},
},
)
best = self.best_match(validator.iter_errors({"foo": {"bar": 12}}))
self.assertEqual(best.validator_value, "array")
