spec validators added

Author: p1c2u

.travis.yml

@@ -2,14 +2,12 @@ language: python
 sudo: false
 matrix:
   include:
-  - python: 2.7
   - python: 3.2
   - python: 3.3
   - python: 3.4
   - python: 3.5
   - python: 3.6
   - python: nightly
-  - python: pypy
   - python: pypy3
   allow_failures:
   - python: 3.2

README.md

@@ -34,10 +34,9 @@ If you want to iterate through validation errors:
 
 ```python
 
-from openapi_spec_validator import openapi_v3_validator_factory
+from openapi_spec_validator import openapi_v3_spec_validator
 
-validator = openapi_v3_validator_factory.create(spec)
-errors_iterator = validator.iter_errors(spec)
+errors_iterator = openapi_v3_spec_validator.iter_errors(spec)
 ```
 
 ## License

openapi_spec_validator/__init__.py

@@ -5,6 +5,7 @@
 from openapi_spec_validator.handlers import UrlHandler
 from openapi_spec_validator.schemas import get_openapi_schema
 from openapi_spec_validator.factories import JSONSpecValidatorFactory
+from openapi_spec_validator.validators import SpecValidator
 
 __author__ = 'Artur Maciąg'
 __email__ = '[email protected]'
@@ -25,8 +26,11 @@
     schema_v3, schema_v3_url,
     resolver_handlers=default_handlers,
 )
+openapi_v3_spec_validator = SpecValidator(
+    openapi_v3_validator_factory,
+    resolver_handlers=default_handlers,
+)
 # shortcuts
-validate_spec = validate_spec_factory(
-    openapi_v3_validator_factory.create)
+validate_spec = validate_spec_factory(openapi_v3_spec_validator.validate)
 validate_spec_url = validate_spec_url_factory(
-    openapi_v3_validator_factory.create, default_handlers)
+    openapi_v3_spec_validator.validate, default_handlers)

openapi_spec_validator/exceptions.py

@@ -0,0 +1,17 @@
+from jsonschema.exceptions import ValidationError
+
+
+class OpenAPIValidationError(ValidationError):
+    pass
+
+
+class ExtraParametersError(OpenAPIValidationError):
+    pass
+
+
+class ParameterDuplicateError(OpenAPIValidationError):
+    pass
+
+
+class UnresolvableParameterError(OpenAPIValidationError):
+    pass

openapi_spec_validator/factories.py

@@ -50,18 +50,14 @@ def __init__(self, schema, schema_url='', resolver_handlers=None):
     def schema_resolver(self):
         return self._get_resolver(self.schema_url, self.schema)
 
