OpenAPI initial support as capability source
This commit is contained in:
@@ -0,0 +1,105 @@
|
||||
{
|
||||
"openapi": "3.1.0",
|
||||
"info": {
|
||||
"title": "Petstore Minimal",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"servers": [
|
||||
{
|
||||
"url": "https://api.example.test"
|
||||
}
|
||||
],
|
||||
"paths": {
|
||||
"/pets/{petId}": {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "petId",
|
||||
"in": "path",
|
||||
"required": true,
|
||||
"schema": { "type": "string" }
|
||||
}
|
||||
],
|
||||
"get": {
|
||||
"operationId": "getPet",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "includeOwner",
|
||||
"in": "query",
|
||||
"required": false,
|
||||
"schema": { "type": "boolean", "default": false }
|
||||
}
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "Pet found",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": { "$ref": "#/components/schemas/Pet" }
|
||||
}
|
||||
}
|
||||
},
|
||||
"404": {
|
||||
"description": "Pet not found",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": { "$ref": "#/components/schemas/Error" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/pets": {
|
||||
"post": {
|
||||
"operationId": "createPet",
|
||||
"requestBody": {
|
||||
"required": true,
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": { "$ref": "#/components/schemas/CreatePetRequest" }
|
||||
}
|
||||
}
|
||||
},
|
||||
"responses": {
|
||||
"201": {
|
||||
"description": "Pet created",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": { "$ref": "#/components/schemas/Pet" }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"components": {
|
||||
"schemas": {
|
||||
"CreatePetRequest": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": { "type": "string" }
|
||||
},
|
||||
"required": ["name"],
|
||||
"additionalProperties": false
|
||||
},
|
||||
"Pet": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": { "type": "string" },
|
||||
"name": { "type": "string" }
|
||||
},
|
||||
"required": ["id", "name"],
|
||||
"additionalProperties": false
|
||||
},
|
||||
"Error": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"message": { "type": "string" }
|
||||
},
|
||||
"required": ["message"],
|
||||
"additionalProperties": false
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from wf_openapi.executor import OpenApiExecutionConfig
|
||||
|
||||
|
||||
def test_openapi_runtime_does_not_require_generated_manifest() -> None:
|
||||
config = OpenApiExecutionConfig(base_url="https://api.example.test")
|
||||
|
||||
assert config.base_url == "https://api.example.test"
|
||||
assert not hasattr(config, "generated_package")
|
||||
assert not hasattr(config, "operation_modules")
|
||||
assert not hasattr(config, "parameter_arguments")
|
||||
@@ -0,0 +1,176 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
from wf_authoring import NodeReturn
|
||||
from wf_openapi.executor import OpenApiExecutionConfig, call_openapi_operation
|
||||
from wf_openapi.executor import OpenApiOperationOutput
|
||||
from wf_openapi.spec import load_openapi_operations
|
||||
from wf_openapi.validation import load_openapi_app
|
||||
|
||||
FIXTURE = Path("tests/openapi/fixtures/petstore_minimal.openapi.json")
|
||||
|
||||
|
||||
def test_call_openapi_operation_maps_success() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
op for op in load_openapi_operations(FIXTURE) if op.name == "get_pet"
|
||||
)
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert request.url.path == "/pets/pet-1"
|
||||
assert request.url.params["includeOwner"] == "true"
|
||||
return httpx.Response(200, json={"id": "pet-1", "name": "Fluffy"})
|
||||
|
||||
async def run() -> NodeReturn[OpenApiOperationOutput]:
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
return await call_openapi_operation(
|
||||
app,
|
||||
operation,
|
||||
OpenApiExecutionConfig(base_url="https://api.example.test"),
|
||||
{"path": {"petId": "pet-1"}, "query": {"includeOwner": "true"}},
|
||||
client=client,
|
||||
)
|
||||
|
||||
result = asyncio.run(run())
|
||||
|
||||
assert result.outcome == "ok"
|
||||
assert result.output.status_code == 200
|
||||
assert result.output.body["id"] == "pet-1"
|
||||
|
||||
|
||||
def test_call_openapi_operation_maps_declared_http_error() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
op for op in load_openapi_operations(FIXTURE) if op.name == "get_pet"
|
||||
)
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
_ = request
|
||||
return httpx.Response(404, json={"message": "missing"})
|
||||
|
||||
async def run() -> NodeReturn[OpenApiOperationOutput]:
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
return await call_openapi_operation(
|
||||
app,
|
||||
operation,
|
||||
OpenApiExecutionConfig(base_url="https://api.example.test"),
|
||||
{"path": {"petId": "missing"}},
|
||||
client=client,
|
||||
)
|
||||
|
||||
result = asyncio.run(run())
|
||||
|
||||
assert result.outcome == "http_error"
|
||||
assert result.output.status_code == 404
|
||||
assert result.output.body["message"] == "missing"
|
||||
|
||||
|
||||
def test_call_openapi_operation_maps_unexpected_status() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
op for op in load_openapi_operations(FIXTURE) if op.name == "get_pet"
|
||||
)
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
_ = request
|
||||
return httpx.Response(418, json={"message": "teapot"})
|
||||
|
||||
async def run() -> NodeReturn[OpenApiOperationOutput]:
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
return await call_openapi_operation(
|
||||
app,
|
||||
operation,
|
||||
OpenApiExecutionConfig(base_url="https://api.example.test"),
|
||||
{"path": {"petId": "pet-1"}},
|
||||
client=client,
|
||||
)
|
||||
|
||||
result = asyncio.run(run())
|
||||
|
||||
assert result.outcome == "unexpected_status"
|
||||
assert result.output.status_code == 418
|
||||
assert result.output.validation_errors
|
||||
|
||||
|
||||
def test_call_openapi_operation_maps_invalid_request_to_validation_error() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
op for op in load_openapi_operations(FIXTURE) if op.name == "create_pet"
|
||||
)
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
raise AssertionError("invalid request should not be sent")
|
||||
|
||||
async def run() -> NodeReturn[OpenApiOperationOutput]:
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
return await call_openapi_operation(
|
||||
app,
|
||||
operation,
|
||||
OpenApiExecutionConfig(base_url="https://api.example.test"),
|
||||
{"body": {"extra": "field"}},
|
||||
client=client,
|
||||
)
|
||||
|
||||
result = asyncio.run(run())
|
||||
|
||||
assert result.outcome == "validation_error"
|
||||
assert result.output.status_code == 0
|
||||
assert result.output.validation_errors
|
||||
|
||||
|
||||
def test_call_openapi_operation_maps_invalid_response_to_validation_error() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
op for op in load_openapi_operations(FIXTURE) if op.name == "get_pet"
|
||||
)
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
_ = request
|
||||
return httpx.Response(200, json={"id": "pet-1"})
|
||||
|
||||
async def run() -> NodeReturn[OpenApiOperationOutput]:
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
return await call_openapi_operation(
|
||||
app,
|
||||
operation,
|
||||
OpenApiExecutionConfig(base_url="https://api.example.test"),
|
||||
{"path": {"petId": "pet-1"}},
|
||||
client=client,
|
||||
)
|
||||
|
||||
result = asyncio.run(run())
|
||||
|
||||
assert result.outcome == "validation_error"
|
||||
assert result.output.status_code == 200
|
||||
assert result.output.validation_errors
|
||||
|
||||
|
||||
def test_call_openapi_operation_maps_transport_error() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
op for op in load_openapi_operations(FIXTURE) if op.name == "get_pet"
|
||||
)
|
||||
|
||||
async def handler(request: httpx.Request) -> httpx.Response:
|
||||
_ = request
|
||||
raise httpx.ConnectError("offline")
|
||||
|
||||
async def run() -> NodeReturn[OpenApiOperationOutput]:
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
return await call_openapi_operation(
|
||||
app,
|
||||
operation,
|
||||
OpenApiExecutionConfig(base_url="https://api.example.test"),
|
||||
{"path": {"petId": "pet-1"}},
|
||||
client=client,
|
||||
)
|
||||
|
||||
result = asyncio.run(run())
|
||||
|
||||
assert result.outcome == "transport_error"
|
||||
assert result.output.status_code == 0
|
||||
assert result.output.validation_errors == ["offline"]
|
||||
@@ -0,0 +1,96 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from wf_openapi.request import HttpRequestParts, build_http_request_parts
|
||||
from wf_openapi.spec import load_openapi_operations
|
||||
from wf_openapi.validation import (
|
||||
HttpResponseParts,
|
||||
load_openapi_app,
|
||||
validate_openapi_request,
|
||||
validate_openapi_response,
|
||||
)
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "petstore_minimal.openapi.json"
|
||||
|
||||
|
||||
def test_validate_openapi_request_accepts_public_payload() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "get_pet"
|
||||
)
|
||||
parts = build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test",
|
||||
payload={"path": {"petId": "pet-1"}, "query": {"includeOwner": "true"}},
|
||||
)
|
||||
|
||||
result = validate_openapi_request(app, parts)
|
||||
|
||||
assert result.valid is True
|
||||
assert result.errors == []
|
||||
|
||||
|
||||
def test_validate_openapi_request_reports_invalid_body() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "create_pet"
|
||||
)
|
||||
parts = build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test",
|
||||
payload={"body": {"extra": "field"}},
|
||||
)
|
||||
|
||||
result = validate_openapi_request(app, parts)
|
||||
|
||||
assert result.valid is False
|
||||
assert result.errors
|
||||
|
||||
|
||||
def test_validate_openapi_response_accepts_declared_response() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "get_pet"
|
||||
)
|
||||
request = build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test",
|
||||
payload={"path": {"petId": "pet-1"}},
|
||||
)
|
||||
response = HttpResponseParts(
|
||||
status_code=200,
|
||||
headers={"content-type": "application/json"},
|
||||
data=json.dumps({"id": "pet-1", "name": "Fluffy"}).encode(),
|
||||
)
|
||||
|
||||
result = validate_openapi_response(app, request, response)
|
||||
|
||||
assert result.valid is True
|
||||
assert result.errors == []
|
||||
assert result.data["id"] == "pet-1"
|
||||
|
||||
|
||||
def test_validate_openapi_response_reports_undeclared_status() -> None:
|
||||
app = load_openapi_app(FIXTURE)
|
||||
request = HttpRequestParts(
|
||||
method="GET",
|
||||
url="https://api.example.test/pets/pet-1",
|
||||
)
|
||||
response = HttpResponseParts(
|
||||
status_code=418,
|
||||
headers={"content-type": "application/json"},
|
||||
data=b'{"message": "teapot"}',
|
||||
)
|
||||
|
||||
result = validate_openapi_response(app, request, response)
|
||||
|
||||
assert result.valid is False
|
||||
assert result.errors
|
||||
@@ -0,0 +1,89 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from wf_openapi.request import build_http_request_parts
|
||||
from wf_openapi.spec import load_openapi_operations
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "petstore_minimal.openapi.json"
|
||||
|
||||
|
||||
def test_build_http_request_parts_uses_public_openapi_names() -> None:
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "get_pet"
|
||||
)
|
||||
|
||||
parts = build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test/v1",
|
||||
payload={
|
||||
"path": {"petId": "pet 1"},
|
||||
"query": {"includeOwner": True},
|
||||
"header": {"X-Trace-ID": "trace-1"},
|
||||
"cookie": {"sessionId": "session-1"},
|
||||
},
|
||||
)
|
||||
|
||||
assert parts.method == "GET"
|
||||
assert parts.url == "https://api.example.test/v1/pets/pet%201"
|
||||
assert parts.params["includeOwner"] is True
|
||||
assert parts.headers["X-Trace-ID"] == "trace-1"
|
||||
assert parts.cookies["sessionId"] == "session-1"
|
||||
|
||||
|
||||
def test_build_http_request_parts_requires_path_parameters() -> None:
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "get_pet"
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="missing path parameter 'petId'"):
|
||||
build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test",
|
||||
payload={"path": {}},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("group", ["path", "query", "header", "cookie"])
|
||||
def test_build_http_request_parts_rejects_non_object_parameter_groups(
|
||||
group: str,
|
||||
) -> None:
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "get_pet"
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=rf"{group} must be an object"):
|
||||
payload: dict[str, object] = {"path": {"petId": "pet-1"}}
|
||||
payload[group] = ["not", "an", "object"]
|
||||
build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test",
|
||||
payload=payload,
|
||||
)
|
||||
|
||||
|
||||
def test_build_http_request_parts_passes_body_through_as_json() -> None:
|
||||
operation = next(
|
||||
operation
|
||||
for operation in load_openapi_operations(FIXTURE)
|
||||
if operation.name == "create_pet"
|
||||
)
|
||||
|
||||
parts = build_http_request_parts(
|
||||
operation,
|
||||
base_url="https://api.example.test",
|
||||
payload={"body": {"name": "Fluffy"}},
|
||||
)
|
||||
|
||||
assert parts.method == "POST"
|
||||
assert parts.url == "https://api.example.test/pets"
|
||||
assert isinstance(parts.json, dict)
|
||||
assert parts.json["name"] == "Fluffy"
|
||||
@@ -0,0 +1,104 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from openapi_core import OpenAPI
|
||||
|
||||
from wf_authoring import NodeReturn
|
||||
from wf_core import RuntimeContext
|
||||
from wf_openapi import source as source_module
|
||||
from wf_openapi.executor import OpenApiExecutionConfig, OpenApiOperationOutput
|
||||
from wf_openapi.models import OpenApiOperation
|
||||
from wf_openapi.source import build_openapi_capability_source
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "petstore_minimal.openapi.json"
|
||||
|
||||
|
||||
def test_build_openapi_capability_source_exposes_operations_as_node_specs() -> None:
|
||||
source = build_openapi_capability_source(
|
||||
source_id="petstore.default",
|
||||
document_path=FIXTURE,
|
||||
base_url="https://api.example.test",
|
||||
)
|
||||
|
||||
assert source.id == "petstore.default"
|
||||
assert source.kind == "connection"
|
||||
assert sorted(source.capabilities.node_specs) == [
|
||||
"petstore.default.create_pet",
|
||||
"petstore.default.get_pet",
|
||||
]
|
||||
|
||||
spec = source.capabilities.node_specs["petstore.default.get_pet"]
|
||||
node_def = spec.to_node_def()
|
||||
|
||||
assert spec.name == "petstore.default.get_pet"
|
||||
assert spec.is_async is True
|
||||
assert (
|
||||
node_def.input_schema.properties["path"]["properties"]["petId"]["type"]
|
||||
== "string"
|
||||
)
|
||||
assert (
|
||||
node_def.output_schema.properties["body"]["$ref"] == "#/components/schemas/Pet"
|
||||
)
|
||||
assert spec.outcomes == (
|
||||
"ok",
|
||||
"http_error",
|
||||
"unexpected_status",
|
||||
"validation_error",
|
||||
"transport_error",
|
||||
)
|
||||
|
||||
|
||||
def test_source_node_passes_operation_config_and_payload_to_execution(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
async def capture_call(
|
||||
app: OpenAPI,
|
||||
operation: OpenApiOperation,
|
||||
config: OpenApiExecutionConfig,
|
||||
payload: dict[str, object],
|
||||
) -> NodeReturn[OpenApiOperationOutput]:
|
||||
captured["app"] = app
|
||||
captured["operation"] = operation
|
||||
captured["config"] = config
|
||||
captured["payload"] = payload
|
||||
return NodeReturn(
|
||||
outcome="ok",
|
||||
output=OpenApiOperationOutput(
|
||||
status_code=200,
|
||||
headers={},
|
||||
body={},
|
||||
validation_errors=[],
|
||||
),
|
||||
)
|
||||
|
||||
monkeypatch.setattr(source_module, "call_openapi_operation", capture_call)
|
||||
source = build_openapi_capability_source(
|
||||
source_id="petstore.default",
|
||||
document_path=FIXTURE,
|
||||
base_url="https://api.example.test",
|
||||
)
|
||||
handler = source.capabilities.node_specs[
|
||||
"petstore.default.get_pet"
|
||||
].to_async_registry_handler()
|
||||
|
||||
async def run_handler() -> dict[str, object]:
|
||||
return await handler(
|
||||
{"path": {"petId": "pet-1"}},
|
||||
RuntimeContext(current_node_id="petstore.default.get_pet"),
|
||||
)
|
||||
|
||||
asyncio.run(run_handler())
|
||||
|
||||
operation = captured["operation"]
|
||||
config = captured["config"]
|
||||
assert isinstance(captured["app"], OpenAPI)
|
||||
assert isinstance(operation, OpenApiOperation)
|
||||
assert operation.name == "get_pet"
|
||||
assert isinstance(config, OpenApiExecutionConfig)
|
||||
assert config.base_url == "https://api.example.test"
|
||||
assert captured["payload"]["path"]["petId"] == "pet-1"
|
||||
@@ -0,0 +1,326 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from wf_openapi.spec import load_openapi_operations
|
||||
from wf_openapi.schemas import input_schema_for_operation, output_schema_for_operation
|
||||
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "petstore_minimal.openapi.json"
|
||||
|
||||
|
||||
def test_load_openapi_operations_discovers_operation_ids() -> None:
|
||||
operations = load_openapi_operations(FIXTURE)
|
||||
|
||||
names = [operation.name for operation in operations]
|
||||
assert names == ["create_pet", "get_pet"]
|
||||
assert operations[0].operation_id == "createPet"
|
||||
assert operations[0].method == "post"
|
||||
assert operations[0].path == "/pets"
|
||||
|
||||
|
||||
def test_load_openapi_operations_uses_deterministic_path_and_method_order(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
spec_path = _write_openapi(
|
||||
tmp_path,
|
||||
{
|
||||
"/z-last": {
|
||||
"post": {"operationId": "createLast"},
|
||||
"get": {"operationId": "getLast"},
|
||||
},
|
||||
"/a-first": {
|
||||
"delete": {"operationId": "deleteFirst"},
|
||||
"get": {"operationId": "getFirst"},
|
||||
"post": {"operationId": "createFirst"},
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
operations = load_openapi_operations(spec_path)
|
||||
|
||||
assert [operation.name for operation in operations] == [
|
||||
"get_first",
|
||||
"create_first",
|
||||
"delete_first",
|
||||
"get_last",
|
||||
"create_last",
|
||||
]
|
||||
|
||||
|
||||
def test_load_openapi_operations_rejects_duplicate_normalized_operation_names(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
spec_path = _write_openapi(
|
||||
tmp_path,
|
||||
{
|
||||
"/pets": {
|
||||
"get": {"operationId": "get-pet"},
|
||||
"post": {"operationId": "get_pet"},
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="Duplicate normalized OpenAPI operation name 'get_pet'"
|
||||
):
|
||||
load_openapi_operations(spec_path)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("paths", "message"),
|
||||
[
|
||||
(
|
||||
{"/pets": {"get": {"operationId": "!!!"}}},
|
||||
"OpenAPI operationId '!!!' does not produce a usable operation name",
|
||||
),
|
||||
(
|
||||
{"///": {"get": {}}},
|
||||
"OpenAPI fallback operation name for GET /// does not produce a usable operation name",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_load_openapi_operations_rejects_unusable_operation_names(
|
||||
tmp_path: Path,
|
||||
paths: dict[str, object],
|
||||
message: str,
|
||||
) -> None:
|
||||
spec_path = _write_openapi(tmp_path, paths)
|
||||
|
||||
with pytest.raises(ValueError, match=message):
|
||||
load_openapi_operations(spec_path)
|
||||
|
||||
|
||||
def test_operation_input_schema_combines_params_and_body() -> None:
|
||||
operations = load_openapi_operations(FIXTURE)
|
||||
create_pet = operations[0]
|
||||
get_pet = operations[1]
|
||||
|
||||
create_schema = input_schema_for_operation(create_pet)
|
||||
get_schema = input_schema_for_operation(get_pet)
|
||||
|
||||
assert (
|
||||
create_schema["properties"]["body"]["$ref"]
|
||||
== "#/components/schemas/CreatePetRequest"
|
||||
)
|
||||
assert "body" in create_schema["required"]
|
||||
assert get_schema["properties"]["path"]["properties"]["petId"]["type"] == "string"
|
||||
assert (
|
||||
get_schema["properties"]["query"]["properties"]["includeOwner"]["type"]
|
||||
== "boolean"
|
||||
)
|
||||
assert "body" not in get_schema["properties"]
|
||||
|
||||
|
||||
def test_operation_input_schema_emits_parameter_groups_in_canonical_order(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
spec_path = _write_openapi(
|
||||
tmp_path,
|
||||
{
|
||||
"/pets/{petId}": {
|
||||
"get": {
|
||||
"operationId": "getPet",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "session",
|
||||
"in": "cookie",
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{"name": "trace", "in": "header", "schema": {"type": "string"}},
|
||||
{
|
||||
"name": "includeOwner",
|
||||
"in": "query",
|
||||
"schema": {"type": "boolean"},
|
||||
},
|
||||
{
|
||||
"name": "petId",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
operation = load_openapi_operations(spec_path)[0]
|
||||
|
||||
schema = input_schema_for_operation(operation)
|
||||
|
||||
assert list(schema["properties"]) == ["path", "query", "header", "cookie"]
|
||||
|
||||
|
||||
def test_operation_effective_parameters_inherit_path_item_and_apply_operation_overrides(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
spec_path = _write_openapi(
|
||||
tmp_path,
|
||||
{
|
||||
"/pets": {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "locale",
|
||||
"in": "query",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "trace",
|
||||
"in": "header",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
],
|
||||
"get": {
|
||||
"operationId": "listPets",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "trace",
|
||||
"in": "header",
|
||||
"schema": {"type": "integer"},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
operation = load_openapi_operations(spec_path)[0]
|
||||
schema = input_schema_for_operation(operation)
|
||||
|
||||
assert [
|
||||
(parameter["name"], parameter["in"])
|
||||
for parameter in operation.effective_parameters
|
||||
] == [
|
||||
("locale", "query"),
|
||||
("trace", "header"),
|
||||
]
|
||||
assert schema["properties"]["query"]["properties"]["locale"]["type"] == "string"
|
||||
assert schema["properties"]["header"]["properties"]["trace"]["type"] == "integer"
|
||||
assert schema["properties"]["header"]["required"] == []
|
||||
|
||||
|
||||
def test_operation_records_optional_request_body_for_execution_boundary(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
operation = load_openapi_operations(
|
||||
_write_openapi(
|
||||
tmp_path,
|
||||
{
|
||||
"/pets": {
|
||||
"post": {
|
||||
"operationId": "createPet",
|
||||
"requestBody": {
|
||||
"required": False,
|
||||
"content": {
|
||||
"application/json": {"schema": {"type": "object"}},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
)[0]
|
||||
|
||||
assert operation.has_request_body is True
|
||||
assert "body" not in input_schema_for_operation(operation)["required"]
|
||||
|
||||
|
||||
def test_extracted_schemas_are_copied_from_raw_operation() -> None:
|
||||
operations = load_openapi_operations(FIXTURE)
|
||||
create_pet = operations[0]
|
||||
get_pet = operations[1]
|
||||
|
||||
input_schema = input_schema_for_operation(get_pet)
|
||||
create_input_schema = input_schema_for_operation(create_pet)
|
||||
output_schema = output_schema_for_operation(create_pet)
|
||||
|
||||
input_schema["properties"]["query"]["properties"]["includeOwner"]["type"] = "string"
|
||||
create_input_schema["properties"]["body"]["$ref"] = "#/mutated/request"
|
||||
output_schema["properties"]["body"]["$ref"] = "#/mutated/response"
|
||||
|
||||
raw_query_schema = get_pet.raw_operation["parameters"][0]["schema"]
|
||||
raw_request_schema = create_pet.raw_operation["requestBody"]["content"][
|
||||
"application/json"
|
||||
]["schema"]
|
||||
raw_response_schema = create_pet.raw_operation["responses"]["201"]["content"][
|
||||
"application/json"
|
||||
]["schema"]
|
||||
assert raw_query_schema["type"] == "boolean"
|
||||
assert raw_request_schema["$ref"] == "#/components/schemas/CreatePetRequest"
|
||||
assert raw_response_schema["$ref"] == "#/components/schemas/Pet"
|
||||
|
||||
second_input_schema = input_schema_for_operation(get_pet)
|
||||
second_create_input_schema = input_schema_for_operation(create_pet)
|
||||
second_output_schema = output_schema_for_operation(create_pet)
|
||||
assert (
|
||||
second_input_schema["properties"]["query"]["properties"]["includeOwner"]["type"]
|
||||
== "boolean"
|
||||
)
|
||||
assert (
|
||||
second_create_input_schema["properties"]["body"]["$ref"]
|
||||
== "#/components/schemas/CreatePetRequest"
|
||||
)
|
||||
assert (
|
||||
second_output_schema["properties"]["body"]["$ref"] == "#/components/schemas/Pet"
|
||||
)
|
||||
|
||||
|
||||
def test_operation_output_schema_uses_first_success_json_response() -> None:
|
||||
operations = load_openapi_operations(FIXTURE)
|
||||
schema = output_schema_for_operation(operations[0])
|
||||
|
||||
assert schema["properties"]["status_code"]["type"] == "integer"
|
||||
assert schema["properties"]["headers"]["type"] == "object"
|
||||
assert schema["properties"]["body"]["$ref"] == "#/components/schemas/Pet"
|
||||
|
||||
|
||||
def test_operation_output_schema_uses_empty_body_when_no_success_json_schema(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
spec_path = _write_openapi(
|
||||
tmp_path,
|
||||
{
|
||||
"/pets": {
|
||||
"post": {
|
||||
"operationId": "createPet",
|
||||
"responses": {
|
||||
"204": {"description": "No content"},
|
||||
"400": {
|
||||
"description": "Invalid pet",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {"$ref": "#/components/schemas/Error"}
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
operation = load_openapi_operations(spec_path)[0]
|
||||
|
||||
schema = output_schema_for_operation(operation)
|
||||
|
||||
assert schema["properties"]["body"] == {}
|
||||
|
||||
|
||||
def _write_openapi(tmp_path: Path, paths: dict[str, object]) -> Path:
|
||||
"""Write the minimum document shape needed by the operation inventory tests."""
|
||||
spec_path = tmp_path / "openapi.json"
|
||||
spec_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"openapi": "3.1.0",
|
||||
"info": {"title": "Test", "version": "1"},
|
||||
"paths": paths,
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
return spec_path
|
||||
Reference in New Issue
Block a user