Extending Schemathesis

Often you need to modify certain aspects of Schemathesis behavior, adjust data generation, modify requests before sending, and so on. Schemathesis offers multiple extension mechanisms.

Hooks

The hook mechanism is similar to pytest’s. Depending on the scope of the changes, there are three scopes of hooks:

Global. These hooks applied to all schemas in the test run;
Schema. Used only for specific schema instance;
Test. Used only for a particular test function;

To register a new hook function, you need to use special decorators - hook for global and schema-local hooks and hooks.apply for test-specific ones:

import schemathesis

@schemathesis.hook
def before_generate_query(context, strategy):
    return strategy.filter(lambda x: x["id"].isdigit())

schema = schemathesis.from_uri("http://0.0.0.0:8080/swagger.json")

@schema.hook("before_generate_query")
def schema_hook(context, strategy):
    return strategy.filter(lambda x: int(x["id"]) % 2 == 0)

def before_generate_headers(context, strategy):
    return strategy.filter(lambda x: len(x["id"]) > 5)

@schema.hooks.apply(before_generate_headers)
@schema.parametrize()
def test_api(case):
    ...

By default, register functions will check the registered hook name to determine when to run it (see all hook specifications in the section below). Still, to avoid name collisions, you can provide a hook name as an argument to register.

Also, these decorators will check the signature of your hook function to match the specification. Each hook should accept context as the first argument, that provides additional context for hook execution.

Important

Do not mutate context.operation in hook functions as Schemathesis relies on its immutability for caching purposes. Mutating it may lead to unpredictable problems.

Hooks registered on the same scope will be applied in the order of registration. When there are multiple hooks in the same hook location, then the global ones will be applied first.

These hooks can be applied both in CLI and in-code use cases.

`before_generate_*`

This group of six hooks shares the same purpose - adjust data generation for specific request’s part or the whole request.

before_generate_path_parameters
before_generate_headers
before_generate_cookies
before_generate_query
before_generate_body
before_generate_case

They have the same signature that looks like this:

import hypothesis
import schemathesis


def before_generate_query(
    context: schemathesis.hooks.HookContext,
    strategy: hypothesis.strategies.SearchStrategy,
) -> hypothesis.strategies.SearchStrategy:
    pass

The strategy argument is a Hypothesis strategy that will generate a certain request part or the whole request (in case of the before_generate_case hook). For example, your API operation under test expects id query parameter that is a number, and you’d like to have only values that have at least three occurrences of “1”. Then your hook might look like this:

def before_generate_query(context, strategy):
    return strategy.filter(lambda x: str(x["id"]).count("1") >= 3)

To filter or modify the whole request:

def before_generate_case(context, strategy):
    op = context.operation

    def tune_case(case):
        if op.method == "PATCH" and op.path == "/users/{user_id}/":
            case.path_parameters["user_id"] = case.body["data"]["id"]
        return case

    return strategy.map(tune_case)

The example above will modify generated test cases for PATCH /users/{user_id}/ by setting the user_id path parameter to the value generated for payload.

`before_process_path`

This hook is called before each API path is processed (if filters select it). You can use it to modify the schema before processing - set some parameters as constants, update schema syntax, etc.

Let’s say you have the following schema:

/orders/{order_id}:
  get:
    parameters:
      - description: Order ID to retrieve
        in: path
        name: order_id
        required: true
        schema:
          format: int64
          type: integer

Then, with this hook, you can query the database for some existing order and set its ID as a constant in the API operation definition:

import schemathesis
from typing import Any, Dict

database = ...  # Init the DB


def before_process_path(
    context: schemathesis.hooks.HookContext, path: str, methods: Dict[str, Any]
) -> None:
    if path == "/orders/{order_id}":
        order_id = database.get_orders().first().id
        methods["get"]["parameters"][0]["schema"]["const"] = order_id

`before_load_schema`

Called just before schema instance is created. Takes a raw schema representation as a dictionary:

