Add fuzz testing for HTTP API with Schemathesis

[lobkovilya]

Integrate Schemathesis to perform fuzz testing based on the OpenAPI spec. This will help identify edge cases, and unexpected failures, and improve the robustness of the HTTP API. The test setup should be automated and ideally integrated into CI.

I've done a small POC using python (as I didn't find a way to set mesh: "default" by using CLI):

import schemathesis

# Load the OpenAPI schema from the correct path
schema = schemathesis.from_path("docs/generated/openapi.yaml")

@schema.parametrize()
def test_api(case):
    # Ensure case.path_parameters exists before overriding
    if case.path_parameters is not None and "mesh" in case.path_parameters:
        case.path_parameters["mesh"] = "default"  # Override the "mesh" parameter
    
    # Call the API with the correct base URL
    response = case.call(base_url="http://localhost:5681")
    
    # Validate the response
    case.validate_response(response)

Run:

pytest test_api.py

It returns a bunch of small and easy-to-fix problems:

Response violates schema: 'total' is a required property

___________________________________________________________________________________________ test_api[GET /global-insight] ____________________________________________________________________________________________

  @wraps(test)
>   def test_function(*args: Any, **kwargs: Any) -> Any:

schemathesis-env/lib/python3.13/site-packages/schemathesis/_hypothesis.py:81:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

case = Case()

  @schema.parametrize()
  def test_api(case):
      # Ensure case.path_parameters exists before overriding
      if case.path_parameters is not None and "mesh" in case.path_parameters:
          case.path_parameters["mesh"] = "default"  # Override the "mesh" parameter

      # Call the API with the correct base URL
      response = case.call(base_url="http://localhost:5681")

      # Validate the response
