responder/tests/test_clientgen.py

"""Client generation from OpenAPI."""

import importlib.util
import shutil
import subprocess
from collections.abc import AsyncIterator

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",
        response_model=ItemOut,
    )
    async def create_item(req, resp, *, item: ItemIn):
        resp.media = {"id": 1, **item.model_dump()}

    @api.get("/items", operation_id="list_items")
    def list_items(req, resp) -> list[ItemOut]:
        return [{"id": 1, "name": "tea"}]

    @api.get("/boom", operation_id="boom")
    def boom(req, resp):
        resp.status_code = 418
        resp.media = {"error": "teapot"}

    return api


def _stream_api():
    api = responder.API(
        title="Streams",
        version="1",
        openapi="3.1.0",
        allowed_hosts=[";"],
        sessions=False,
    )

    @api.sse("/events", operation_id="stream_events")
    async def stream_events(req, resp) -> AsyncIterator[responder.SSE[ItemOut]]:
        yield responder.SSE(
            ItemOut(id=1, name="tea"), event="item", id="1", retry=1000
        )

    @api.ndjson("/items", operation_id="stream_items")
    async def stream_items(req, resp) -> AsyncIterator[ItemOut]:
        yield ItemOut(id=1, name="tea")
        yield ItemOut(id=2, name="coffee")

    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"}
    assert client.list_items() == [{"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"}
    assert excinfo.value.problem is None


def test_generated_python_client_validates_request_body(tmp_path):
    api = _api()
    path = tmp_path / "service_client.py"
    api.generate_client(path, class_name="ServiceClient")
    module = _load_module(path)
    client = module.ServiceClient(session=api.requests, validate=True)

    with pytest.raises(module.APIValidationError) as excinfo:
        client.create_item(body={"name": 123})

    assert str(excinfo.value) == "body.name expected string"
    assert excinfo.value.path == "body.name"
    assert excinfo.value.expected == "string"


def test_generated_python_client_validates_response_body(tmp_path):
    api = _api()
    path = tmp_path / "service_client.py"
    api.generate_client(path, class_name="ServiceClient")
    module = _load_module(path)

    class Response:
        status_code = 200
        headers = {"content-type": "application/json"}
        content = b'{"id": "bad", "name": "tea"}'

    class Session:
        def request(self, *args, **kwargs):
            return Response()

    client = module.ServiceClient(session=Session(), validate=True)
    with pytest.raises(module.APIValidationError) as excinfo:
        client.create_item(body={"name": "tea"})

    assert str(excinfo.value) == "response.id expected integer"
    assert excinfo.value.path == "response.id"
    assert excinfo.value.value == "bad"


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 "class APIValidationError(Exception):" in source
    assert "class APIProblem(TypedDict, total=False):" in source
    assert "validate: bool = False" in source
    # 8.0.1: an inferred Pydantic body is documented required:true, so the
    # generated client requires it too (it used to default to None and 422).
    assert "def create_item(self, body: ItemIn) -> ItemOut" in source
    assert "def list_items(self) -> list[ItemOut]" in source


def test_generated_python_client_exposes_problem_details(tmp_path):
    api = responder.API(
        title="Service",
        version="1",
        openapi="3.0.2",
        allowed_hosts=[";"],
        session_https_only=False,
    )

    @api.get("/missing", operation_id="missing")
    def missing(req, resp):
        responder.abort(404)

    path = tmp_path / "service_client.py"
    api.generate_client(path, class_name="ServiceClient")
    module = _load_module(path)
    client = module.ServiceClient(session=api.requests)

    with pytest.raises(module.APIError) as excinfo:
        client.missing()

    assert excinfo.value.status_code == 404
    assert excinfo.value.problem["title"] == "Not Found"
    assert excinfo.value.title == "Not Found"
    assert str(excinfo.value) == "Not Found"


def test_generated_python_client_consumes_typed_streams(tmp_path):
    api = _stream_api()
    path = tmp_path / "stream_client.py"
    api.generate_client(path, class_name="StreamClient")
    module = _load_module(path)
    client = module.StreamClient(session=api.requests, validate=True)

    events = list(client.stream_events())
    assert events == [
        module.SSEEvent(
            data={"id": 1, "name": "tea"},
            event="item",
            id="1",
            retry=1000,
            comment=None,
        )
    ]
    assert list(client.stream_items()) == [
        {"id": 1, "name": "tea"},
        {"id": 2, "name": "coffee"},
    ]


@pytest.mark.parametrize(
    ("language", "expected"),
    [
        ("python", "Iterator[SSEEvent[ItemOut]]"),
        ("javascript", "async *stream_events()"),
        ("typescript", "AsyncGenerator<SSEEvent<ItemOut>>"),
        ("ruby", "Enumerator.new do |out|"),
        ("php", "public function stream_events(): \\Generator"),
    ],
)
def test_generated_clients_expose_streaming_iterators(language, expected):
    source = _stream_api().generate_client(language=language)
    assert expected in source


@pytest.mark.parametrize(
    ("language", "suffix", "command"),
    [
        ("javascript", ".mjs", ("node", "--check")),
        ("typescript", ".ts", ("deno", "check")),
        ("ruby", ".rb", ("ruby", "-c")),
        ("php", ".php", ("php", "-l")),
    ],
)
def test_generated_streaming_clients_pass_native_syntax_check(
    language, suffix, command, tmp_path
):
    tool = shutil.which(command[0])
    if tool is None:
        pytest.skip(f"{command[0]} not installed")
    path = tmp_path / f"stream_client{suffix}"
    _stream_api().generate_client(path, language=language)
    result = subprocess.run(  # noqa: S603 - test-controlled executable/path
        [tool, *command[1:], str(path)],
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stdout + result.stderr


def test_generated_python_client_uses_later_success_response_schema(tmp_path):
    api = responder.API(
        title="Service",
        version="1",
        openapi="3.0.2",
        allowed_hosts=[";"],
        session_https_only=False,
    )
    created_schema = {
        "type": "object",
        "required": ["id"],
        "properties": {"id": {"type": "integer"}},
    }

    @api.post(
        "/jobs",
        operation_id="create_job",
        responses={
            200: "Queued without a body",
            201: {
                "description": "Created",
                "content": {"application/json": {"schema": created_schema}},
            },
        },
    )
    def create_job(req, resp):
        resp.created({"id": 1}, location="/jobs/1")

    path = tmp_path / "service_client.py"
    api.generate_client(path, class_name="ServiceClient")
    module = _load_module(path)

    class Response:
        status_code = 201
        headers = {"content-type": "application/json"}
        content = b'{"id": "bad"}'

    class Session:
        def request(self, *args, **kwargs):
            return Response()

    client = module.ServiceClient(session=Session(), validate=True)
    with pytest.raises(module.APIValidationError) as excinfo:
        client.create_job()

    assert str(excinfo.value) == "response.id expected integer"
    assert excinfo.value.path == "response.id"


@pytest.mark.parametrize(
    ("language", "expected"),
    [
        (
            "javascript",
            [
                "export class ServiceClient",
                "get_user(userId, includeDetails = null)",
                "create_item(body)",
                "fetchImpl",
                "APIValidationError",
                "this.problem = problem",
                "validate = false",
            ],
        ),
        (
            "typescript",
            [
                "export class ServiceClient",
                "get_user(userId: number, includeDetails: boolean | null = null)",
                "export interface ItemIn",
                "export interface ItemOut",
                "export interface ProblemDetails",
                "create_item(body: ItemIn): Promise<ItemOut>",
                "list_items(): Promise<Array<ItemOut>>",
                "export class APIValidationError",
                "responseSchema",
            ],
        ),
        (
            "ruby",
            [
                "class ServiceClient",
                "def get_user(user_id, include_details: nil)",
                "def create_item(body)",
                "Net::HTTP",
            ],
        ),
        (
            "php",
            [
                "class ServiceClient",
                "public function get_user($user_id, $include_details = null): mixed",
                "public function create_item($body): 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()


def test_generated_typescript_validator_is_annotated():
    """The TS validation helpers must be type-annotated (noImplicitAny)."""
    source = _api().generate_client(class_name="ServiceClient", language="typescript")
    assert "const SCHEMAS: Record<string, any>" in source
    assert "const validateValue = (value: any, schema: any, path: string" in source
    assert "const resolveSchema = (schema: any)" in source
    assert "variants.some((variant: any)" in source


@pytest.mark.parametrize(
    ("language", "suffix", "command"),
    [
        ("javascript", ".mjs", ("node", "--check")),
        ("typescript", ".ts", ("deno", "check")),
        ("ruby", ".rb", ("ruby", "-c")),
        ("php", ".php", ("php", "-l")),
    ],
)
def test_generated_client_passes_native_syntax_check(
    language, suffix, command, tmp_path
):
    """Each generated client parses/type-checks with its own language's tool.

    Skipped when the toolchain isn't installed; CI installs all four so the
    generated JS/TS/Ruby/PHP are verified before release, not after.
    """
    tool = shutil.which(command[0])
    if tool is None:
        pytest.skip(f"{command[0]} not installed")
    path = tmp_path / f"client{suffix}"
    _api().generate_client(str(path), class_name="ServiceClient", language=language)
    result = subprocess.run(  # noqa: S603 - tool path + temp file are test-controlled
        [tool, *command[1:], str(path)],
        capture_output=True,
        text=True,
        timeout=120,
    )
    assert result.returncode == 0, result.stdout + result.stderr