openapi-first/tests/test_codegen_routes.py

"""Smoke tests for openapi_first.codegen_routes."""

import json

import pytest

from openapi_first.codegen_routes import generate_routes


def _write_spec(tmp_path, spec: dict):
    path = tmp_path / "openapi.json"
    path.write_text(json.dumps(spec), encoding="utf-8")
    return path


def _spec(paths: dict, schemas: dict | None = None) -> dict:
    spec = {
        "openapi": "3.0.3",
        "info": {"title": "Codegen Test API", "version": "1.0.0"},
        "paths": paths,
    }
    if schemas is not None:
        spec["components"] = {"schemas": schemas}
    return spec


def test_flat_output_names_files_by_resource(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/items": {
                    "get": {
                        "operationId": "list_items",
                        "responses": {"200": {"description": "OK"}},
                    }
                }
            }
        ),
    )
    files = generate_routes(spec, tmp_path / "routes")
    assert [f.name for f in files] == ["items.py"]


def test_resource_name_is_sanitized_to_identifier(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/fetch-requests": {
                    "get": {
                        "operationId": "list_fetch_requests",
                        "responses": {"200": {"description": "OK"}},
                    }
                }
            }
        ),
    )
    files = generate_routes(spec, tmp_path / "routes")
    assert [f.name for f in files] == ["fetch_requests.py"]
    source = files[0].read_text(encoding="utf-8")
    compile(source, str(files[0]), "exec")


def test_inject_request_prepends_request_param(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/items": {
                    "get": {
                        "operationId": "list_items",
                        "responses": {"200": {"description": "OK"}},
                    }
                }
            }
        ),
    )
    files = generate_routes(spec, tmp_path / "routes", inject_request=True)
    source = files[0].read_text(encoding="utf-8")
    assert "from fastapi import Request" in source or "Request" in source
    assert "def list_items(request: Request):" in source


def test_package_layout_emits_routes_package(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/items/{id}": {
                    "get": {
                        "operationId": "get_item",
                        "parameters": [
                            {
                                "name": "id",
                                "in": "path",
                                "required": True,
                                "schema": {"type": "string"},
                            }
                        ],
                        "responses": {"200": {"description": "OK"}},
                    }
                }
            }
        ),
    )
    out = tmp_path / "interfaces"
    files = generate_routes(spec, out, package=True)
    assert files[0] == out / "items" / "routes.py"
    assert (out / "items" / "__init__.py").exists()
    assert (out / "__init__.py").exists()
    assert "def get_item(id: str):" in files[0].read_text(encoding="utf-8")


def test_multipart_body_expands_to_uploadfile(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/pets/{id}/photo": {
                    "post": {
                        "operationId": "upload_pet_photo",
                        "parameters": [
                            {
                                "name": "id",
                                "in": "path",
                                "required": True,
                                "schema": {"type": "string"},
                            }
                        ],
                        "requestBody": {
                            "required": True,
                            "content": {
                                "multipart/form-data": {
                                    "schema": {
                                        "type": "object",
                                        "required": ["file"],
                                        "properties": {
                                            "file": {
                                                "type": "string",
                                                "format": "binary",
                                            },
                                            "caption": {"type": "string"},
                                        },
                                    }
                                }
                            },
                        },
                        "responses": {"200": {"description": "OK"}},
                    }
                }
            }
        ),
    )
    files = generate_routes(spec, tmp_path / "routes")
    source = files[0].read_text(encoding="utf-8")
    assert "UploadFile" in source
    assert "file: UploadFile = File(...)" in source
    assert "caption: str = Form(None)" in source
    assert "payload: dict" not in source
    compile(source, str(files[0]), "exec")


def test_json_body_unchanged_by_default(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/items": {
                    "post": {
                        "operationId": "create_item",
                        "requestBody": {
                            "required": True,
                            "content": {
                                "application/json": {
                                    "schema": {"$ref": "#/components/schemas/ItemInput"}
                                }
                            },
                        },
                        "responses": {"201": {"description": "Created"}},
                    }
                }
            },
            {"ItemInput": {"type": "object", "properties": {"name": {"type": "string"}}}},
        ),
    )
    files = generate_routes(spec, tmp_path / "routes")
    source = files[0].read_text(encoding="utf-8")
    assert "payload: dict" in source
    assert "response: Response" in source


def test_missing_operation_id_raises(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec({"/items": {"get": {"responses": {"200": {"description": "OK"}}}}}),
    )
    with pytest.raises(ValueError):
        generate_routes(spec, tmp_path / "routes")


def test_template_style_async_body_model(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/items": {
                    "post": {
                        "operationId": "create_item",
                        "requestBody": {
                            "required": True,
                            "content": {
                                "application/json": {
                                    "schema": {"$ref": "#/components/schemas/ItemInput"}
                                }
                            },
                        },
                        "responses": {"201": {"description": "Created"}},
                    }
                }
            },
            {"ItemInput": {"type": "object", "properties": {"name": {"type": "string"}}}},
        ),
    )
    files = generate_routes(
        spec,
        tmp_path / "routes",
        use_models=True,
        models_module="_gen.models",
        inject_request=True,
        template_style=True,
    )
    source = files[0].read_text(encoding="utf-8")
    assert "async def create_item(request: Request, body: ItemInput, response: Response):" in source
    assert "payload" not in source
    assert "db_resources" in source
    compile(source, str(files[0]), "exec")


def test_template_style_dict_body_without_models(tmp_path):
    spec = _write_spec(
        tmp_path,
        _spec(
            {
                "/items": {
                    "post": {
                        "operationId": "create_item",
                        "requestBody": {
                            "required": True,
                            "content": {
                                "application/json": {
                                    "schema": {"$ref": "#/components/schemas/ItemInput"}
                                }
                            },
                        },
                        "responses": {"201": {"description": "Created"}},
                    }
                }
            }
        ),
    )
    files = generate_routes(
        spec, tmp_path / "routes", inject_request=True, template_style=True
    )
    source = files[0].read_text(encoding="utf-8")
    assert "async def create_item(request: Request, body: dict, response: Response):" in source
    assert "payload" not in source
    compile(source, str(files[0]), "exec")