131 lines
3.6 KiB
Python
131 lines
3.6 KiB
Python
"""Smoke tests for openapi_first.client."""
|
|
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from openapi_first.client import OpenAPIClient, OpenAPIClientError
|
|
|
|
SPEC = {
|
|
"openapi": "3.0.3",
|
|
"info": {"title": "Client API", "version": "1.0.0"},
|
|
"servers": [{"url": "https://api.example.com/v1"}],
|
|
"paths": {
|
|
"/users/{user_id}": {
|
|
"get": {
|
|
"operationId": "get_user",
|
|
"parameters": [
|
|
{
|
|
"name": "user_id",
|
|
"in": "path",
|
|
"required": True,
|
|
"schema": {"type": "integer"},
|
|
}
|
|
],
|
|
"responses": {"200": {"description": "OK"}},
|
|
}
|
|
},
|
|
"/users": {
|
|
"post": {
|
|
"operationId": "create_user",
|
|
"requestBody": {
|
|
"content": {"application/json": {"schema": {"type": "object"}}}
|
|
},
|
|
"responses": {"201": {"description": "Created"}},
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class _RecordingClient:
|
|
"""Minimal httpx.Client stand-in that records and routes requests."""
|
|
|
|
def __init__(self, base_url=None):
|
|
self.base_url = base_url
|
|
self.requests = []
|
|
|
|
def request(
|
|
self,
|
|
*,
|
|
method,
|
|
url,
|
|
params=None,
|
|
headers=None,
|
|
json=None,
|
|
content=None,
|
|
timeout=None,
|
|
):
|
|
self.requests.append(
|
|
{
|
|
"method": method,
|
|
"url": url,
|
|
"params": params,
|
|
"headers": headers,
|
|
"json": json,
|
|
"content": content,
|
|
"timeout": timeout,
|
|
}
|
|
)
|
|
response = types.SimpleNamespace(status_code=200, json=lambda: {"ok": True})
|
|
return response
|
|
|
|
|
|
def test_default_base_url_from_servers():
|
|
client = OpenAPIClient(SPEC)
|
|
assert client.base_url == "https://api.example.com/v1/"
|
|
|
|
|
|
def test_operation_methods_are_generated(minimal_spec):
|
|
client = OpenAPIClient(minimal_spec, base_url="http://test")
|
|
assert "get_health" in client.operations()
|
|
assert callable(client.get_health)
|
|
|
|
|
|
def test_unknown_operation_raises_attribute_error(minimal_spec):
|
|
client = OpenAPIClient(minimal_spec, base_url="http://test")
|
|
with pytest.raises(AttributeError):
|
|
client.does_not_exist()
|
|
|
|
|
|
def test_path_param_formatting_and_urljoin():
|
|
rec = _RecordingClient()
|
|
client = OpenAPIClient(SPEC, base_url="https://host", client=rec)
|
|
client.get_user(path_params={"user_id": 42})
|
|
assert rec.requests[0]["url"] == "https://host/users/42"
|
|
|
|
|
|
def test_missing_required_path_param():
|
|
rec = _RecordingClient()
|
|
client = OpenAPIClient(SPEC, base_url="https://host", client=rec)
|
|
with pytest.raises(OpenAPIClientError):
|
|
client.get_user(path_params={})
|
|
|
|
|
|
def test_json_body_defaults_content_type():
|
|
rec = _RecordingClient()
|
|
client = OpenAPIClient(SPEC, base_url="https://host", client=rec)
|
|
client.create_user(body={"name": "Ada"})
|
|
req = rec.requests[0]
|
|
assert req["json"] == {"name": "Ada"}
|
|
assert req["headers"].get("Content-Type") == "application/json"
|
|
|
|
|
|
def test_missing_server_raises():
|
|
spec = dict(SPEC)
|
|
spec.pop("servers")
|
|
with pytest.raises(OpenAPIClientError):
|
|
OpenAPIClient(spec)
|
|
|
|
|
|
def test_duplicate_operation_id_raises():
|
|
spec = dict(SPEC)
|
|
spec["paths"]["/extra"] = {
|
|
"get": {
|
|
"operationId": "get_user",
|
|
"responses": {"200": {"description": "OK"}},
|
|
}
|
|
}
|
|
with pytest.raises(OpenAPIClientError):
|
|
OpenAPIClient(spec)
|