Files
responder/tests/test_clientgen.py
T

168 lines
4.5 KiB
Python

"""Client generation from OpenAPI."""
import importlib.util
import pytest
from pydantic import BaseModel
import responder
from responder import Path, Query
def _load_module(path):
spec = importlib.util.spec_from_file_location("generated_client", path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
class ItemIn(BaseModel):
name: str
class ItemOut(BaseModel):
id: int
name: str
def _api():
api = responder.API(
title="Service",
version="1",
openapi="3.0.2",
allowed_hosts=[";"],
session_https_only=False,
)
@api.get("/users/{user_id}", operation_id="get_user")
def get_user(
req,
resp,
*,
user_id: int = Path(...),
include_details: bool = Query(False),
):
resp.media = {"id": user_id, "details": include_details}
@api.post(
"/items",
operation_id="create_item",
request_model=ItemIn,
response_model=ItemOut,
)
async def create_item(req, resp):
item = await req.media()
resp.media = {"id": 1, **item}
@api.get("/boom", operation_id="boom")
def boom(req, resp):
resp.status_code = 418
resp.media = {"error": "teapot"}
return api
def test_generate_client_source_and_call_in_process_session(tmp_path):
api = _api()
path = tmp_path / "service_client.py"
written = api.generate_client(path, class_name="ServiceClient")
assert written == path
module = _load_module(path)
client = module.ServiceClient(session=api.requests)
assert client.get_user(7, include_details=True) == {
"id": 7,
"details": True,
}
assert client.create_item(body={"name": "tea"}) == {"id": 1, "name": "tea"}
with pytest.raises(module.APIError) as excinfo:
client.boom()
assert excinfo.value.status_code == 418
assert excinfo.value.body == {"error": "teapot"}
def test_generate_client_returns_source():
api = _api()
source = api.generate_client(class_name="ServiceClient")
assert "class ServiceClient" in source
assert "def get_user(self, user_id: int" in source
assert "class ItemIn(TypedDict):" in source
assert "class ItemOut(TypedDict):" in source
assert "def create_item(self, body: ItemIn | None = None) -> ItemOut" in source
@pytest.mark.parametrize(
("language", "expected"),
[
(
"javascript",
[
"export class ServiceClient",
"get_user(userId, includeDetails = null)",
"create_item(body = null)",
"fetchImpl",
],
),
(
"typescript",
[
"export class ServiceClient",
"get_user(userId: number, includeDetails: boolean | null = null)",
"export interface ItemIn",
"export interface ItemOut",
"create_item(body: ItemIn | null = null): Promise<ItemOut>",
],
),
(
"ruby",
[
"class ServiceClient",
"def get_user(user_id, include_details: nil)",
"def create_item(body: nil)",
"Net::HTTP",
],
),
(
"php",
[
"class ServiceClient",
"public function get_user($user_id, $include_details = null): mixed",
"public function create_item($body = null): mixed",
"file_get_contents",
],
),
],
)
def test_generate_client_supports_other_languages(language, expected):
api = _api()
source = api.generate_client(class_name="ServiceClient", language=language)
for snippet in expected:
assert snippet in source
def test_generate_client_writes_other_language(tmp_path):
api = _api()
path = tmp_path / "service_client.ts"
written = api.generate_client(
path, class_name="ServiceClient", language="typescript"
)
assert written == path
assert "export class ServiceClient" in path.read_text()
def test_generate_client_rejects_unknown_language():
api = _api()
with pytest.raises(ValueError, match="language must be one of"):
api.generate_client(class_name="ServiceClient", language="elvish")
def test_generate_client_requires_openapi():
api = responder.API(allowed_hosts=[";"], session_https_only=False)
with pytest.raises(RuntimeError, match="OpenAPI is not enabled"):
api.generate_client()