>       case.validate_response(response)
E       schemathesis.exceptions.CheckFailed:
E
E       1. Response violates schema
E
E           'total' is a required property
E
E           Schema at /allOf/0/properties/services/allOf/0:
E
E               {
E                   "type": "object",
E                   "title": "ServicesStats",
E                   "description": "Services statistics",
E                   "required": [
E                       "total",
E                       "internal",
E                       "external",
E                       "gatewayBuiltin",
E                       "gatewayDelegated"
E                   ],
E                   "properties": {
E                       "internal": {
E                           "description": "Internal services statistics",
E                           "allOf": [
E                               {
E                                   "$ref": "#/components/schemas/FullStatus"
E                               }
E                           ]
E                   // Output truncated...
E               }
E
E           Value:
E
E               {
E                   "external": {
E                       "total": 0
E                   },
E                   "gatewayBuiltin": {
E                       "offline": 0,
E                       "online": 0,
E                       "partiallyDegraded": 0,
E                       "total": 0
E                   },
E                   "gatewayDelegated": {
E                       "offline": 0,
E                       "online": 0,
E                       "partiallyDegraded": 0,
E                       "total": 0
E                   },
E                   "internal": {
E                       "offline": 0,
E                       "online": 0,
E                   // Output truncated...
E               }
E
E       [200] OK:
E
E           `{
E            "createdAt": "2025-02-22T00:44:05.766728+01:00",
E            "dataplanes": {
E             "gatewayBuiltin": {
E              "offline": 0,
E              "online": 0,
E              "partiallyDegraded": 0,
E              "total": 0
E             },
E             "gatewayDelegated": {
E              "offline": 0,
E              "online": 0,
E              "partiallyDegraded": 0,
E              "total": 0
E             },
E             "standard": {
E              "offline": 0,
E              "online": 0,
E              "partiallyDegraded": 0,
E              "total": 0
E             }
E            },
E            "meshes": {
E             "total": 1
E            },
E            "policies": {
E             "total": 4
E            },
E            "resources": {
E             "MeshCircuitBreaker": {
E              "total": 1
E             },
E             "MeshRetry": {
E             // Output truncated...`
E
E       Reproduce with:
E
E           curl -X GET http://localhost/global-insight

test_api.py:16: CheckFailed

Undocumented HTTP status code

_________________________________________________________________________ test_api[GET /meshes/{mesh}/{resourceType}/{resourceName}/_rules] __________________________________________________________________________

  @wraps(test)
>   def test_function(*args: Any, **kwargs: Any) -> Any:

schemathesis-env/lib/python3.13/site-packages/schemathesis/_hypothesis.py:81:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
schemathesis-env/lib/python3.13/site-packages/hypothesis/core.py:1478: in _raise_to_user
  raise the_error_hypothesis_found
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

case = Case(path_parameters={'mesh': 'default', 'resourceType': 'dataplanes', 'resourceName': 'my-dp'})

  @schema.parametrize()
  def test_api(case):
      # Ensure case.path_parameters exists before overriding
      if case.path_parameters is not None and "mesh" in case.path_parameters:
          case.path_parameters["mesh"] = "default"  # Override the "mesh" parameter

      # Call the API with the correct base URL
      response = case.call(base_url="http://localhost:5681")

      # Validate the response
>       case.validate_response(response)
E       schemathesis.exceptions.CheckFailed:
E
E       1. Undocumented HTTP status code
E
E           Received: 404
E           Documented: 200, 400, 500
E
E       [404] Not Found:
E
E           `{
E            "type": "/std-errors",
E            "status": 404,
E            "title": "Could not retrieve Dataplane",
E            "detail": "Not found",
E            "details": "Not found"
E           }`
E
E       Reproduce with:
E
E           curl -X GET http://localhost/meshes/default/dataplanes/my-dp/_rules
E
E       Falsifying explicit example: test_api(
E           case=,
E       )

test_api.py:16: CheckFailed

Undocumented Content-Type

_________________________________________________________________________________ test_api[GET /meshes/{mesh}/meshaccesslogs/{name}] _________________________________________________________________________________

  @wraps(test)
>   def test_function(*args: Any, **kwargs: Any) -> Any:

schemathesis-env/lib/python3.13/site-packages/schemathesis/_hypothesis.py:81:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

case = Case(path_parameters={'mesh': 'default', 'name': '0'})

  @schema.parametrize()
  def test_api(case):
      # Ensure case.path_parameters exists before overriding
      if case.path_parameters is not None and "mesh" in case.path_parameters:
          case.path_parameters["mesh"] = "default"  # Override the "mesh" parameter

      # Call the API with the correct base URL
      response = case.call(base_url="http://localhost:5681")

      # Validate the response
>       case.validate_response(response)
E       schemathesis.exceptions.CheckFailed:
E
E       1. Undocumented Content-Type
E
E           Received: application/json
E           Documented: application/problem+json
E
E       2. Response violates schema
E
E           'instance' is a required property
E
E           Schema at /allOf/0:
E
E               {
E                   "type": "object",
E                   "title": "Error",
E                   "description": "standard error",
E                   "x-examples": {
E                       "Example 1": {
E                           "status": 404,
E                           "title": "Not Found",
E                           "type": "https://kongapi.info/konnect/not-found",
E                           "instance": "portal:trace:2287285207635123011",
E                           "detail": "The requested document was not found"
E                       }
E                   },
E                   "required": [
E                       "status",
E                       "title",
E                       "instance"
E                   ],
E                   "properties": {
E                   // Output truncated...
E               }
E
E           Value:
E
E               {
E                   "type": "/std-errors",
E                   "status": 404,
E                   "title": "Could not retrieve a resource",
E                   "detail": "Not found",
E                   "details": "Not found"
E               }
E
E       [404] Not Found:
E
E           `{
E            "type": "/std-errors",
E            "status": 404,
E            "title": "Could not retrieve a resource",
E            "detail": "Not found",
E            "details": "Not found"
E           }`
E
E       Reproduce with:
E
E           curl -X GET http://localhost/meshes/default/meshaccesslogs/0

test_api.py:16: CheckFailed

Response violates schema: None is not of type 'string'

_________________________________________________________________________________ test_api[GET /meshes/{mesh}/meshmultizoneservices] _________________________________________________________________________________

    @wraps(test)
>   def test_function(*args: Any, **kwargs: Any) -> Any:

schemathesis-env/lib/python3.13/site-packages/schemathesis/_hypothesis.py:81:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
schemathesis-env/lib/python3.13/site-packages/hypothesis/core.py:1478: in _raise_to_user
    raise the_error_hypothesis_found
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

case = Case(path_parameters={'mesh': 'default'}, query={'offset': 0, 'filter': {'label.k8s.kuma.io/namespace': 'my-ns'}})

    @schema.parametrize()
    def test_api(case):
        # Ensure case.path_parameters exists before overriding
        if case.path_parameters is not None and "mesh" in case.path_parameters:
            case.path_parameters["mesh"] = "default"  # Override the "mesh" parameter

        # Call the API with the correct base URL
        response = case.call(base_url="http://localhost:5681")

        # Validate the response
>       case.validate_response(response)
E       schemathesis.exceptions.CheckFailed:
E
E       1. Response violates schema
E
E           None is not of type 'string'
E
E           Schema at /properties/next:
E
E               {
E                   "type": "string",
E                   "description": "URL to the next page"
E               }
E
E           Value:
E
E               null
E
E       [200] OK:
E
E           `{
E            "total": 0,
E            "items": [],
E            "next": null
E           }`
E
E       Reproduce with:
E
E           curl -X GET 'http://localhost/meshes/default/meshmultizoneservices?offset=0&filter=label.k8s.kuma.io%2Fnamespace'
E
E       Falsifying explicit example: test_api(
E           case=,
E       )

test_api.py:16: CheckFailed

[Stranger6667]

@lobkovilya FYI, in CLI you could set mesh=default with --set-path 'mesh=default'

[lobkovilya]

Hi @Stranger6667, I tried this one:

schemathesis run --hypothesis-verbosity=debug --base-url http://localhost:5681 --include-path "/meshes/{mesh}/meshservices/{name}" --exclude-method "DELETE" --set-path "mesh=default" docs/generated/openapi.yaml

And I see all the tests have path_parameters=,:

Trying example: network_test(
    ctx=,
    checks=(not_a_server_error,),
    targets=(),
    result=,
    errors=[],
    headers={},
    data_generation_methods=[<DataGenerationMethod.positive: 'positive'>],
    feedback=Feedback(stateful=Stateful.links,
     operation=APIOperation(path='/meshes/{mesh}/meshservices/{name}',
      method='put',
      definition=,
      schema=,
      verbose_name='PUT /meshes/{mesh}/meshservices/{name}',
      app=None,
      base_url='http://localhost:5681',
      path_parameters=,
      headers=,
      cookies=,
      query=,
      body=,
      case_cls=),
     stateful_tests={}),
    max_response_time=None,
    session=<requests.sessions.Session object at 0x106867b10>,
    request_config=,
    store_interactions=False,
    dry_run=False,
    case=,
)

I am not sure if I'm debugging this correctly, but I also see:

WARNINGS:
  - `GET /meshes/{mesh}/meshservices/{name}` returned only 4xx responses during unit tests. Check base URL or adjust data generation settings
  - `PUT /meshes/{mesh}/meshservices/{name}` returned only 4xx responses during unit tests. Check base URL or adjust data generation settings

And I see no new resources were created on the server after PUT.

[Stranger6667]

I think it would be better to use --casette-path=vcr.yaml which will store network requests in a VCR-compatible format and inspect it there. Schemathesis v4 removes --hypothesis-verbosity as it exposes quite a lot of not so useful info.

But otherwise, mesh should be used globally in all tests (this limitation will soon go away)

P.S. I don't want to spam too much here, just want to make sure that CLI would work for your use case. Let me know if there is a better place to discuss / debug this issue.