From 29199ff2c49b7eacf780b71e6692ac7e730d2aa8 Mon Sep 17 00:00:00 2001 From: Dmitry Dygalo Date: Fri, 8 May 2020 20:20:36 +0200 Subject: [PATCH] refactor: Move logic related to Open API to a separate class --- src/schemathesis/loaders.py | 14 +++---- src/schemathesis/models.py | 4 -- src/schemathesis/schemas.py | 50 ++++++++++++------------ src/schemathesis/specs/openapi/checks.py | 11 +++++- 4 files changed, 42 insertions(+), 37 deletions(-) diff --git a/src/schemathesis/loaders.py b/src/schemathesis/loaders.py index 2088a39805..3fe98cd0bd 100644 --- a/src/schemathesis/loaders.py +++ b/src/schemathesis/loaders.py @@ -13,7 +13,7 @@ from .exceptions import HTTPError from .hooks import HookContext, dispatch from .lazy import LazySchema -from .schemas import BaseSchema, OpenApi30, SwaggerV20 +from .schemas import BaseOpenAPISchema, OpenApi30, SwaggerV20 from .specs.openapi import definitions from .types import Filter, PathLike from .utils import NOT_SET, StringDatesYAMLLoader, WSGIResponse, get_base_url @@ -29,7 +29,7 @@ def from_path( *, app: Any = None, validate_schema: bool = True, -) -> BaseSchema: +) -> BaseOpenAPISchema: """Load a file from OS path and parse to schema instance.""" with open(path) as fd: return from_file( @@ -56,7 +56,7 @@ def from_uri( app: Any = None, validate_schema: bool = True, **kwargs: Any, -) -> BaseSchema: +) -> BaseOpenAPISchema: """Load a remote resource and parse to schema instance.""" kwargs.setdefault("headers", {}).setdefault("User-Agent", USER_AGENT) response = requests.get(uri, **kwargs) @@ -91,7 +91,7 @@ def from_file( app: Any = None, validate_schema: bool = True, **kwargs: Any, # needed in runner to have compatible API across all loaders -) -> BaseSchema: +) -> BaseOpenAPISchema: """Load a file content and parse to schema instance. `file` could be a file descriptor, string or bytes. @@ -121,7 +121,7 @@ def from_dict( *, app: Any = None, validate_schema: bool = True, -) -> BaseSchema: +) -> BaseOpenAPISchema: """Get a proper abstraction for the given raw schema.""" dispatch("before_load_schema", HookContext(), raw_schema) if "swagger" in raw_schema: @@ -191,7 +191,7 @@ def from_wsgi( operation_id: Optional[Filter] = None, validate_schema: bool = True, **kwargs: Any, -) -> BaseSchema: +) -> BaseOpenAPISchema: kwargs.setdefault("headers", {}).setdefault("User-Agent", USER_AGENT) client = Client(app, WSGIResponse) response = client.get(schema_path, **kwargs) @@ -229,7 +229,7 @@ def from_aiohttp( *, validate_schema: bool = True, **kwargs: Any, -) -> BaseSchema: +) -> BaseOpenAPISchema: from .extra._aiohttp import run_server # pylint: disable=import-outside-toplevel port = run_server(app) diff --git a/src/schemathesis/models.py b/src/schemathesis/models.py index da3ecbb28d..dbf16cf2f3 100644 --- a/src/schemathesis/models.py +++ b/src/schemathesis/models.py @@ -263,10 +263,6 @@ def as_strategy(self, hooks: Optional["HookDispatcher"] = None) -> SearchStrateg return get_case_strategy(self, hooks) - def get_content_types(self, response: GenericResponse) -> List[str]: - """Content types available for this endpoint.""" - return self.schema.get_content_types(self, response) - class Status(IntEnum): """Status of an action or multiple actions.""" diff --git a/src/schemathesis/schemas.py b/src/schemathesis/schemas.py index b619ba93fe..6987abef01 100644 --- a/src/schemathesis/schemas.py +++ b/src/schemathesis/schemas.py @@ -1,4 +1,4 @@ -# pylint: disable=too-many-instance-attributes +# pylint: disable=too-many-instance-attributes,too-many-public-methods,too-many-ancestors """Schema objects provide a convenient interface to raw schemas. Their responsibilities: @@ -29,13 +29,9 @@ from .types import Filter, GenericTest, Hook, NotSet from .utils import NOT_SET, GenericResponse, deprecated -# Reference resolving will stop after this depth -RECURSION_DEPTH_LIMIT = 100 - @attr.s() # pragma: no mutate class BaseSchema(Mapping): - nullable_name: str raw_schema: Dict[str, Any] = attr.ib() # pragma: no mutate location: Optional[str] = attr.ib(default=None) # pragma: no mutate base_url: Optional[str] = attr.ib(default=None) # pragma: no mutate @@ -57,10 +53,6 @@ def __getitem__(self, item: str) -> CaseInsensitiveDict: def __len__(self) -> int: return len(self.endpoints) - @property # pragma: no mutate - def spec_version(self) -> str: - raise NotImplementedError - @property # pragma: no mutate def verbose_name(self) -> str: raise NotImplementedError @@ -73,13 +65,6 @@ def endpoints(self) -> Dict[str, CaseInsensitiveDict]: self._endpoints = endpoints_to_dict(endpoints) return self._endpoints - @property - def resolver(self) -> ConvertingResolver: - if not hasattr(self, "_resolver"): - # pylint: disable=attribute-defined-outside-init - self._resolver = ConvertingResolver(self.location or "", self.raw_schema, nullable_name=self.nullable_name,) - return self._resolver - @property def endpoints_count(self) -> int: return len(list(self.get_all_endpoints())) @@ -147,10 +132,6 @@ def clone( # pylint: disable=too-many-arguments validate_schema=validate_schema, # type: ignore ) - def get_response_schema(self, definition: Dict[str, Any], scope: str) -> Tuple[List[str], Optional[Dict[str, Any]]]: - """Extract response schema from `responses`.""" - raise NotImplementedError - @deprecated("'register_hook` is deprecated, use `hooks.register' instead") def register_hook(self, place: str, hook: Hook) -> None: warn_deprecated_hook(hook) @@ -182,21 +163,40 @@ def dispatch_hook(self, name: str, context: HookContext, *args: Any, **kwargs: A if local_dispatcher is not None: local_dispatcher.dispatch(name, context, *args, **kwargs) + +class BaseOpenAPISchema(BaseSchema): + nullable_name: str + + @property # pragma: no mutate + def spec_version(self) -> str: + raise NotImplementedError + + def __repr__(self) -> str: + info = self.raw_schema["info"] + return f"{self.__class__.__name__} for {info['title']} ({info['version']})" + + @property + def resolver(self) -> ConvertingResolver: + if not hasattr(self, "_resolver"): + # pylint: disable=attribute-defined-outside-init + self._resolver = ConvertingResolver(self.location or "", self.raw_schema, nullable_name=self.nullable_name) + return self._resolver + def get_content_types(self, endpoint: Endpoint, response: GenericResponse) -> List[str]: """Content types available for this endpoint.""" raise NotImplementedError + def get_response_schema(self, definition: Dict[str, Any], scope: str) -> Tuple[List[str], Optional[Dict[str, Any]]]: + """Extract response schema from `responses`.""" + raise NotImplementedError + -class SwaggerV20(BaseSchema): # pylint: disable=too-many-public-methods +class SwaggerV20(BaseOpenAPISchema): nullable_name = "x-nullable" example_field = "x-example" operations: Tuple[str, ...] = ("get", "put", "post", "delete", "options", "head", "patch") security = SwaggerSecurityProcessor() - def __repr__(self) -> str: - info = self.raw_schema["info"] - return f"{self.__class__.__name__} for {info['title']} ({info['version']})" - @property def spec_version(self) -> str: return self.raw_schema["swagger"] diff --git a/src/schemathesis/specs/openapi/checks.py b/src/schemathesis/specs/openapi/checks.py index d15c7d4d58..f481de18f7 100644 --- a/src/schemathesis/specs/openapi/checks.py +++ b/src/schemathesis/specs/openapi/checks.py @@ -1,3 +1,4 @@ +# pylint: disable=import-outside-toplevel import string from contextlib import ExitStack, contextmanager from itertools import product @@ -37,7 +38,11 @@ def _expand_responses(responses: Dict[Union[str, int], Any]) -> Generator[int, N def content_type_conformance(response: GenericResponse, case: "Case") -> None: - content_types = case.endpoint.get_content_types(response) + from ...schemas import BaseOpenAPISchema + + if not isinstance(case.endpoint.schema, BaseOpenAPISchema): + raise TypeError("This check can be used only with Open API schemas") + content_types = case.endpoint.schema.get_content_types(case.endpoint, response) if not content_types: return content_type = response.headers["Content-Type"] @@ -55,6 +60,10 @@ def content_type_conformance(response: GenericResponse, case: "Case") -> None: def response_schema_conformance(response: GenericResponse, case: "Case") -> None: + from ...schemas import BaseOpenAPISchema + + if not isinstance(case.endpoint.schema, BaseOpenAPISchema): + raise TypeError("This check can be used only with Open API schemas") try: content_type = response.headers["Content-Type"] except KeyError: