feat: Third-party compatibility fixups mechanism

docs/changelog.rst

@@ -13,6 +13,7 @@ Added
   for hook functions that do not accept this argument.
 - A new hook system that allows generic hook dispatching. It comes with new hook locations. For more details see "Customization" section in our documentation.
 - New ``before_process_path`` hook.
+- Third-party compatibility fixups mechanism. Currently there is one fixup for `FastAPI <https://github.com/tiangolo/fastapi>`_. `#503`_
 
 Deprecated
 ~~~~~~~~~~
@@ -1008,6 +1009,7 @@ Fixed
 .. _#519: https://github.com/kiwicom/schemathesis/issues/519
 .. _#513: https://github.com/kiwicom/schemathesis/issues/513
 .. _#504: https://github.com/kiwicom/schemathesis/issues/504
+.. _#503: https://github.com/kiwicom/schemathesis/issues/503
 .. _#499: https://github.com/kiwicom/schemathesis/issues/499
 .. _#497: https://github.com/kiwicom/schemathesis/issues/497
 .. _#496: https://github.com/kiwicom/schemathesis/issues/496

docs/compatibility.rst

@@ -0,0 +1,30 @@
+.. _compatibility:
+
+Compatibility
+=============
+
+By default, Schemathesis is strict on Open API spec interpretation, but other 3rd-party tools often are more flexible
+and not always comply with the spec.
+
+FastAPI
+-------
+
+`FastAPI <https://github.com/tiangolo/fastapi>`_ uses `pydantic <https://github.com/samuelcolvin/pydantic>`_ for JSON Schema
+generation and it produces Draft 7 compatible schemas. But Open API 2 / 3.0.x use earlier versions of JSON Schema (Draft 4 and Wright Draft 00 respectively) which leads
+to incompatibilities when Schemathesis parses input schema.
+
+This is a `known issue <https://github.com/tiangolo/fastapi/issues/240>`_ on the FastAPI side
+and Schemathesis provides a way to handle such schemas. The idea is based on converting Draft 7 keywords syntax to Draft 4.
+
+To use it you need to add this code before you load your schema with Schemathesis:
+
+.. code:: python
+
+    import schemathesis
+
+    # will install all available compatibility fixups.
+    schemathesis.fixups.install()
+    # You can also provide a list of fixup names as the first argument
+    # schemathesis.fixups.install(["fastapi"])
+
+If you use CLI, then you can utilize ``--fixups=all`` option.

docs/faq.rst

@@ -54,6 +54,24 @@ The ``case`` object, that is injected in each test can be modified, assuming you
     def test_api(case):
         case.path_parameters["user_id"] = 42
 
+Why does Schemathesis fail to parse my API schema generate by FastAPI?
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Because FastAPI uses JSON Draft 7 under the hood (via ``pydantic``) which is not compatible with JSON drafts defined by
+the Open API 2 / 3.0.x versions. It is a `known issue <https://github.com/tiangolo/fastapi/issues/240>`_ on the FastAPI side.
+Schemathesis is more strict in schema handling by default, but we provide optional fixups for this case:
+
+.. code:: python
+
+    import schemathesis
+
+    # will install all available compatibility fixups.
+    schemathesis.fixups.install()
+    # You can also provide a list of fixup names as the first argument
+    # schemathesis.fixups.install(["fastapi"])
+
+For more information, take a look into the "Compatibility" section.
+
 Working with API schemas
 ------------------------
 

docs/index.rst

@@ -9,6 +9,7 @@ Welcome to schemathesis's documentation!
    :caption: Contents:
 
    usage
+   compatibility
    customization
    targeted
    faq

src/schemathesis/__init__.py

@@ -1,4 +1,4 @@
-from . import hooks
+from . import fixups, hooks
 from ._hypothesis import init_default_strategies, register_string_format
 from .cli import register_check
 from .constants import __version__

src/schemathesis/cli/__init__.py

@@ -10,6 +10,7 @@
 
 from .. import checks as checks_module
 from .. import models, runner
+from ..fixups import ALL_FIXUPS
 from ..runner import events
 from ..runner.targeted import DEFAULT_TARGETS_NAMES, Target
 from ..types import Filter
@@ -151,7 +152,13 @@ def schemathesis(pre_run: Optional[str] = None) -> None:
     is_eager=True,
     default=False,
 )
-@click.option("--store-network-log", help="Store requests and responses into a file", type=click.File("w"))
+@click.option("--store-network-log", help="Store requests and responses into a file.", type=click.File("w"))
+@click.option(
+    "--fixups",
+    help="Install specified compatibility fixups.",
+    multiple=True,
+    type=click.Choice(list(ALL_FIXUPS) + ["all"]),
+)
 @click.option(
     "--hypothesis-deadline",
     help="Duration in milliseconds that each individual example with a test is not allowed to exceed.",
@@ -199,6 +206,7 @@ def run(  # pylint: disable=too-many-arguments
     junit_xml: Optional[click.utils.LazyFile] = None,
     show_errors_tracebacks: bool = False,
     store_network_log: Optional[click.utils.LazyFile] = None,
+    fixups: Tuple[str] = (),  # type: ignore
     hypothesis_deadline: Optional[Union[int, NotSet]] = None,
     hypothesis_derandomize: Optional[bool] = None,
     hypothesis_max_examples: Optional[int] = None,
@@ -238,6 +246,7 @@ def run(  # pylint: disable=too-many-arguments
         targets=selected_targets,
         workers_num=workers_num,
         validate_schema=validate_schema,
+        fixups=fixups,
         hypothesis_deadline=hypothesis_deadline,
         hypothesis_derandomize=hypothesis_derandomize,
         hypothesis_max_examples=hypothesis_max_examples,

src/schemathesis/converter.py

@@ -1,5 +1,7 @@
 from copy import deepcopy
-from typing import Any, Dict, List, Union, overload
+from typing import Any, Dict
+
+from .utils import traverse_schema
 
 
 def to_json_schema(schema: Dict[str, Any], nullable_name: str) -> Dict[str, Any]:
@@ -25,40 +27,10 @@ def to_json_schema(schema: Dict[str, Any], nullable_name: str) -> Dict[str, Any]
     return schema
 
 
-Schema = Union[Dict[str, Any], List, str, float, int]
-
-
-@overload
 def to_json_schema_recursive(schema: Dict[str, Any], nullable_name: str) -> Dict[str, Any]:
-    pass
-
-
-@overload
-def to_json_schema_recursive(schema: List, nullable_name: str) -> List:
-    pass
-
-
-@overload
-def to_json_schema_recursive(schema: str, nullable_name: str) -> str:
-    pass
-
-
-@overload
-def to_json_schema_recursive(schema: float, nullable_name: str) -> float:
-    pass
-
-
-def to_json_schema_recursive(schema: Schema, nullable_name: str) -> Schema:
     """Apply ``to_json_schema`` recursively.
 
     This version is needed for cases where the input schema was not resolved and ``to_json_schema`` wasn't applied
     recursively.
     """
-    if isinstance(schema, dict):
-        schema = to_json_schema(schema, nullable_name)
-        for key, sub_item in schema.items():
-            schema[key] = to_json_schema_recursive(sub_item, nullable_name)
-    elif isinstance(schema, list):
-        for idx, sub_item in enumerate(schema):
-            schema[idx] = to_json_schema_recursive(sub_item, nullable_name)
-    return schema
+    return traverse_schema(schema, to_json_schema, nullable_name)

src/schemathesis/fixups/__init__.py

@@ -0,0 +1,25 @@
+from typing import Iterable, List, Optional
+
+from . import fast_api
+
+ALL_FIXUPS = {"fast_api": fast_api}
+
+
+def install(fixups: Optional[Iterable[str]] = None) -> None:
+    """Install fixups.
+
+    Without the first argument installs all available fixups.
+    """
+    fixups = fixups or list(ALL_FIXUPS.keys())
+    for name in fixups:
+        ALL_FIXUPS[name].install()  # type: ignore
+
+
+def uninstall(fixups: Optional[Iterable[str]] = None) -> None:
+    """Uninstall fixups.
+
+    Without the first argument uninstalls all available fixups.
+    """
+    fixups = fixups or list(ALL_FIXUPS.keys())
+    for name in fixups:
+        ALL_FIXUPS[name].uninstall()  # type: ignore

src/schemathesis/fixups/fast_api.py

@@ -0,0 +1,30 @@
+from typing import Any, Dict
+
+from ..hooks import HookContext, register, unregister
+from ..utils import traverse_schema
+
+
+def install() -> None:
+    register(before_load_schema)
+
+
+def uninstall() -> None:
+    unregister(before_load_schema)
+
+
+def before_load_schema(context: HookContext, schema: Dict[str, Any]) -> None:
+    traverse_schema(schema, _handle_boundaries)
+
+
+def _handle_boundaries(schema: Dict[str, Any]) -> Dict[str, Any]:
+    """Convert Draft 7 keywords to Draft 4 compatible versions.
+
+    FastAPI uses ``pydantic``, which generates Draft 7 compatible schemas.
+    """
+    for boundary_name, boundary_exclusive_name in (("maximum", "exclusiveMaximum"), ("minimum", "exclusiveMinimum")):
+        value = schema.get(boundary_exclusive_name)
+        # `bool` check is needed, since in Python `True` is an instance of `int`
+        if isinstance(value, (int, float)) and not isinstance(value, bool):
+            schema[boundary_exclusive_name] = True
+            schema[boundary_name] = value
+    return schema

src/schemathesis/hooks.py

@@ -131,6 +131,12 @@ def dispatch(self, name: str, context: HookContext, *args: Any, **kwargs: Any) -
         for hook in self.get_hooks(name):
             hook(context, *args, **kwargs)
 
+    def unregister(self, hook: Callable) -> None:
+        """Unregister a specific hook."""
+        # It removes this function from all places
+        for hooks in self._hooks.values():
+            hooks[:] = [item for item in hooks if item is not hook]
+
     def unregister_all(self) -> None:
         """Remove all registered hooks.
 
@@ -174,9 +180,15 @@ def before_process_path(context: HookContext, path: str, methods: Dict[str, Any]
     pass
 
 
+@HookDispatcher.register_spec
+def before_load_schema(context: HookContext, raw_schema: Dict[str, Any]) -> None:
+    pass
+
+
 GLOBAL_HOOK_DISPATCHER = HookDispatcher()
 dispatch = GLOBAL_HOOK_DISPATCHER.dispatch
 get_hooks = GLOBAL_HOOK_DISPATCHER.get_hooks
+unregister = GLOBAL_HOOK_DISPATCHER.unregister
 unregister_all = GLOBAL_HOOK_DISPATCHER.unregister_all
 
 

src/schemathesis/loaders.py

@@ -12,6 +12,7 @@
 from . import spec_schemas
 from .constants import USER_AGENT
 from .exceptions import HTTPError
+from .hooks import HookContext, dispatch
 from .lazy import LazySchema
 from .schemas import BaseSchema, OpenApi30, SwaggerV20
 from .types import Filter, PathLike
@@ -115,6 +116,7 @@ def from_dict(
     validate_schema: bool = True,
 ) -> BaseSchema:
     """Get a proper abstraction for the given raw schema."""
+    dispatch("before_load_schema", HookContext(), raw_schema)
     if "swagger" in raw_schema:
         _maybe_validate_schema(raw_schema, spec_schemas.SWAGGER_20, validate_schema)
         return SwaggerV20(

src/schemathesis/runner/__init__.py

@@ -3,6 +3,7 @@
 
 import hypothesis.errors
 
+from .. import fixups as _fixups
 from .. import loaders
 from ..checks import DEFAULT_CHECKS
 from ..models import CheckFunction
@@ -24,6 +25,7 @@ def prepare(  # pylint: disable=too-many-arguments
     seed: Optional[int] = None,
     exit_first: bool = False,
     store_interactions: bool = False,
+    fixups: Iterable[str] = (),
     # Schema loading
     loader: Callable = loaders.from_uri,
     base_url: Optional[str] = None,
@@ -82,6 +84,7 @@ def prepare(  # pylint: disable=too-many-arguments
         headers=headers,
         request_timeout=request_timeout,
         store_interactions=store_interactions,
+        fixups=fixups,
     )
 
 
@@ -125,13 +128,20 @@ def execute_from_schema(
     seed: Optional[int] = None,
     exit_first: bool = False,
     store_interactions: bool = False,
+    fixups: Iterable[str] = (),
 ) -> Generator[events.ExecutionEvent, None, None]:
     """Execute tests for the given schema.
 
     Provides the main testing loop and preparation step.
     """
     # pylint: disable=too-many-locals
     try:
+        if fixups:
+            if "all" in fixups:
+                _fixups.install()
+            else:
+                _fixups.install(fixups)
+
         if app is not None:
             app = import_app(app)
         schema = load_schema(

src/schemathesis/utils.py

@@ -6,7 +6,7 @@
 import warnings
 from contextlib import contextmanager
 from functools import wraps
-from typing import Any, Callable, Dict, Generator, List, Optional, Set, Tuple, Type, Union
+from typing import Any, Callable, Dict, Generator, List, Optional, Set, Tuple, Type, Union, overload
 from urllib.parse import urlsplit, urlunsplit
 
 import requests
@@ -196,3 +196,38 @@ def import_app(path: str) -> Any:
     # may return a parent module (system dependent)
     module = sys.modules[path]
     return getattr(module, name)
+
+
+Schema = Union[Dict[str, Any], List, str, float, int]
+
+
+@overload
+def traverse_schema(schema: Dict[str, Any], callback: Callable, *args: Any, **kwargs: Any) -> Dict[str, Any]:
+    pass
+
+
+@overload
+def traverse_schema(schema: List, callback: Callable, *args: Any, **kwargs: Any) -> List:
+    pass
+
+
+@overload
+def traverse_schema(schema: str, callback: Callable, *args: Any, **kwargs: Any) -> str:
+    pass
+
+
+@overload
+def traverse_schema(schema: float, callback: Callable, *args: Any, **kwargs: Any) -> float:
+    pass
+
+
+def traverse_schema(schema: Schema, callback: Callable[..., Dict[str, Any]], *args: Any, **kwargs: Any) -> Schema:
+    """Apply callback recursively to the given schema."""
+    if isinstance(schema, dict):
+        schema = callback(schema, *args, **kwargs)
+        for key, sub_item in schema.items():
+            schema[key] = traverse_schema(sub_item, callback, *args, **kwargs)
+    elif isinstance(schema, list):
+        for idx, sub_item in enumerate(schema):
+            schema[idx] = traverse_schema(sub_item, callback, *args, **kwargs)
+    return schema