import schemathesis
from typing import Any, Dict


def before_load_schema(
    context: schemathesis.hooks.HookContext,
    raw_schema: Dict[str, Any],
) -> None:
    ...

This hook allows you to modify schema before loading.

`after_load_schema`

Called just after schema instance is created. Takes a loaded schema:

import schemathesis


def after_load_schema(
    context: schemathesis.hooks.HookContext,
    schema: schemathesis.schemas.BaseSchema,
) -> None:
    ...

For example, with this hook you can programmatically add Open API links before tests.

`before_init_operation`

Allows you to modify just initialized API operation:

import schemathesis
from schemathesis.models import APIOperation


def before_init_operation(
    context: schemathesis.hooks.HookContext, operation: APIOperation
) -> None:
    # Overrides the existing schema
    operation.query[0].definition["schema"] = {"enum": [42]}

`before_add_examples`

With this hook, you can add additional test cases that will be executed in Hypothesis explicit phase:

import schemathesis
from schemathesis import Case
from typing import List


def before_add_examples(
    context: schemathesis.hooks.HookContext,
    examples: List[Case],
) -> None:
    examples.append(Case(operation=context.operation, query={"foo": "bar"}))

To load CLI hooks, you need to put them into a separate module and pass an importable path via the SCHEMATHESIS_HOOKS environment variable. For example, you have your hooks definition in myproject/hooks.py, and myproject is importable:

SCHEMATHESIS_HOOKS=myproject.hooks
st run http://127.0.0.1/openapi.yaml

`after_init_cli_run_handlers`

This hook allows you to extend or redefine a list of CLI handlers that will be used to process runner events:

import click
import schemathesis
from schemathesis.cli.handlers import EventHandler
from schemathesis.cli.context import ExecutionContext
from schemathesis.runner import events
from typing import List


class SimpleHandler(EventHandler):
    def handle_event(self, context, event):
        if isinstance(event, events.Finished):
            click.echo("Done!")


@schemathesis.hook
def after_init_cli_run_handlers(
    context: HookContext,
    handlers: List[EventHandler],
    execution_context: ExecutionContext,
) -> None:
    handlers[:] = [SimpleHandler()]

With this simple handler, only Done! will be displayed at the end of the test run. For example, you can use this hook to:

Send events over the network
Store logs in a custom format
Change the output visual style
Display additional information in the output

`add_case`

For each add_case hook and each API operation, we create an additional, duplicate test case. We pass the Case object from the duplicate test to the add_case hook. The user may change the Case object (and therefore the request’s data) before the request is sent to the server. The add_case allows the user to target specific behavior in the API by changing the duplicate request’s specific details.

from schemathesis import Case, GenericResponse, hooks
from typing import Optional


def add_case(
    context: hooks.HookContext, case: Case, response: GenericResponse
) -> Optional[Case]:
    case.headers["Content-Type"] = "application/json"
    return case

Important

The add_case hook works only in CLI.

If you only want to create another case conditionally, you may return None, and no additional test will be created. For example, you may only want to create an additional test case if the original case received a successful response from the server.

from schemathesis import Case, GenericResponse, hooks
from typing import Optional


def add_case(
    context: hooks.HookContext, case: Case, response: GenericResponse
) -> Optional[Case]:
    if 200 <= response.status_code < 300:
        # if the original case was successful, see if an invalid content type header produces a failure
        case.headers["Content-Type"] = "invalid/content/type"
        return case
    else:
        # original case produced non-2xx response, do not create additional test case
        return None

Note: A partial deep copy of the Case object is passed to each add_case hook. Case.operation.app is a reference to the original app, and Case.operation.schema is a shallow copy, so changes to these fields will be reflected in other tests.

`before_call`

Called right before any test request during CLI runs. With this hook, you can modify generated cases in-place:

import schemathesis


@schemathesis.hook
def before_call(context, case):
    case.query = {"q": "42"}

