fix: Not serializing shared parameters for an API operation

Signed-off-by: Dmitry Dygalo <dmitry@dygalo.dev>

docs/changelog.rst

@@ -4,6 +4,10 @@ Changelog
 :version:`Unreleased <v3.28.1...HEAD>` - TBD
 --------------------------------------------
 
+**Fixed**
+
+- Not serializing shared parameters for an API operation.
+
 .. _v3.28.1:
 
 :version:`3.28.1 <v3.28.0...v3.28.1>` - 2024-05-11

src/schemathesis/specs/openapi/schemas.py

@@ -437,7 +437,7 @@ def get_case_strategy(
         )
 
     def get_parameter_serializer(self, operation: APIOperation, location: str) -> Callable | None:
-        definitions = [item for item in operation.definition.resolved.get("parameters", []) if item["in"] == location]
+        definitions = [item.definition for item in operation.iter_parameters() if item.location == location]
         security_parameters = self.security.get_security_definitions_as_parameters(
             self.raw_schema, operation, self.resolver, location
         )

test/specs/openapi/randomized/factory.py

@@ -0,0 +1,24 @@
+from dataclasses import is_dataclass, fields
+
+from .types import Missing
+
+
+def _asdict_inner(obj):
+    if is_dataclass(obj):
+        result = {}
+        for f in fields(obj):
+            value = getattr(obj, f.name)
+            if not isinstance(value, Missing):
+                result[f.name] = _asdict_inner(value)
+        if hasattr(obj, "map_value"):
+            result = obj.map_value(result)
+        return result
+    elif isinstance(obj, (list, tuple)):
+        return type(obj)(_asdict_inner(v) for v in obj)
+    elif isinstance(obj, dict):
+        return type(obj)((_asdict_inner(k), _asdict_inner(v)) for k, v in obj.items())
+    return obj
+
+
+def asdict(schema):
+    return _asdict_inner(schema)

test/specs/openapi/randomized/test_randomized.py

@@ -0,0 +1,27 @@
+from hypothesis import given
+from hypothesis import strategies as st, settings, Phase
+
+import pytest
+import schemathesis
+import schemathesis.runner
+import sys
+from schemathesis.runner import events
+
+if sys.version_info < (3, 10):
+    pytest.skip("Required Python 3.10+", allow_module_level=True)
+
+from .factory import asdict
+from .v2 import Swagger
+
+
+@given(st.from_type(Swagger).map(asdict))
+@settings(max_examples=25, phases=[Phase.generate], deadline=None)
+def test_swagger(schema):
+    schema = schemathesis.from_dict(schema, validate_schema=True)
+    schema.validate_schema = False
+    for event in schemathesis.runner.from_schema(schema, dry_run=True).execute():
+        assert not isinstance(event, events.InternalError), repr(event)
+        if isinstance(event, events.AfterExecution):
+            if event.result.errors:
+                error = event.result.errors[0]
+                raise AssertionError(error.exception_with_traceback)

test/specs/openapi/randomized/types.py

@@ -0,0 +1,28 @@
+from typing import Annotated
+
+from hypothesis import strategies as st
+
+
+class Pattern:
+    def __class_getitem__(cls, pattern: str):
+        return Annotated[str, st.from_regex(pattern)]
+
+
+class UniqueList:
+    def __class_getitem__(cls, inner: type):
+        return Annotated[list, st.lists(st.from_type(inner), unique=True)]
+
+
+class CombinedDict:
+    def __class_getitem__(cls, args):
+        keys, values, defaults = args
+
+        def update(d):
+            d.update(defaults)
+            return d
+
+        return Annotated[dict, st.dictionaries(st.from_type(keys), st.from_type(values)).map(update)]
+
+
+class Missing:
+    pass

test/specs/openapi/randomized/v2.py

@@ -0,0 +1,202 @@
+# ruff: noqa: F722, F821
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Literal
+
+from .types import CombinedDict, Missing, Pattern, UniqueList
+
+RESPONSE_SUCCESS = {"description": "OK"}
+PATH_ITEM_SAMPLE = {"get": {"responses": {"default": RESPONSE_SUCCESS}}}
+PARAMETER_SAMPLE = {"name": "sample", "in": "query", "type": "string"}
+ResponseId = Pattern["^([0-9]{3})$|^(default)$"]
+
+
+@dataclass
+class PathItemReference:
+    def map_value(self, value):
+        value["$ref"] = "#/x-paths/Entry"
+        return value
+
+
+@dataclass
+class ParameterReference:
+    def map_value(self, value):
+        value["$ref"] = "#/parameters/SampleParameter"
+        return value
+
+
+@dataclass
+class ResponseReference:
+    def map_value(self, value):
+        value["$ref"] = "#/responses/Success"
+        return value
+
+
+MediaType = (
+    Literal["application/json"]
+    | Literal["application/xml"]
+    | Literal["text/plain"]
+    | Literal["text/yaml"]
+    | Literal["application/x-www-form-urlencoded"]
+    | Literal["multipart/form-data"]
+    | Literal["application/octet-stream"]
+)
+
+
+@dataclass
+class Swagger:
+    swagger: Literal["2.0"]
+    info: Info
+    host: Pattern["^[^{}/ :\\\\]+(?::\\d+)?$"] | Missing
+    basePath: Pattern["^/"] | Missing
+    paths: dict[Pattern["^/"], PathItem | PathItemReference]
+    consumes: MediaTypeList | Missing
+    produces: MediaTypeList | Missing
+    parameters: CombinedDict[str, Parameter, {"SampleParameter": PARAMETER_SAMPLE}]
+    responses: CombinedDict[ResponseId, Response, {"Success": RESPONSE_SUCCESS}]
+    definitions: Definitions
+
+    def map_value(self, value):
+        value["x-paths"] = {"Entry": PATH_ITEM_SAMPLE}
+        return value
+
+
+@dataclass
+class Definitions:
+    pass
+
+
+@dataclass
+class Info:
+    version: Literal["1.0.0"]
+    title: Literal["Example API"]
+
+
+@dataclass
+class PathItem:
+    get: Operation | Missing
+    put: Operation | Missing
+    post: Operation | Missing
+    delete: Operation | Missing
+    options: Operation | Missing
+    head: Operation | Missing
+    patch: Operation | Missing
+    parameters: ParameterList | Missing
+
+    def map_value(self, value):
+        if "parameters" in value:
+            value["parameters"] = deduplicate_parameters(value["parameters"])
+        return value
+
+
+@dataclass
+class Operation:
+    operationId: str | Missing
+    parameters: ParameterList | Missing
+    tags: UniqueList[str] | Missing
+    consumes: MediaTypeList | Missing
+    produces: MediaTypeList | Missing
+    responses: dict[ResponseId, Response]
+
+    def map_value(self, value):
+        if not value["responses"]:
+            value["responses"] = {"default": RESPONSE_SUCCESS}
+        if "parameters" in value:
+            value["parameters"] = deduplicate_parameters(value["parameters"])
+        return value
+
+
+def deduplicate_parameters(parameters):
+    parameter_names = set()
+    deduplicated = []
+    has_reference = False
+    for parameter in parameters:
+        if "$ref" in parameter:
+            if has_reference:
+                continue
+            has_reference = True
+            deduplicated.append(parameter)
+        else:
+            name = parameter["name"]
+            if name in parameter_names:
+                continue
+            parameter_names.add(name)
+            deduplicated.append(parameter)
+    return deduplicated
+
+
+@dataclass
+class Response:
+    description: Literal["Ok"]
+
+
+@dataclass
+class BodyParameter:
+    name: str
+    schema: dict = field(default_factory=dict)
+
+    def map_value(self, value):
+        value["in"] = "body"
+        return value
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+@dataclass
+class QueryParameter:
+    name: str
+    type: Literal["string", "number", "boolean", "integer", "array"]
+
+    def map_value(self, value):
+        value["in"] = "query"
+        return value
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+@dataclass
+class HeaderParameter:
+    name: str
+    type: Literal["string", "number", "boolean", "integer", "array"]
+
+    def map_value(self, value):
+        value["in"] = "header"
+        return value
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+@dataclass
+class PathParameter:
+    name: str
+    type: Literal["string", "number", "boolean", "integer", "array"]
+
+    def map_value(self, value):
+        value["in"] = "path"
+        value["required"] = True
+        return value
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+@dataclass
+class FormDataParameter:
+    name: str
+    type: Literal["string", "number", "boolean", "integer", "array", "file"]
+
+    def map_value(self, value):
+        value["in"] = "formData"
+        return value
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+Parameter = BodyParameter | QueryParameter | HeaderParameter | PathParameter | FormDataParameter
+ParameterList = list[Parameter | ParameterReference]
+MediaTypeList = UniqueList[MediaType]

test/specs/openapi/test_hypothesis.py

@@ -383,6 +383,32 @@ def test(case):
     mocked.assert_called()
 
 
+def test_serializing_shared_header_parameters():
+    raw_schema = {
+        "swagger": "2.0",
+        "info": {"version": "1.0.0", "title": "Example API"},
+        "paths": {
+            "/data": {
+                "get": {
+                    "responses": {"default": {"description": "Ok"}},
+                },
+                "parameters": [
+                    {"name": "key", "type": "boolean", "in": "header"},
+                ],
+            },
+        },
+    }
+
+    schema = schemathesis.from_dict(raw_schema)
+
+    @given(schema["/data"]["GET"].as_strategy())
+    def test(case):
+        print(case)
+        assert is_valid_header(case.headers)
+
+    test()
+
+
 def test_filter_urlencoded(empty_open_api_3_schema):
     # When API schema allows for inputs that can't be serialized to `application/x-www-form-urlencoded`
     # Then such examples should be filtered out during generation