114 lines
3.3 KiB
Python
114 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from apipilot.errors import DuplicateToolError, InvalidConfigurationError
|
|
from apipilot.openapi_parser import (
|
|
build_tool_definition,
|
|
extract_operations,
|
|
validate_operations,
|
|
)
|
|
|
|
from .conftest import build_sample_app
|
|
|
|
|
|
def test_extract_operations_picks_enabled_only():
|
|
app, _ = build_sample_app()
|
|
ops = extract_operations(app)
|
|
names = {
|
|
(op.apipilot.get("name") or op.operation_id or op.path): op.apipilot.get("enabled")
|
|
for op in ops
|
|
}
|
|
assert "list_items" in names
|
|
assert "get_item" in names
|
|
assert "create_item" in names
|
|
assert all(v is True for v in names.values())
|
|
|
|
|
|
def test_build_tool_definition_path_and_query():
|
|
app, _ = build_sample_app()
|
|
root = app.openapi()
|
|
ops = extract_operations(app)
|
|
list_items_op = next(op for op in ops if op.apipilot.get("name") == "list_items")
|
|
definition, specs = build_tool_definition(list_items_op, root)
|
|
assert definition["name"] == "list_items"
|
|
props = definition["parameters"]["properties"]
|
|
assert "category" in props
|
|
assert props["category"]["type"] == "string"
|
|
assert "limit" in props
|
|
assert props["limit"]["type"] == "integer"
|
|
locs = {s.name: s.location for s in specs}
|
|
assert locs["category"] == "query"
|
|
assert locs["limit"] == "query"
|
|
assert "category" not in definition["parameters"]["required"]
|
|
|
|
|
|
def test_build_tool_definition_body():
|
|
app, _ = build_sample_app()
|
|
root = app.openapi()
|
|
ops = extract_operations(app)
|
|
create_op = next(op for op in ops if op.apipilot.get("name") == "create_item")
|
|
definition, specs = build_tool_definition(create_op, root)
|
|
assert definition["name"] == "create_item"
|
|
props = definition["parameters"]["properties"]
|
|
assert "name" in props
|
|
assert props["name"]["type"] == "string"
|
|
assert "name" in definition["parameters"]["required"]
|
|
|
|
|
|
def test_validate_duplicate_raises():
|
|
from fastapi import APIRouter, FastAPI
|
|
|
|
app = FastAPI()
|
|
router = APIRouter()
|
|
|
|
@router.get(
|
|
"/a",
|
|
openapi_extra={
|
|
"x-apipilot": {"enabled": True, "name": "dup_name", "description": "first"}
|
|
},
|
|
)
|
|
async def a():
|
|
return 1
|
|
|
|
@router.post(
|
|
"/b",
|
|
openapi_extra={
|
|
"x-apipilot": {"enabled": True, "name": "dup_name", "description": "second"}
|
|
},
|
|
)
|
|
async def b():
|
|
return 2
|
|
|
|
app.include_router(router)
|
|
with pytest.raises(DuplicateToolError):
|
|
from apipilot.openapi_parser import extract_operations as e, validate_operations as v
|
|
ops = e(app)
|
|
v(ops, app.openapi())
|
|
|
|
|
|
def test_validate_missing_name_and_operation_id_raises():
|
|
from fastapi import APIRouter, FastAPI
|
|
|
|
app = FastAPI()
|
|
router = APIRouter()
|
|
|
|
@router.get(
|
|
"/no-name",
|
|
openapi_extra={"x-apipilot": {"enabled": True, "description": "no name"}},
|
|
)
|
|
async def no_name():
|
|
return 1
|
|
|
|
app.include_router(router)
|
|
from apipilot.openapi_parser import extract_operations as e, validate_operations as v
|
|
|
|
# FastAPI generera' un operationId automatico se non fornito, ma simuliamo
|
|
# un caso senza assegnando None.
|
|
ops = e(app)
|
|
for op in ops:
|
|
op.operation_id = None
|
|
op.apipilot.pop("name", None)
|
|
with pytest.raises(InvalidConfigurationError):
|
|
v(ops, app.openapi())
|