`after_call`

Called right after any successful test request during CLI runs. With this hook, you can inspect (and modify in-place if you want) the received responses and their source cases:

import json
import schemathesis


@schemathesis.hook
def after_call(context, case, response):
    parsed = response.json()
    response._content = json.dumps({"my-wrapper": parsed}).encode()

Important

Won’t be called if request times-out.

Depending on whether you use your Python app in-process, you might get different types for the response argument. For the WSGI case, it will be schemathesis.utils.WSGIResponse.

`process_call_kwargs`

If you want to modify what keyword arguments will be given to case.call / case.call_wsgi / case.call_asgi in CLI, then you can use this hook:

import schemathesis


@schemathesis.hook
def process_call_kwargs(context, case, kwargs):
    kwargs["allow_redirects"] = False

Important

The process_call_kwargs hook works only in CLI.

If you test your app via the real network, then the hook above will disable resolving redirects during network calls. For WSGI integration, the keywords are different. See the documentation for werkzeug.Client.open.

Checks

Schemathesis provides a way to check app responses via user-defined functions called “checks”. Each check is a function that accepts two arguments:

def my_check(response, case):
    ...

The first one is the app response, which is requests.Response or schemathesis.utils.WSGIResponse, depending on whether you used the WSGI integration or not. The second one is the Case instance that was used to send data to the tested application.

To indicate a failure, you need to raise AssertionError explicitly:

def my_check(response, case):
    if response.text == "I am a teapot":
        raise AssertionError("It is a teapot!")

If the assertion fails, you’ll see the assertion message in Schemathesis output. In the case of missing assertion message, Schemathesis will report “Check my_check failed”.

Note

If you use the assert statement and pytest as the test runner, then pytest may rewrite assertions which affects error messages.

Custom string strategies

Open API allows you to set a custom string format for a property via the format keyword. For example, you may use the card_number format and validate input with the Luhn algorithm.

You can teach Schemathesis to generate values that fit this format by registering a custom Hypothesis strategy:

Create a Hypothesis strategy that generates valid string values
Register it via schemathesis.openapi.format

from hypothesis import strategies as st
import schemathesis

strategy = st.from_regex(r"\A4[0-9]{15}\Z").filter(luhn_validator)
schemathesis.openapi.format("visa_cards", strategy)

Schemathesis test runner

If you’re looking for a way to extend Schemathesis or reuse it in your own application, then the runner module might help you. It can run tests against the given schema URI and will do some simple checks for you.

import schemathesis

schema = schemathesis.from_uri("http://127.0.0.1:8080/swagger.json")

runner = schemathesis.runner.from_schema(schema)
for event in runner.execute():
    ...  # do something with event

runner.execute creates a generator that yields events of different kinds - BeforeExecution, AfterExecution, etc. They provide a lot of useful information about what happens during tests, but your responsibility is handling these events. You can take some inspiration from Schemathesis CLI implementation. See the full description of events in the source code.

You can provide your custom checks to the execute function; the check is a callable that accepts one argument of requests.Response type.

from datetime import timedelta
from schemathesis import runner, models


def not_too_long(response, case: models.Case):
    assert response.elapsed < timedelta(milliseconds=300)


schema = schemathesis.from_uri("http://127.0.0.1:8080/swagger.json")
runner = schemathesis.runner.from_schema(schema, checks=[not_too_long])
for event in runner.execute():
    ...  # do something with event

Extending Schemathesis

Hooks

before_generate_*

before_process_path

before_load_schema

after_load_schema

before_init_operation

before_add_examples

after_init_cli_run_handlers

add_case

before_call

after_call

process_call_kwargs

Checks

Custom string strategies

Schemathesis test runner

`before_generate_*`

`before_process_path`

`before_load_schema`

`after_load_schema`

`before_init_operation`

`before_add_examples`

`after_init_cli_run_handlers`

`add_case`

`before_call`

`after_call`

`process_call_kwargs`