Rework things a bit, remove py.test, allow multiple instances, and a validator.
This commit is contained in:
parent
fad8a030f3
commit
80ec8c0ec2
155
jsonschema/_reflect.py
Normal file
155
jsonschema/_reflect.py
Normal file
@ -0,0 +1,155 @@
|
||||
# -*- test-case-name: twisted.test.test_reflect -*-
|
||||
# Copyright (c) Twisted Matrix Laboratories.
|
||||
# See LICENSE for details.
|
||||
|
||||
"""
|
||||
Standardized versions of various cool and/or strange things that you can do
|
||||
with Python's reflection capabilities.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
from jsonschema.compat import PY3
|
||||
|
||||
|
||||
class _NoModuleFound(Exception):
|
||||
"""
|
||||
No module was found because none exists.
|
||||
"""
|
||||
|
||||
|
||||
|
||||
class InvalidName(ValueError):
|
||||
"""
|
||||
The given name is not a dot-separated list of Python objects.
|
||||
"""
|
||||
|
||||
|
||||
|
||||
class ModuleNotFound(InvalidName):
|
||||
"""
|
||||
The module associated with the given name doesn't exist and it can't be
|
||||
imported.
|
||||
"""
|
||||
|
||||
|
||||
|
||||
class ObjectNotFound(InvalidName):
|
||||
"""
|
||||
The object associated with the given name doesn't exist and it can't be
|
||||
imported.
|
||||
"""
|
||||
|
||||
|
||||
|
||||
if PY3:
|
||||
def reraise(exception, traceback):
|
||||
raise exception.with_traceback(traceback)
|
||||
else:
|
||||
exec("""def reraise(exception, traceback):
|
||||
raise exception.__class__, exception, traceback""")
|
||||
|
||||
reraise.__doc__ = """
|
||||
Re-raise an exception, with an optional traceback, in a way that is compatible
|
||||
with both Python 2 and Python 3.
|
||||
|
||||
Note that on Python 3, re-raised exceptions will be mutated, with their
|
||||
C{__traceback__} attribute being set.
|
||||
|
||||
@param exception: The exception instance.
|
||||
@param traceback: The traceback to use, or C{None} indicating a new traceback.
|
||||
"""
|
||||
|
||||
|
||||
def _importAndCheckStack(importName):
|
||||
"""
|
||||
Import the given name as a module, then walk the stack to determine whether
|
||||
the failure was the module not existing, or some code in the module (for
|
||||
example a dependent import) failing. This can be helpful to determine
|
||||
whether any actual application code was run. For example, to distiguish
|
||||
administrative error (entering the wrong module name), from programmer
|
||||
error (writing buggy code in a module that fails to import).
|
||||
|
||||
@param importName: The name of the module to import.
|
||||
@type importName: C{str}
|
||||
@raise Exception: if something bad happens. This can be any type of
|
||||
exception, since nobody knows what loading some arbitrary code might
|
||||
do.
|
||||
@raise _NoModuleFound: if no module was found.
|
||||
"""
|
||||
try:
|
||||
return __import__(importName)
|
||||
except ImportError:
|
||||
excType, excValue, excTraceback = sys.exc_info()
|
||||
while excTraceback:
|
||||
execName = excTraceback.tb_frame.f_globals["__name__"]
|
||||
# in Python 2 execName is None when an ImportError is encountered,
|
||||
# where in Python 3 execName is equal to the importName.
|
||||
if execName is None or execName == importName:
|
||||
reraise(excValue, excTraceback)
|
||||
excTraceback = excTraceback.tb_next
|
||||
raise _NoModuleFound()
|
||||
|
||||
|
||||
|
||||
def namedAny(name):
|
||||
"""
|
||||
Retrieve a Python object by its fully qualified name from the global Python
|
||||
module namespace. The first part of the name, that describes a module,
|
||||
will be discovered and imported. Each subsequent part of the name is
|
||||
treated as the name of an attribute of the object specified by all of the
|
||||
name which came before it. For example, the fully-qualified name of this
|
||||
object is 'twisted.python.reflect.namedAny'.
|
||||
|
||||
@type name: L{str}
|
||||
@param name: The name of the object to return.
|
||||
|
||||
@raise InvalidName: If the name is an empty string, starts or ends with
|
||||
a '.', or is otherwise syntactically incorrect.
|
||||
|
||||
@raise ModuleNotFound: If the name is syntactically correct but the
|
||||
module it specifies cannot be imported because it does not appear to
|
||||
exist.
|
||||
|
||||
@raise ObjectNotFound: If the name is syntactically correct, includes at
|
||||
least one '.', but the module it specifies cannot be imported because
|
||||
it does not appear to exist.
|
||||
|
||||
@raise AttributeError: If an attribute of an object along the way cannot be
|
||||
accessed, or a module along the way is not found.
|
||||
|
||||
@return: the Python object identified by 'name'.
|
||||
"""
|
||||
if not name:
|
||||
raise InvalidName('Empty module name')
|
||||
|
||||
names = name.split('.')
|
||||
|
||||
# if the name starts or ends with a '.' or contains '..', the __import__
|
||||
# will raise an 'Empty module name' error. This will provide a better error
|
||||
# message.
|
||||
if '' in names:
|
||||
raise InvalidName(
|
||||
"name must be a string giving a '.'-separated list of Python "
|
||||
"identifiers, not %r" % (name,))
|
||||
|
||||
topLevelPackage = None
|
||||
moduleNames = names[:]
|
||||
while not topLevelPackage:
|
||||
if moduleNames:
|
||||
trialname = '.'.join(moduleNames)
|
||||
try:
|
||||
topLevelPackage = _importAndCheckStack(trialname)
|
||||
except _NoModuleFound:
|
||||
moduleNames.pop()
|
||||
else:
|
||||
if len(names) == 1:
|
||||
raise ModuleNotFound("No module named %r" % (name,))
|
||||
else:
|
||||
raise ObjectNotFound('%r does not name an object' % (name,))
|
||||
|
||||
obj = topLevelPackage
|
||||
for n in names[1:]:
|
||||
obj = getattr(obj, n)
|
||||
|
||||
return obj
|
@ -1,40 +1,72 @@
|
||||
from __future__ import absolute_import
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
|
||||
from . import (
|
||||
validate, Draft4Validator, Draft3Validator,
|
||||
draft3_format_checker, draft4_format_checker,
|
||||
from jsonschema._reflect import namedAny
|
||||
from jsonschema.validators import validator_for
|
||||
|
||||
|
||||
def _namedAnyWithDefault(name):
|
||||
if "." not in name:
|
||||
name = "jsonschema." + name
|
||||
return namedAny(name)
|
||||
|
||||
|
||||
def _json_file(path):
|
||||
with open(path) as file:
|
||||
return json.load(file)
|
||||
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description="JSON Schema Validation CLI",
|
||||
)
|
||||
from .validators import validator_for
|
||||
parser.add_argument(
|
||||
"-i", "--instance",
|
||||
action="append",
|
||||
dest="instances",
|
||||
type=_json_file,
|
||||
help="a path to a JSON instance to validate "
|
||||
"(may be specified multiple times)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-F", "--error-format",
|
||||
default="{error.instance}: {error.message}\n",
|
||||
help="the format to use for each error output message, specified in "
|
||||
"a form suitable for passing to str.format, which will be called "
|
||||
"with 'error' for each error",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-V", "--validator",
|
||||
type=_namedAnyWithDefault,
|
||||
help="the fully qualified object name of a validator to use, or, for "
|
||||
"validators that are registered with jsonschema, simply the name "
|
||||
"of the class.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"schema",
|
||||
help="the JSON Schema to validate with",
|
||||
type=_json_file,
|
||||
)
|
||||
|
||||
|
||||
def parse_args(args):
|
||||
arguments = vars(parser.parse_args(args=args or ["--help"]))
|
||||
if arguments["validator"] is None:
|
||||
arguments["validator"] = validator_for(arguments["schema"])
|
||||
return arguments
|
||||
|
||||
|
||||
def main(args=sys.argv[1:]):
|
||||
parser = argparse.ArgumentParser(description='JSON Schema validator')
|
||||
parser.add_argument('schema', help='filename of the JSON Schema')
|
||||
parser.add_argument('document', help='filename of the JSON document to validate')
|
||||
parser.add_argument('--format', help='validate value format', action='store_true')
|
||||
args = parser.parse_args(args)
|
||||
sys.exit(run(arguments=parse_args(args=args)))
|
||||
|
||||
schema = json.load(open(args.schema, 'r'))
|
||||
document = json.load(open(args.document, 'r'))
|
||||
|
||||
validator = validator_for(schema)
|
||||
if args.format:
|
||||
if validator == Draft4Validator:
|
||||
format_checker = draft4_format_checker
|
||||
elif validator == Draft3Validator:
|
||||
format_checker = draft3_format_checker
|
||||
else:
|
||||
raise NotImplementedError("No format validator for %s specified"
|
||||
% validator.__name__)
|
||||
else:
|
||||
format_checker = None
|
||||
|
||||
validate(document, schema, validator, format_checker=format_checker)
|
||||
# validate raises if the document is invalid, and will show a Traceback to
|
||||
# the user. If the document is valid, show a congratulating message.
|
||||
print("JSON document is valid.")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
def run(arguments, stdout=sys.stdout, stderr=sys.stderr):
|
||||
error_format = arguments["error_format"]
|
||||
validator = arguments["validator"](schema=arguments["schema"])
|
||||
errored = False
|
||||
for instance in arguments["instances"] or ():
|
||||
for error in validator.iter_errors(instance):
|
||||
stderr.write(error_format.format(error=error))
|
||||
errored = True
|
||||
return errored
|
||||
|
@ -11,6 +11,7 @@ PY3 = sys.version_info[0] >= 3
|
||||
|
||||
if PY3:
|
||||
zip = zip
|
||||
from io import StringIO
|
||||
from urllib.parse import (
|
||||
unquote, urljoin, urlunsplit, SplitResult, urlsplit as _urlsplit
|
||||
)
|
||||
@ -20,6 +21,7 @@ if PY3:
|
||||
iteritems = operator.methodcaller("items")
|
||||
else:
|
||||
from itertools import izip as zip # noqa
|
||||
from StringIO import StringIO
|
||||
from urlparse import (
|
||||
urljoin, urlunsplit, SplitResult, urlsplit as _urlsplit # noqa
|
||||
)
|
||||
|
@ -1,78 +1,105 @@
|
||||
import StringIO
|
||||
from jsonschema import Draft4Validator, ValidationError, cli
|
||||
from jsonschema.compat import StringIO
|
||||
from jsonschema.tests.compat import mock, unittest
|
||||
|
||||
import pytest
|
||||
|
||||
from .compat import mock, unittest
|
||||
from .. import (
|
||||
cli, Draft4Validator, Draft3Validator,
|
||||
draft3_format_checker, draft4_format_checker,
|
||||
)
|
||||
def fake_validator(*errors):
|
||||
errors = list(reversed(errors))
|
||||
|
||||
MOCK_SCHEMAS = {
|
||||
'draft3': {"$schema": "http://json-schema.org/draft-03/schema#"},
|
||||
'draft4': {"$schema": "http://json-schema.org/draft-04/schema#"},
|
||||
}
|
||||
class FakeValidator(object):
|
||||
def __init__(self, *args, **kwargs):
|
||||
pass
|
||||
|
||||
def iter_errors(self, instance):
|
||||
if errors:
|
||||
return errors.pop()
|
||||
return []
|
||||
return FakeValidator
|
||||
|
||||
|
||||
class TestParser(unittest.TestCase):
|
||||
|
||||
FakeValidator = fake_validator()
|
||||
|
||||
def setUp(self):
|
||||
self.open = mock.mock_open(read_data='{}')
|
||||
patch = mock.patch.object(cli, "open", self.open, create=True)
|
||||
patch.start()
|
||||
self.addCleanup(patch.stop)
|
||||
|
||||
def test_find_validator_by_fully_qualified_object_name(self):
|
||||
arguments = cli.parse_args(
|
||||
[
|
||||
"--validator",
|
||||
"jsonschema.tests.test_cli.TestParser.FakeValidator",
|
||||
"--instance", "foo.json",
|
||||
"schema.json",
|
||||
]
|
||||
)
|
||||
self.assertIs(arguments["validator"], self.FakeValidator)
|
||||
|
||||
def test_find_validator_in_jsonschema(self):
|
||||
arguments = cli.parse_args(
|
||||
[
|
||||
"--validator", "Draft4Validator",
|
||||
"--instance", "foo.json",
|
||||
"schema.json",
|
||||
]
|
||||
)
|
||||
self.assertIs(arguments["validator"], Draft4Validator)
|
||||
|
||||
|
||||
class TestCLI(unittest.TestCase):
|
||||
def test_missing_arguments(self):
|
||||
with pytest.raises(SystemExit) as e:
|
||||
cli.main([])
|
||||
def test_successful_validation(self):
|
||||
stdout, stderr = StringIO(), StringIO()
|
||||
exit_code = cli.run(
|
||||
{
|
||||
"validator" : fake_validator(),
|
||||
"schema" : {},
|
||||
"instances" : [1],
|
||||
"error_format" : "{error.message}",
|
||||
},
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
)
|
||||
self.assertFalse(stdout.getvalue())
|
||||
self.assertFalse(stderr.getvalue())
|
||||
self.assertEqual(exit_code, 0)
|
||||
|
||||
@mock.patch('__builtin__.open')
|
||||
@mock.patch('jsonschema.cli.validate')
|
||||
def test_filename_argument_order(self, validate, open_):
|
||||
def mock_file(filename, mode):
|
||||
return StringIO.StringIO('{"filename": "%s"}' % filename)
|
||||
open_.side_effect = mock_file
|
||||
def test_unsuccessful_validation(self):
|
||||
error = ValidationError("I am an error!", instance=1)
|
||||
stdout, stderr = StringIO(), StringIO()
|
||||
exit_code = cli.run(
|
||||
{
|
||||
"validator" : fake_validator([error]),
|
||||
"schema" : {},
|
||||
"instances" : [1],
|
||||
"error_format" : "{error.instance} - {error.message}",
|
||||
},
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
)
|
||||
self.assertFalse(stdout.getvalue())
|
||||
self.assertEqual(stderr.getvalue(), "1 - I am an error!")
|
||||
self.assertEqual(exit_code, 1)
|
||||
|
||||
cli.main(['document.json', 'schema.json'])
|
||||
|
||||
open_.assert_has_calls([mock.call('document.json', 'r'),
|
||||
mock.call('schema.json', 'r')],
|
||||
any_order=True)
|
||||
self.assertEqual(open_.call_count, 2)
|
||||
|
||||
validate.assert_called_once_with({'filename': 'schema.json'},
|
||||
{'filename': 'document.json'},
|
||||
Draft4Validator,
|
||||
format_checker=None)
|
||||
|
||||
@mock.patch('__builtin__.open')
|
||||
@mock.patch('jsonschema.cli.json.load')
|
||||
@mock.patch('jsonschema.cli.validate')
|
||||
def test_raise_exception(self, validate, json_load, open_):
|
||||
validate.side_effect = Exception('Did not validate correctly')
|
||||
with pytest.raises(Exception) as e:
|
||||
cli.main([None, None])
|
||||
self.assertEqual(e.exconly(), "Exception: Did not validate correctly")
|
||||
|
||||
@mock.patch('__builtin__.open')
|
||||
@mock.patch('jsonschema.cli.json.load')
|
||||
@mock.patch('jsonschema.cli.validate')
|
||||
def test_format(self, validate, json_load, open_):
|
||||
schema = {"$schema": "http://json-schema.org/draft-04/schema#"}
|
||||
json_load.return_value = schema
|
||||
|
||||
cli.main([None, None])
|
||||
validate.assert_called_once_with(schema, schema, Draft4Validator,
|
||||
format_checker=None)
|
||||
validate.reset_mock()
|
||||
cli.main([None, None, '--format'])
|
||||
validate.assert_called_once_with(schema, schema, Draft4Validator,
|
||||
format_checker=draft4_format_checker)
|
||||
|
||||
@mock.patch('__builtin__.open')
|
||||
@mock.patch('jsonschema.cli.json.load')
|
||||
@mock.patch('jsonschema.cli.validate')
|
||||
def test_draft3(self, validate, json_load, open_):
|
||||
schema = {"$schema": "http://json-schema.org/draft-03/schema#"}
|
||||
json_load.return_value = schema
|
||||
|
||||
cli.main([None, None])
|
||||
validate.assert_called_once_with(schema, schema, Draft3Validator,
|
||||
format_checker=None)
|
||||
validate.reset_mock()
|
||||
cli.main([None, None, '--format'])
|
||||
validate.assert_called_once_with(schema, schema, Draft3Validator,
|
||||
format_checker=draft3_format_checker)
|
||||
def test_unsuccessful_validation_multiple_instances(self):
|
||||
first_errors = [
|
||||
ValidationError("9", instance=1),
|
||||
ValidationError("8", instance=1),
|
||||
]
|
||||
second_errors = [ValidationError("7", instance=2)]
|
||||
stdout, stderr = StringIO(), StringIO()
|
||||
exit_code = cli.run(
|
||||
{
|
||||
"validator" : fake_validator(first_errors, second_errors),
|
||||
"schema" : {},
|
||||
"instances" : [1, 2],
|
||||
"error_format" : "{error.instance} - {error.message}\t",
|
||||
},
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
)
|
||||
self.assertFalse(stdout.getvalue())
|
||||
self.assertEqual(stderr.getvalue(), "1 - 9\t1 - 8\t2 - 7\t")
|
||||
self.assertEqual(exit_code, 1)
|
||||
|
Loading…
Reference in New Issue
Block a user