mirror of
https://github.com/tiennm99/litellm.git
synced 2026-07-13 17:10:08 +00:00
Fix: Remove exec() usage and handle invalid OpenAPI parameter names
- Add to_safe_identifier() to convert any parameter name to valid Python identifier - Refactor create_tool_function() to use closure with **kwargs instead of exec() - Handle edge cases: hyphens, dots, leading digits, Python keywords, special chars - Add comprehensive test suite covering all edge cases - Fixes #18471: OpenAPI MCP server crashes on invalid parameter names - Security: Eliminates arbitrary code execution risk from untrusted OpenAPI specs
This commit is contained in:
hamzaq453
@@ -3,6 +3,8 @@ This module is used to generate MCP tools from OpenAPI specs.
|
||||
"""
|
||||
|
||||
import json
|
||||
import keyword
|
||||
import re
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import httpx
|
||||
@@ -17,6 +19,64 @@ BASE_URL = ""
|
||||
HEADERS: Dict[str, str] = {}
|
||||
|
||||
|
||||
def to_safe_identifier(name: str) -> str:
|
||||
"""
|
||||
Convert an OpenAPI parameter name to a safe Python identifier.
|
||||
|
||||
This function ensures that any parameter name from an OpenAPI spec can be
|
||||
used as a Python function parameter without causing syntax errors or
|
||||
security issues. It handles:
|
||||
- Hyphens, dots, and other special characters
|
||||
- Leading digits
|
||||
- Python keywords
|
||||
- Special characters like $, @, etc.
|
||||
|
||||
Args:
|
||||
name: The original parameter name from the OpenAPI spec
|
||||
|
||||
Returns:
|
||||
A valid Python identifier that can be used in function signatures
|
||||
|
||||
Examples:
|
||||
>>> to_safe_identifier("repository-id")
|
||||
'repository_id'
|
||||
>>> to_safe_identifier("2fa-code")
|
||||
'_2fa_code'
|
||||
>>> to_safe_identifier("user.name")
|
||||
'user_name'
|
||||
>>> to_safe_identifier("$filter")
|
||||
'_filter'
|
||||
>>> to_safe_identifier("class")
|
||||
'class_'
|
||||
"""
|
||||
if not name:
|
||||
return "_empty_"
|
||||
|
||||
# Start with underscore if first char is not a letter
|
||||
# Replace all non-alphanumeric chars (except underscore) with underscore
|
||||
# Collapse multiple underscores
|
||||
safe = re.sub(r'[^a-zA-Z0-9_]', '_', name)
|
||||
safe = re.sub(r'_+', '_', safe) # Collapse multiple underscores
|
||||
|
||||
# If starts with digit, prefix with underscore
|
||||
if safe and safe[0].isdigit():
|
||||
safe = '_' + safe
|
||||
|
||||
# If empty after sanitization, use a default
|
||||
if not safe:
|
||||
safe = '_param_'
|
||||
|
||||
# If it's a Python keyword, append underscore
|
||||
if keyword.iskeyword(safe):
|
||||
safe = safe + '_'
|
||||
|
||||
# Ensure it doesn't start with a digit (shouldn't happen after above, but double-check)
|
||||
if safe and safe[0].isdigit():
|
||||
safe = '_' + safe
|
||||
|
||||
return safe
|
||||
|
||||
|
||||
def load_openapi_spec(filepath: str) -> Dict[str, Any]:
|
||||
"""Load OpenAPI specification from JSON file."""
|
||||
with open(filepath, "r") as f:
|
||||
@@ -112,12 +172,20 @@ def create_tool_function(
|
||||
):
|
||||
"""Create a tool function for an OpenAPI operation.
|
||||
|
||||
This function creates an async tool function that can be called with
|
||||
keyword arguments. Parameter names from the OpenAPI spec are safely
|
||||
mapped to valid Python identifiers to avoid syntax errors and security
|
||||
issues.
|
||||
|
||||
Args:
|
||||
path: API endpoint path
|
||||
method: HTTP method (get, post, put, delete, patch)
|
||||
operation: OpenAPI operation object
|
||||
base_url: Base URL for the API
|
||||
headers: Optional headers to include in requests (e.g., authentication)
|
||||
|
||||
Returns:
|
||||
An async function that accepts **kwargs and makes the HTTP request
|
||||
"""
|
||||
if headers is None:
|
||||
headers = {}
|
||||
@@ -125,77 +193,109 @@ def create_tool_function(
|
||||
path_params, query_params, body_params = extract_parameters(operation)
|
||||
all_params = path_params + query_params + body_params
|
||||
|
||||
# Build function signature dynamically
|
||||
if all_params:
|
||||
params_str = ", ".join(f"{p}: str = ''" for p in all_params)
|
||||
else:
|
||||
params_str = ""
|
||||
|
||||
# Create the function code as a string
|
||||
func_code = f'''
|
||||
async def tool_function({params_str}) -> str:
|
||||
"""Dynamically generated tool function."""
|
||||
url = base_url + path
|
||||
# Create mapping from original parameter names to safe identifiers
|
||||
# This allows us to accept kwargs with original names but use safe names internally
|
||||
param_name_map: Dict[str, str] = {}
|
||||
safe_to_original_map: Dict[str, str] = {}
|
||||
|
||||
# Replace path parameters
|
||||
path_param_names = {path_params}
|
||||
for param_name in path_param_names:
|
||||
param_value = locals().get(param_name, "")
|
||||
if param_value:
|
||||
url = url.replace("{{" + param_name + "}}", str(param_value))
|
||||
|
||||
# Build query params
|
||||
query_param_names = {query_params}
|
||||
params = {{}}
|
||||
for param_name in query_param_names:
|
||||
param_value = locals().get(param_name, "")
|
||||
if param_value:
|
||||
params[param_name] = param_value
|
||||
|
||||
# Build request body
|
||||
body_param_names = {body_params}
|
||||
json_body = None
|
||||
if body_param_names:
|
||||
body_value = locals().get("body", {{}})
|
||||
if isinstance(body_value, dict):
|
||||
json_body = body_value
|
||||
elif body_value:
|
||||
# If it's a string, try to parse as JSON
|
||||
import json as json_module
|
||||
try:
|
||||
json_body = json_module.loads(body_value) if isinstance(body_value, str) else {{"data": body_value}}
|
||||
except:
|
||||
json_body = {{"data": body_value}}
|
||||
|
||||
# Make HTTP request
|
||||
async with httpx.AsyncClient() as client:
|
||||
if "{method.lower()}" == "get":
|
||||
response = await client.get(url, params=params, headers=headers)
|
||||
elif "{method.lower()}" == "post":
|
||||
response = await client.post(url, params=params, json=json_body, headers=headers)
|
||||
elif "{method.lower()}" == "put":
|
||||
response = await client.put(url, params=params, json=json_body, headers=headers)
|
||||
elif "{method.lower()}" == "delete":
|
||||
response = await client.delete(url, params=params, headers=headers)
|
||||
elif "{method.lower()}" == "patch":
|
||||
response = await client.patch(url, params=params, json=json_body, headers=headers)
|
||||
else:
|
||||
return "Unsupported HTTP method: {method}"
|
||||
for orig_name in all_params:
|
||||
safe_name = to_safe_identifier(orig_name)
|
||||
# Handle collisions: if safe name already exists, append a counter
|
||||
counter = 1
|
||||
original_safe = safe_name
|
||||
while safe_name in safe_to_original_map:
|
||||
safe_name = f"{original_safe}_{counter}"
|
||||
counter += 1
|
||||
|
||||
return response.text
|
||||
'''
|
||||
param_name_map[orig_name] = safe_name
|
||||
safe_to_original_map[safe_name] = orig_name
|
||||
|
||||
# Execute the function code to create the actual function
|
||||
local_vars = {
|
||||
"httpx": httpx,
|
||||
"headers": headers,
|
||||
"base_url": base_url,
|
||||
"path": path,
|
||||
"method": method,
|
||||
}
|
||||
exec(func_code, local_vars)
|
||||
# Store original parameter lists for use in the closure
|
||||
original_path_params = path_params
|
||||
original_query_params = query_params
|
||||
original_body_params = body_params
|
||||
original_method = method.lower()
|
||||
|
||||
return local_vars["tool_function"]
|
||||
async def tool_function(**kwargs: Any) -> str:
|
||||
"""
|
||||
Dynamically generated tool function.
|
||||
|
||||
Accepts keyword arguments where keys are the original OpenAPI parameter names.
|
||||
The function safely handles parameter names that aren't valid Python identifiers.
|
||||
"""
|
||||
# Build URL from base_url and path
|
||||
url = base_url + path
|
||||
|
||||
# Replace path parameters using original names from OpenAPI spec
|
||||
for orig_param_name in original_path_params:
|
||||
# Try to get value using original name first, then safe name
|
||||
param_value = kwargs.get(orig_param_name, "")
|
||||
if not param_value and orig_param_name in param_name_map:
|
||||
safe_name = param_name_map[orig_param_name]
|
||||
param_value = kwargs.get(safe_name, "")
|
||||
|
||||
if param_value:
|
||||
# Replace {param_name} or {{param_name}} in URL
|
||||
url = url.replace("{" + orig_param_name + "}", str(param_value))
|
||||
url = url.replace("{{" + orig_param_name + "}}", str(param_value))
|
||||
|
||||
# Build query params using original parameter names
|
||||
params: Dict[str, Any] = {}
|
||||
for orig_param_name in original_query_params:
|
||||
# Try to get value using original name first, then safe name
|
||||
param_value = kwargs.get(orig_param_name, "")
|
||||
if not param_value and orig_param_name in param_name_map:
|
||||
safe_name = param_name_map[orig_param_name]
|
||||
param_value = kwargs.get(safe_name, "")
|
||||
|
||||
if param_value:
|
||||
# Use original parameter name in query string (as expected by API)
|
||||
params[orig_param_name] = param_value
|
||||
|
||||
# Build request body
|
||||
json_body: Optional[Dict[str, Any]] = None
|
||||
if original_body_params:
|
||||
# Try "body" first (most common), then check all body param names
|
||||
body_value = kwargs.get("body", {})
|
||||
if not body_value:
|
||||
for orig_param_name in original_body_params:
|
||||
body_value = kwargs.get(orig_param_name, {})
|
||||
if body_value:
|
||||
break
|
||||
# Also try safe name
|
||||
if orig_param_name in param_name_map:
|
||||
safe_name = param_name_map[orig_param_name]
|
||||
body_value = kwargs.get(safe_name, {})
|
||||
if body_value:
|
||||
break
|
||||
|
||||
if isinstance(body_value, dict):
|
||||
json_body = body_value
|
||||
elif body_value:
|
||||
# If it's a string, try to parse as JSON
|
||||
try:
|
||||
json_body = json.loads(body_value) if isinstance(body_value, str) else {"data": body_value}
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
json_body = {"data": body_value}
|
||||
|
||||
# Make HTTP request
|
||||
async with httpx.AsyncClient() as client:
|
||||
if original_method == "get":
|
||||
response = await client.get(url, params=params, headers=headers)
|
||||
elif original_method == "post":
|
||||
response = await client.post(url, params=params, json=json_body, headers=headers)
|
||||
elif original_method == "put":
|
||||
response = await client.put(url, params=params, json=json_body, headers=headers)
|
||||
elif original_method == "delete":
|
||||
response = await client.delete(url, params=params, headers=headers)
|
||||
elif original_method == "patch":
|
||||
response = await client.patch(url, params=params, json=json_body, headers=headers)
|
||||
else:
|
||||
return f"Unsupported HTTP method: {original_method}"
|
||||
|
||||
return response.text
|
||||
|
||||
return tool_function
|
||||
|
||||
|
||||
def register_tools_from_openapi(spec: Dict[str, Any], base_url: str):
|
||||
|
||||
@@ -0,0 +1,528 @@
|
||||
"""
|
||||
Tests for OpenAPI to MCP generator, focusing on security and edge cases.
|
||||
|
||||
This test suite ensures that:
|
||||
1. Parameter names with invalid Python identifiers are handled safely
|
||||
2. No exec() is used (security)
|
||||
3. All edge cases (hyphens, dots, keywords, special chars) work correctly
|
||||
"""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
from typing import Dict, Any
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import (
|
||||
to_safe_identifier,
|
||||
create_tool_function,
|
||||
build_input_schema,
|
||||
extract_parameters,
|
||||
)
|
||||
|
||||
|
||||
class TestToSafeIdentifier:
|
||||
"""Test the to_safe_identifier function for various edge cases."""
|
||||
|
||||
def test_hyphen_in_name(self):
|
||||
"""Test parameter names with hyphens."""
|
||||
assert to_safe_identifier("repository-id") == "repository_id"
|
||||
assert to_safe_identifier("user-name") == "user_name"
|
||||
assert to_safe_identifier("api-key") == "api_key"
|
||||
|
||||
def test_leading_digit(self):
|
||||
"""Test parameter names starting with digits."""
|
||||
assert to_safe_identifier("2fa-code") == "_2fa_code"
|
||||
assert to_safe_identifier("123abc") == "_123abc"
|
||||
assert to_safe_identifier("0test") == "_0test"
|
||||
|
||||
def test_dots_in_name(self):
|
||||
"""Test parameter names with dots."""
|
||||
assert to_safe_identifier("user.name") == "user_name"
|
||||
assert to_safe_identifier("config.value") == "config_value"
|
||||
assert to_safe_identifier("api.v2") == "api_v2"
|
||||
|
||||
def test_dollar_sign(self):
|
||||
"""Test parameter names with dollar signs (OData style)."""
|
||||
assert to_safe_identifier("$filter") == "_filter"
|
||||
assert to_safe_identifier("$context") == "_context"
|
||||
assert to_safe_identifier("$select") == "_select"
|
||||
|
||||
def test_python_keywords(self):
|
||||
"""Test Python keywords are handled."""
|
||||
assert to_safe_identifier("class") == "class_"
|
||||
assert to_safe_identifier("from") == "from_"
|
||||
assert to_safe_identifier("not") == "not_"
|
||||
assert to_safe_identifier("def") == "def_"
|
||||
assert to_safe_identifier("import") == "import_"
|
||||
|
||||
def test_special_characters(self):
|
||||
"""Test various special characters."""
|
||||
assert to_safe_identifier("user@domain") == "user_domain"
|
||||
assert to_safe_identifier("test#hash") == "test_hash"
|
||||
assert to_safe_identifier("path/to/resource") == "path_to_resource"
|
||||
assert to_safe_identifier("param+value") == "param_value"
|
||||
|
||||
def test_multiple_special_chars(self):
|
||||
"""Test names with multiple special characters."""
|
||||
assert to_safe_identifier("user-name.email@domain") == "user_name_email_domain"
|
||||
assert to_safe_identifier("$filter.value") == "_filter_value"
|
||||
|
||||
def test_already_valid_identifier(self):
|
||||
"""Test that valid identifiers remain unchanged (except keywords)."""
|
||||
assert to_safe_identifier("valid_name") == "valid_name"
|
||||
assert to_safe_identifier("validName123") == "validName123"
|
||||
assert to_safe_identifier("_private") == "_private"
|
||||
|
||||
def test_empty_string(self):
|
||||
"""Test empty string handling."""
|
||||
assert to_safe_identifier("") == "_empty_"
|
||||
|
||||
def test_only_special_chars(self):
|
||||
"""Test names that are only special characters."""
|
||||
result = to_safe_identifier("---")
|
||||
assert result.startswith("_")
|
||||
assert len(result) > 0
|
||||
|
||||
def test_collision_handling(self):
|
||||
"""Test that similar names produce different safe identifiers."""
|
||||
# These should produce different results
|
||||
name1 = to_safe_identifier("user-name")
|
||||
name2 = to_safe_identifier("user_name")
|
||||
# They might be the same after sanitization, which is acceptable
|
||||
# The important thing is they're both valid identifiers
|
||||
|
||||
|
||||
class TestCreateToolFunction:
|
||||
"""Test create_tool_function with various parameter name edge cases."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hyphenated_path_parameter(self):
|
||||
"""Test function with hyphenated path parameter (e.g., repository-id)."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "repository-id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/repos/{repository-id}",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
# Should not raise SyntaxError
|
||||
assert callable(func)
|
||||
assert func.__name__ == "tool_function"
|
||||
|
||||
# Test calling with original parameter name
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = '{"id": "123"}'
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(**{"repository-id": "test-repo"})
|
||||
assert result == '{"id": "123"}'
|
||||
|
||||
# Verify URL was constructed correctly
|
||||
call_args = mock_client.return_value.__aenter__.return_value.get.call_args
|
||||
assert "repository-id" in str(call_args[0][0]) or "test-repo" in str(call_args[0][0])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_leading_digit_parameter(self):
|
||||
"""Test function with parameter starting with digit (e.g., 2fa-code)."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "2fa-code",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/verify",
|
||||
method="post",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "verified"
|
||||
mock_client.return_value.__aenter__.return_value.post = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(**{"2fa-code": "123456"})
|
||||
assert result == "verified"
|
||||
|
||||
# Verify query parameter was included
|
||||
call_args = mock_client.return_value.__aenter__.return_value.post.call_args
|
||||
assert call_args[1]["params"]["2fa-code"] == "123456"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dot_in_parameter_name(self):
|
||||
"""Test function with dot in parameter name (e.g., user.name)."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "user.name",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/search",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "found"
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(**{"user.name": "john.doe"})
|
||||
assert result == "found"
|
||||
|
||||
call_args = mock_client.return_value.__aenter__.return_value.get.call_args
|
||||
assert call_args[1]["params"]["user.name"] == "john.doe"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dollar_sign_parameter(self):
|
||||
"""Test function with dollar sign parameter (OData style, e.g., $filter)."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "$filter",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/entities",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "[]"
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(**{"$filter": "name eq 'test'"})
|
||||
assert result == "[]"
|
||||
|
||||
call_args = mock_client.return_value.__aenter__.return_value.get.call_args
|
||||
assert call_args[1]["params"]["$filter"] == "name eq 'test'"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_python_keyword_parameter(self):
|
||||
"""Test function with Python keyword as parameter name (e.g., class)."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "class",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/items",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "items"
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(**{"class": "premium"})
|
||||
assert result == "items"
|
||||
|
||||
call_args = mock_client.return_value.__aenter__.return_value.get.call_args
|
||||
assert call_args[1]["params"]["class"] == "premium"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_problematic_parameters(self):
|
||||
"""Test function with multiple problematic parameter names."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "repository-id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "2fa-code",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "$filter",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/repos/{repository-id}",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "success"
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(
|
||||
**{
|
||||
"repository-id": "test-repo",
|
||||
"2fa-code": "123",
|
||||
"$filter": "active",
|
||||
}
|
||||
)
|
||||
assert result == "success"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_body_parameter(self):
|
||||
"""Test function with request body parameter."""
|
||||
operation = {
|
||||
"requestBody": {
|
||||
"required": True,
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/create",
|
||||
method="post",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "created"
|
||||
mock_client.return_value.__aenter__.return_value.post = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func(**{"body": {"name": "test"}})
|
||||
assert result == "created"
|
||||
|
||||
call_args = mock_client.return_value.__aenter__.return_value.post.call_args
|
||||
assert call_args[1]["json"] == {"name": "test"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_parameters(self):
|
||||
"""Test function with no parameters."""
|
||||
operation = {}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/health",
|
||||
method="get",
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "ok"
|
||||
mock_client.return_value.__aenter__.return_value.get = AsyncMock(
|
||||
return_value=mock_response
|
||||
)
|
||||
|
||||
result = await func()
|
||||
assert result == "ok"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_all_http_methods(self):
|
||||
"""Test all supported HTTP methods."""
|
||||
methods = ["get", "post", "put", "delete", "patch"]
|
||||
|
||||
for method in methods:
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "repository-id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
func = create_tool_function(
|
||||
path="/repos/{repository-id}",
|
||||
method=method,
|
||||
operation=operation,
|
||||
base_url="https://api.example.com",
|
||||
)
|
||||
|
||||
assert callable(func)
|
||||
|
||||
with patch("httpx.AsyncClient") as mock_client:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.text = "success"
|
||||
|
||||
client_method = getattr(
|
||||
mock_client.return_value.__aenter__.return_value, method
|
||||
)
|
||||
client_method.return_value = mock_response
|
||||
client_method = AsyncMock(return_value=mock_response)
|
||||
setattr(
|
||||
mock_client.return_value.__aenter__.return_value,
|
||||
method,
|
||||
client_method,
|
||||
)
|
||||
|
||||
result = await func(**{"repository-id": "test"})
|
||||
assert result == "success"
|
||||
|
||||
def test_no_exec_usage(self):
|
||||
"""Verify that create_tool_function does not use exec()."""
|
||||
import ast
|
||||
import inspect
|
||||
|
||||
# Get the source code of create_tool_function
|
||||
source = inspect.getsource(create_tool_function)
|
||||
|
||||
# Parse the AST
|
||||
tree = ast.parse(source)
|
||||
|
||||
# Check for exec() calls
|
||||
exec_calls = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Call):
|
||||
if isinstance(node.func, ast.Name) and node.func.id == "exec":
|
||||
exec_calls.append(node)
|
||||
|
||||
# Should have no exec() calls
|
||||
assert len(exec_calls) == 0, "create_tool_function should not use exec()"
|
||||
|
||||
|
||||
class TestBuildInputSchema:
|
||||
"""Test that build_input_schema preserves original parameter names."""
|
||||
|
||||
def test_original_parameter_names_preserved(self):
|
||||
"""Test that original parameter names are preserved in input schema."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{
|
||||
"name": "repository-id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "2fa-code",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "$filter",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
schema = build_input_schema(operation)
|
||||
|
||||
# Original names should be in the schema
|
||||
assert "repository-id" in schema["properties"]
|
||||
assert "2fa-code" in schema["properties"]
|
||||
assert "$filter" in schema["properties"]
|
||||
|
||||
# Required should include original names
|
||||
assert "repository-id" in schema["required"]
|
||||
|
||||
|
||||
class TestExtractParameters:
|
||||
"""Test parameter extraction from OpenAPI operations."""
|
||||
|
||||
def test_extract_path_query_body_params(self):
|
||||
"""Test extraction of different parameter types."""
|
||||
operation = {
|
||||
"parameters": [
|
||||
{"name": "repo-id", "in": "path"},
|
||||
{"name": "filter", "in": "query"},
|
||||
{"name": "data", "in": "body"},
|
||||
],
|
||||
"requestBody": {
|
||||
"content": {"application/json": {"schema": {"type": "object"}}}
|
||||
},
|
||||
}
|
||||
|
||||
path_params, query_params, body_params = extract_parameters(operation)
|
||||
|
||||
assert "repo-id" in path_params
|
||||
assert "filter" in query_params
|
||||
assert "data" in body_params
|
||||
assert "body" in body_params # From requestBody
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
|
||||
Reference in New Issue
Block a user