-    def create(self, spec, spec_url=''):
-        """Creates json documents validator from spec.
-        :param spec: json document in the form of a list or dict.
-        :param spec_url: base uri to use when creating a
-            RefResolver for the passed in spec_dict.
+    def create(self, spec_resolver):
+        """Creates json documents validator from spec resolver.
+        :param spec_resolver: reference resolver.
 
         :return: RefResolver for spec with cached remote $refs used during
             validation.
         :rtype: :class:`jsonschema.RefResolver`
         """
-        spec_resolver = self._get_resolver(spec_url, spec)
-
         validator_cls = self.spec_validator_factory.from_resolver(
             spec_resolver)
 

openapi_spec_validator/managers.py

@@ -17,3 +17,20 @@ def visit(self, key):
             yield key
         finally:
             del self[key]
+
+
+class ResolverManager(object):
+    def __init__(self, resolver):
+        self.resolver = resolver
+
+    @contextmanager
+    def in_scope(self, item, scope='x-scope'):
+        if scope not in item:
+            yield self.resolver
+        else:
+            saved_scope_stack = self.resolver._scopes_stack
+            try:
+                self.resolver._scopes_stack = item[scope]
+                yield self.resolver
+            finally:
+                self.resolver._scopes_stack = saved_scope_stack

openapi_spec_validator/shortcuts.py

@@ -2,18 +2,16 @@
 from six.moves.urllib import parse
 
 
-def validate_spec_factory(validator_factory_callable):
+def validate_spec_factory(validator_callable):
     def validate(spec):
-        validator = validator_factory_callable(spec)
-        return validator.validate(spec)
+        return validator_callable(spec)
     return validate
 
 
-def validate_spec_url_factory(validator_factory_callable, handlers):
+def validate_spec_url_factory(validator_callable, handlers):
     def validate(url):
         result = parse.urlparse(url)
         handler = handlers[result.scheme]
         spec = handler(url)
-        validator = validator_factory_callable(spec, spec_url=url)
-        return validator.validate(spec)
+        return validator_callable(spec, spec_url=url)
     return validate

openapi_spec_validator/validators.py

@@ -0,0 +1,240 @@
+import logging
+import string
+
+from jsonschema.validators import RefResolver
+from six import iteritems
+
+from openapi_spec_validator.exceptions import (
+    ParameterDuplicateError, ExtraParametersError, UnresolvableParameterError,
+)
+from openapi_spec_validator.managers import ResolverManager
+
+log = logging.getLogger(__name__)
+
+
+def is_ref(spec):
+    return isinstance(spec, dict) and '$ref' in spec
+
+
+class Dereferencer(object):
+
+    def __init__(self, spec_resolver):
+        self.resolver_manager = ResolverManager(spec_resolver)
+
+    def dereference(self, item):
+        log.debug("Dereferencing %s", item)
+        if item is None or not is_ref(item):
+            return item
+
+        ref = item['$ref']
+        with self.resolver_manager.in_scope(item) as resolver:
+            with resolver.resolving(ref) as target:
+                return target
+
+
+class SpecValidator(object):
+
+    def __init__(self, validator_factory, resolver_handlers):
+        self.validator_factory = validator_factory
+        self.resolver_handlers = resolver_handlers
+
+    def validate(self, spec, spec_url=''):
+        for error in self.iter_errors(spec, spec_url=spec_url):
+            raise error
+
+    def iter_errors(self, spec, spec_url=''):
+        spec_resolver = self._get_resolver(spec_url, spec)
+        dereferencer = self._get_dereferencer(spec_resolver)
+
+        validator = self._get_validator(spec_resolver)
+        yield from validator.iter_errors(spec)
+
+        paths = spec.get('paths', {})
+        yield from self._iter_paths_errors(paths, dereferencer)
+
+        components = spec.get('components', {})
+        yield from self._iter_components_errors(components, dereferencer)
+
+    def _get_resolver(self, base_uri, referrer):
+        return RefResolver(
+            base_uri, referrer, handlers=self.resolver_handlers)
+
+    def _get_dereferencer(self, spec_resolver):
+        return Dereferencer(spec_resolver)
+
+    def _get_validator(self, spec_resolver):
+        return self.validator_factory.create(spec_resolver)
+
+    def _iter_paths_errors(self, paths, dereferencer):
+        return PathsValidator(dereferencer).iter_errors(paths)
+
+    def _iter_components_errors(self, components, dereferencer):
+        return ComponentsValidator(dereferencer).iter_errors(components)
+
+
+class ComponentsValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, components):
+        components_deref = self.dereferencer.dereference(components)
+
+        schemas = components_deref.get('schemas', {})
+        yield from self._iter_schemas_errors(schemas)
+
+    def _iter_schemas_errors(self, schemas):
+        return SchemasValidator(self.dereferencer).iter_errors(schemas)
+
+
+class SchemasValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, schemas):
+        schemas_deref = self.dereferencer.dereference(schemas)
+        for name, schema in iteritems(schemas_deref):
+            yield from self._iter_schem_errors(schema)
+
+    def _iter_schem_errors(self, schema):
+        return SchemaValidator(self.dereferencer).iter_errors(schema)
+
+
+class SchemaValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, schema):
+        schema_deref = self.dereferencer.dereference(schema)
+
+        if 'allOf' in schema_deref:
+            for inner_schema in schema_deref['allOf']:
+                yield from self.iter_errors(inner_schema)
+
+        required = schema_deref.get('required', [])
+        properties = schema_deref.get('properties', {}).keys()
+        extra_properties = list(set(required) - set(properties))
+        if extra_properties:
+            yield ExtraParametersError(
+                "Required list has not defined properties: {0}".format(
+                    extra_properties
+                )
+            )
+
+
+class PathsValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, paths):
+        paths_deref = self.dereferencer.dereference(paths)
+        for url, path_item in iteritems(paths_deref):
+            yield from self._iter_path_errors(url, path_item)
+
+    def _iter_path_errors(self, url, path_item):
+        return PathValidator(self.dereferencer).iter_errors(url, path_item)
+
+
+class PathValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, url, path_item):
+        path_item_deref = self.dereferencer.dereference(path_item)
+
+        yield from self._iter_path_item_errors(url, path_item_deref)
+
+    def _iter_path_item_errors(self, url, path_item):
+        return PathItemValidator(self.dereferencer).iter_errors(url, path_item)
+
+
+class PathItemValidator(object):
+
+    OPERATIONS = [
+        'get', 'put', 'post', 'delete', 'options', 'head', 'patch', 'trace',
+    ]
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, url, path_item):
+        path_item_deref = self.dereferencer.dereference(path_item)
+
+        parameters = path_item_deref.get('parameters', [])
+        yield from self._iter_parameters_errors(parameters)
+
+        for field_name, operation in iteritems(path_item):
+            if field_name not in self.OPERATIONS:
+                continue
+
+            yield from self._iter_operation_errors(
+                url, field_name, operation, parameters)
+
+    def _iter_operation_errors(self, url, name, operation, path_parameters):
+        return OperationValidator(self.dereferencer).iter_errors(
+            url, name, operation, path_parameters)
+
+    def _iter_parameters_errors(self, parameters):
+        return ParametersValidator(self.dereferencer).iter_errors(parameters)
+
+
+class OperationValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, url, name, operation, path_parameters=None):
+        path_parameters = path_parameters or []
+        operation_deref = self.dereferencer.dereference(operation)
+
+        parameters = operation_deref.get('parameters', [])
+        yield from self._iter_parameters_errors(parameters)
+
+        all_params = list(set(
+            list(self._get_path_param_names(path_parameters)) +
+            list(self._get_path_param_names(parameters))
+        ))
+
+        for path in self._get_path_params_from_url(url):
+            if path not in all_params:
+                yield UnresolvableParameterError(
+                    "Path parameter '{0}' for '{1}' operation in '{2}' "
+                    "was not resolved".format(path, name, url)
+                )
+
+        return []
+
+    def _get_path_param_names(self, params):
+        for param in params:
+            param_deref = self.dereferencer.dereference(param)
+            if param_deref['in'] == 'path':
+                yield param_deref['name']
+
+    def _get_path_params_from_url(self, url):
+        formatter = string.Formatter()
+        path_params = [item[1] for item in formatter.parse(url)]
+        return filter(None, path_params)
+
+    def _iter_parameters_errors(self, parameters):
+        return ParametersValidator(self.dereferencer).iter_errors(parameters)
+
+
+class ParametersValidator(object):
+
+    def __init__(self, dereferencer):
+        self.dereferencer = dereferencer
+
+    def iter_errors(self, parameters):
+        seen = set()
+        for parameter in parameters:
+            parameter_deref = self.dereferencer.dereference(parameter)
+            key = (parameter_deref['name'], parameter_deref['in'])
+            if key in seen:
+                yield ParameterDuplicateError(
+                    "Duplicate parameter `{0}`".format(parameter_deref['name'])
+                )
+            seen.add(key)

setup.py

@@ -70,11 +70,12 @@ def run_tests(self):
     tests_require=read_requirements('requirements_dev.txt'),
     cmdclass={'test': PyTest},
     classifiers=[
-        "Development Status :: 5 - Production/Stable",
+        "Development Status :: 4 - Beta",
         "Intended Audience :: Developers",
         "Topic :: Software Development :: Libraries :: Python Modules",
         "Operating System :: OS Independent",
-        "Programming Language :: Python :: 2.7",
+        "Programming Language :: Python :: 3.3",
+        "Programming Language :: Python :: 3.4",
         "Programming Language :: Python :: 3.5",
         "Programming Language :: Python :: 3.6",
     ],