refactor: Move logic related to Open API to a separate class

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)

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."""

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"]

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: