Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -198,6 +198,17 @@ def url_join(self, base_url: str, path: str):
full_path = urljoin(base_path, path.lstrip("/"))
return urlunparse(parsed_base._replace(path=full_path))

@staticmethod
def _serialize_parameter_value(value: Any) -> Any:
"""Serialize a scalar parameter value the way OpenAPI expects it on the wire.

Booleans use the JSON spelling (``true``/``false``) instead of Python's
``True``/``False``; every other value is passed through unchanged.
"""
if isinstance(value, bool):
return "true" if value else "false"
return value

def build_headers(self, arguments: dict[str, Any]) -> dict[str, str]:
"""Build the headers for the operation."""
headers = {}
Expand All @@ -215,7 +226,7 @@ def build_headers(self, arguments: dict[str, Any]) -> dict[str, str]:
)
continue

headers[parameter.name] = str(argument)
headers[parameter.name] = str(self._serialize_parameter_value(argument))

return headers

Expand Down Expand Up @@ -328,7 +339,9 @@ def build_path(self, path_template: str, arguments: dict[str, Any]) -> str:
f"required parameter of the operation - `{self.id}`."
)
continue
path_template = path_template.replace(f"{{{parameter.name}}}", quote(str(argument), safe=""))
path_template = path_template.replace(
f"{{{parameter.name}}}", quote(str(self._serialize_parameter_value(argument)), safe="")
)
self._validate_path_segments(path_template)
return path_template

Expand Down Expand Up @@ -396,8 +409,12 @@ def build_query_string(self, arguments: dict[str, Any]) -> str:
f"required parameter of the operation - `{self.id}`."
)
continue
if isinstance(argument, (list, tuple)):
argument = [self._serialize_parameter_value(item) for item in argument]
else:
argument = self._serialize_parameter_value(argument)
segments.append((parameter.name, argument))
return urlencode(segments)
return urlencode(segments, doseq=True)

def replace_invalid_symbols(self, parameter_name):
"""Replace invalid symbols in the parameter name with underscores."""
Expand Down
50 changes: 50 additions & 0 deletions python/tests/unit/connectors/openapi_plugin/test_sk_openapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -685,6 +685,56 @@ def test_build_query_string_with_optional_and_required_parameters():
assert operation.build_query_string(arguments) == expected_query_string


def test_build_query_string_with_array_and_boolean_parameters():
parameters = [
RestApiParameter(name="tags", type="array", location=RestApiParameterLocation.QUERY, is_required=False),
RestApiParameter(name="active", type="boolean", location=RestApiParameterLocation.QUERY, is_required=False),
]
operation = RestApiOperation(
id="test", method="GET", servers=["https://example.com/"], path="/resource", params=parameters
)
arguments = {"tags": ["red", "blue"], "active": True}
expected_query_string = "tags=red&tags=blue&active=true"
assert operation.build_query_string(arguments) == expected_query_string


def test_build_query_string_with_false_and_boolean_array_parameters():
parameters = [
RestApiParameter(name="active", type="boolean", location=RestApiParameterLocation.QUERY, is_required=False),
RestApiParameter(name="flags", type="array", location=RestApiParameterLocation.QUERY, is_required=False),
]
operation = RestApiOperation(
id="test", method="GET", servers=["https://example.com/"], path="/resource", params=parameters
)
arguments = {"active": False, "flags": [True, False]}
expected_query_string = "active=false&flags=true&flags=false"
assert operation.build_query_string(arguments) == expected_query_string


def test_build_path_with_boolean_parameter():
parameters = [
RestApiParameter(name="active", type="boolean", location=RestApiParameterLocation.PATH, is_required=True)
]
operation = RestApiOperation(
id="test", method="GET", servers=["https://example.com/"], path="/resource/{active}", params=parameters
)
arguments = {"active": True}
expected_path = "/resource/true"
assert operation.build_path(operation.path, arguments) == expected_path


def test_build_headers_with_boolean_parameter():
parameters = [
RestApiParameter(name="X-Active", type="boolean", location=RestApiParameterLocation.HEADER, is_required=True)
]
operation = RestApiOperation(
id="test", method="GET", servers=["https://example.com"], path="test/path", params=parameters
)
arguments = {"X-Active": False}
expected_headers = {"X-Active": "false"}
assert operation.build_headers(arguments) == expected_headers


def test_create_payload_artificial_parameter_with_text_plain():
properties = [
RestApiPayloadProperty(
Expand Down
Loading