Kye Gomez
2 months ago
parent
5e8a3d7860
commit
fb494267eb
@ -1,6 +0,0 @@
|
||||
pytest
|
||||
swarms
|
||||
loguru
|
||||
pydantic
|
||||
swarm-models
|
||||
loguru
|
||||
@ -1,72 +0,0 @@
|
||||
import pytest
|
||||
|
||||
from swarm_models import OpenAIChat
|
||||
from swarms.structs.agent import Agent
|
||||
from swarms.structs.company import Company
|
||||
|
||||
# Mock OpenAIChat instance
|
||||
llm = OpenAIChat(openai_api_key="test_key", max_tokens=4000)
|
||||
|
||||
# Mock Agents
|
||||
ceo = Agent(llm=llm, name="CEO")
|
||||
dev = Agent(llm=llm, name="Developer")
|
||||
va = Agent(llm=llm, name="VA")
|
||||
hr = Agent(llm=llm, name="HR")
|
||||
|
||||
shared_instructions = "Listen to your boss"
|
||||
|
||||
|
||||
def test_add_agent():
|
||||
company = Company(
|
||||
org_chart=[[ceo, [dev, va]]],
|
||||
shared_instructions=shared_instructions,
|
||||
)
|
||||
company.add(hr)
|
||||
assert hr in company.agents
|
||||
|
||||
|
||||
def test_get_agent():
|
||||
company = Company(
|
||||
org_chart=[[ceo, [dev, va]]],
|
||||
shared_instructions=shared_instructions,
|
||||
)
|
||||
company.add(hr)
|
||||
assert company.get("HR") == hr
|
||||
|
||||
|
||||
def test_remove_agent():
|
||||
company = Company(
|
||||
org_chart=[[ceo, [dev, va]]],
|
||||
shared_instructions=shared_instructions,
|
||||
)
|
||||
company.add(hr)
|
||||
company.remove(hr)
|
||||
assert hr not in company.agents
|
||||
|
||||
|
||||
def test_add_existing_agent():
|
||||
company = Company(
|
||||
org_chart=[[ceo, [dev, va]]],
|
||||
shared_instructions=shared_instructions,
|
||||
)
|
||||
company.add(hr)
|
||||
with pytest.raises(ValueError):
|
||||
company.add(hr)
|
||||
|
||||
|
||||
def test_get_nonexistent_agent():
|
||||
company = Company(
|
||||
org_chart=[[ceo, [dev, va]]],
|
||||
shared_instructions=shared_instructions,
|
||||
)
|
||||
with pytest.raises(ValueError):
|
||||
company.get("Nonexistent")
|
||||
|
||||
|
||||
def test_remove_nonexistent_agent():
|
||||
company = Company(
|
||||
org_chart=[[ceo, [dev, va]]],
|
||||
shared_instructions=shared_instructions,
|
||||
)
|
||||
with pytest.raises(ValueError):
|
||||
company.remove(hr)
|
||||
@ -1,52 +0,0 @@
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from swarm_models import OpenAIChat
|
||||
from swarms.structs import Agent, Task
|
||||
from swarms.structs.team import Team
|
||||
|
||||
|
||||
class TestTeam(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.agent = Agent(
|
||||
llm=OpenAIChat(openai_api_key=""),
|
||||
max_loops=1,
|
||||
dashboard=False,
|
||||
)
|
||||
self.task = Task(
|
||||
description="What's the weather in miami",
|
||||
agent=self.agent,
|
||||
)
|
||||
self.team = Team(
|
||||
tasks=[self.task],
|
||||
agents=[self.agent],
|
||||
architecture="sequential",
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
def test_check_config(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self.team.check_config({"config": None})
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
self.team.check_config(
|
||||
{"config": json.dumps({"agents": [], "tasks": []})}
|
||||
)
|
||||
|
||||
def test_run(self):
|
||||
self.assertEqual(self.team.run(), self.task.run())
|
||||
|
||||
def test_sequential_loop(self):
|
||||
self.assertEqual(
|
||||
self.team._Team__sequential_loop(), self.task.run()
|
||||
)
|
||||
|
||||
def test_log(self):
|
||||
self.assertIsNone(self.team._Team__log("Test message"))
|
||||
|
||||
self.team.verbose = True
|
||||
self.assertIsNone(self.team._Team__log("Test message"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@ -1,784 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel
|
||||
|
||||
from swarms.tools.tool import (
|
||||
BaseTool,
|
||||
Runnable,
|
||||
StructuredTool,
|
||||
Tool,
|
||||
tool,
|
||||
)
|
||||
|
||||
# Define test data
|
||||
test_input = {"key1": "value1", "key2": "value2"}
|
||||
expected_output = "expected_output_value"
|
||||
|
||||
# Test with global variables
|
||||
global_var = "global"
|
||||
|
||||
|
||||
# Basic tests for BaseTool
|
||||
def test_base_tool_init():
|
||||
# Test BaseTool initialization
|
||||
tool = BaseTool()
|
||||
assert isinstance(tool, BaseTool)
|
||||
|
||||
|
||||
def test_base_tool_invoke():
|
||||
# Test BaseTool invoke method
|
||||
tool = BaseTool()
|
||||
result = tool.invoke(test_input)
|
||||
assert result == expected_output
|
||||
|
||||
|
||||
# Basic tests for Tool
|
||||
def test_tool_init():
|
||||
# Test Tool initialization
|
||||
tool = Tool()
|
||||
assert isinstance(tool, Tool)
|
||||
|
||||
|
||||
def test_tool_invoke():
|
||||
# Test Tool invoke method
|
||||
tool = Tool()
|
||||
result = tool.invoke(test_input)
|
||||
assert result == expected_output
|
||||
|
||||
|
||||
# Basic tests for StructuredTool
|
||||
def test_structured_tool_init():
|
||||
# Test StructuredTool initialization
|
||||
tool = StructuredTool()
|
||||
assert isinstance(tool, StructuredTool)
|
||||
|
||||
|
||||
def test_structured_tool_invoke():
|
||||
# Test StructuredTool invoke method
|
||||
tool = StructuredTool()
|
||||
result = tool.invoke(test_input)
|
||||
assert result == expected_output
|
||||
|
||||
|
||||
# Test additional functionality and edge cases as needed
|
||||
|
||||
|
||||
def test_tool_creation():
|
||||
tool = Tool(
|
||||
name="test_tool", func=lambda x: x, description="Test tool"
|
||||
)
|
||||
assert tool.name == "test_tool"
|
||||
assert tool.func is not None
|
||||
assert tool.description == "Test tool"
|
||||
|
||||
|
||||
def test_tool_ainvoke():
|
||||
tool = Tool(
|
||||
name="test_tool", func=lambda x: x, description="Test tool"
|
||||
)
|
||||
result = tool.ainvoke("input_data")
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_tool_ainvoke_with_coroutine():
|
||||
async def async_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = Tool(
|
||||
name="test_tool",
|
||||
coroutine=async_function,
|
||||
description="Test tool",
|
||||
)
|
||||
result = tool.ainvoke("input_data")
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_tool_args():
|
||||
def sample_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = Tool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
description="Test tool",
|
||||
)
|
||||
assert tool.args == {"tool_input": {"type": "string"}}
|
||||
|
||||
|
||||
# Basic tests for StructuredTool class
|
||||
|
||||
|
||||
def test_structured_tool_creation():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=lambda x: x,
|
||||
description="Test tool",
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
assert tool.name == "test_tool"
|
||||
assert tool.func is not None
|
||||
assert tool.description == "Test tool"
|
||||
assert tool.args_schema == SampleArgsSchema
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=lambda x: x,
|
||||
description="Test tool",
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
result = tool.ainvoke({"tool_input": "input_data"})
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke_with_coroutine():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
async def async_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
coroutine=async_function,
|
||||
description="Test tool",
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
result = tool.ainvoke({"tool_input": "input_data"})
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_structured_tool_args():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
def sample_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
description="Test tool",
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
assert tool.args == {"tool_input": {"type": "string"}}
|
||||
|
||||
|
||||
# Additional tests for exception handling
|
||||
|
||||
|
||||
def test_tool_ainvoke_exception():
|
||||
tool = Tool(name="test_tool", func=None, description="Test tool")
|
||||
with pytest.raises(NotImplementedError):
|
||||
tool.ainvoke("input_data")
|
||||
|
||||
|
||||
def test_tool_ainvoke_with_coroutine_exception():
|
||||
tool = Tool(
|
||||
name="test_tool", coroutine=None, description="Test tool"
|
||||
)
|
||||
with pytest.raises(NotImplementedError):
|
||||
tool.ainvoke("input_data")
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke_exception():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=None,
|
||||
description="Test tool",
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
with pytest.raises(NotImplementedError):
|
||||
tool.ainvoke({"tool_input": "input_data"})
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke_with_coroutine_exception():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
coroutine=None,
|
||||
description="Test tool",
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
with pytest.raises(NotImplementedError):
|
||||
tool.ainvoke({"tool_input": "input_data"})
|
||||
|
||||
|
||||
def test_tool_description_not_provided():
|
||||
tool = Tool(name="test_tool", func=lambda x: x)
|
||||
assert tool.name == "test_tool"
|
||||
assert tool.func is not None
|
||||
assert tool.description == ""
|
||||
|
||||
|
||||
def test_tool_invoke_with_callbacks():
|
||||
def sample_function(input_data, callbacks=None):
|
||||
if callbacks:
|
||||
callbacks.on_start()
|
||||
callbacks.on_finish()
|
||||
return input_data
|
||||
|
||||
tool = Tool(name="test_tool", func=sample_function)
|
||||
callbacks = MagicMock()
|
||||
result = tool.invoke("input_data", callbacks=callbacks)
|
||||
assert result == "input_data"
|
||||
callbacks.on_start.assert_called_once()
|
||||
callbacks.on_finish.assert_called_once()
|
||||
|
||||
|
||||
def test_tool_invoke_with_new_argument():
|
||||
def sample_function(input_data, callbacks=None):
|
||||
return input_data
|
||||
|
||||
tool = Tool(name="test_tool", func=sample_function)
|
||||
result = tool.invoke("input_data", callbacks=None)
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_tool_ainvoke_with_new_argument():
|
||||
async def async_function(input_data, callbacks=None):
|
||||
return input_data
|
||||
|
||||
tool = Tool(name="test_tool", coroutine=async_function)
|
||||
result = tool.ainvoke("input_data", callbacks=None)
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_tool_description_from_docstring():
|
||||
def sample_function(input_data):
|
||||
"""Sample function docstring"""
|
||||
return input_data
|
||||
|
||||
tool = Tool(name="test_tool", func=sample_function)
|
||||
assert tool.description == "Sample function docstring"
|
||||
|
||||
|
||||
def test_tool_ainvoke_with_exceptions():
|
||||
async def async_function(input_data):
|
||||
raise ValueError("Test exception")
|
||||
|
||||
tool = Tool(name="test_tool", coroutine=async_function)
|
||||
with pytest.raises(ValueError):
|
||||
tool.ainvoke("input_data")
|
||||
|
||||
|
||||
# Additional tests for StructuredTool class
|
||||
|
||||
|
||||
def test_structured_tool_infer_schema_false():
|
||||
def sample_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
args_schema=None,
|
||||
infer_schema=False,
|
||||
)
|
||||
assert tool.args_schema is None
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke_with_callbacks():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
def sample_function(input_data, callbacks=None):
|
||||
if callbacks:
|
||||
callbacks.on_start()
|
||||
callbacks.on_finish()
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
callbacks = MagicMock()
|
||||
result = tool.ainvoke(
|
||||
{"tool_input": "input_data"}, callbacks=callbacks
|
||||
)
|
||||
assert result == "input_data"
|
||||
callbacks.on_start.assert_called_once()
|
||||
callbacks.on_finish.assert_called_once()
|
||||
|
||||
|
||||
def test_structured_tool_description_not_provided():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=lambda x: x,
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
assert tool.name == "test_tool"
|
||||
assert tool.func is not None
|
||||
assert tool.description == ""
|
||||
|
||||
|
||||
def test_structured_tool_args_schema():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
def sample_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
assert tool.args_schema == SampleArgsSchema
|
||||
|
||||
|
||||
def test_structured_tool_args_schema_inference():
|
||||
def sample_function(input_data):
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
args_schema=None,
|
||||
infer_schema=True,
|
||||
)
|
||||
assert tool.args_schema is not None
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke_with_new_argument():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
def sample_function(input_data, callbacks=None):
|
||||
return input_data
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
func=sample_function,
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
result = tool.ainvoke(
|
||||
{"tool_input": "input_data"}, callbacks=None
|
||||
)
|
||||
assert result == "input_data"
|
||||
|
||||
|
||||
def test_structured_tool_ainvoke_with_exceptions():
|
||||
class SampleArgsSchema:
|
||||
pass
|
||||
|
||||
async def async_function(input_data):
|
||||
raise ValueError("Test exception")
|
||||
|
||||
tool = StructuredTool(
|
||||
name="test_tool",
|
||||
coroutine=async_function,
|
||||
args_schema=SampleArgsSchema,
|
||||
)
|
||||
with pytest.raises(ValueError):
|
||||
tool.ainvoke({"tool_input": "input_data"})
|
||||
|
||||
|
||||
def test_base_tool_verbose_logging(caplog):
|
||||
# Test verbose logging in BaseTool
|
||||
tool = BaseTool(verbose=True)
|
||||
result = tool.invoke(test_input)
|
||||
assert result == expected_output
|
||||
assert "Verbose logging" in caplog.text
|
||||
|
||||
|
||||
def test_tool_exception_handling():
|
||||
# Test exception handling in Tool
|
||||
tool = Tool()
|
||||
with pytest.raises(Exception):
|
||||
tool.invoke(test_input, raise_exception=True)
|
||||
|
||||
|
||||
def test_structured_tool_async_invoke():
|
||||
# Test asynchronous invoke in StructuredTool
|
||||
tool = StructuredTool()
|
||||
result = tool.ainvoke(test_input)
|
||||
assert result == expected_output
|
||||
|
||||
|
||||
# Add more tests for specific functionalities and edge cases as needed
|
||||
# Import necessary libraries and modules
|
||||
|
||||
|
||||
# Example of a mock function to be used in testing
|
||||
def mock_function(arg: str) -> str:
|
||||
"""A simple mock function for testing."""
|
||||
return f"Processed {arg}"
|
||||
|
||||
|
||||
# Example of a Runnable class for testing
|
||||
class MockRunnable(Runnable):
|
||||
# Define necessary methods and properties
|
||||
pass
|
||||
|
||||
|
||||
# Fixture for creating a mock function
|
||||
@pytest.fixture
|
||||
def mock_func():
|
||||
return mock_function
|
||||
|
||||
|
||||
# Fixture for creating a Runnable instance
|
||||
@pytest.fixture
|
||||
def mock_runnable():
|
||||
return MockRunnable()
|
||||
|
||||
|
||||
# Basic functionality tests
|
||||
def test_tool_with_callable(mock_func):
|
||||
# Test creating a tool with a simple callable
|
||||
tool_instance = tool(mock_func)
|
||||
assert isinstance(tool_instance, BaseTool)
|
||||
|
||||
|
||||
def test_tool_with_runnable(mock_runnable):
|
||||
# Test creating a tool with a Runnable instance
|
||||
tool_instance = tool(mock_runnable)
|
||||
assert isinstance(tool_instance, BaseTool)
|
||||
|
||||
|
||||
# ... more basic functionality tests ...
|
||||
|
||||
|
||||
# Argument handling tests
|
||||
def test_tool_with_invalid_argument():
|
||||
# Test passing an invalid argument type
|
||||
with pytest.raises(ValueError):
|
||||
tool(
|
||||
123
|
||||
) # Using an integer instead of a string/callable/Runnable
|
||||
|
||||
|
||||
def test_tool_with_multiple_arguments(mock_func):
|
||||
# Test passing multiple valid arguments
|
||||
tool_instance = tool("mock", mock_func)
|
||||
assert isinstance(tool_instance, BaseTool)
|
||||
|
||||
|
||||
# ... more argument handling tests ...
|
||||
|
||||
|
||||
# Schema inference and application tests
|
||||
class TestSchema(BaseModel):
|
||||
arg: str
|
||||
|
||||
|
||||
def test_tool_with_args_schema(mock_func):
|
||||
# Test passing a custom args_schema
|
||||
tool_instance = tool(mock_func, args_schema=TestSchema)
|
||||
assert tool_instance.args_schema == TestSchema
|
||||
|
||||
|
||||
# ... more schema tests ...
|
||||
|
||||
|
||||
# Exception handling tests
|
||||
def test_tool_function_without_docstring():
|
||||
# Test that a ValueError is raised if the function lacks a docstring
|
||||
def no_doc_func(arg: str) -> str:
|
||||
return arg
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
tool(no_doc_func)
|
||||
|
||||
|
||||
# Test suite starts here
|
||||
class TestTool:
|
||||
# Basic Functionality Tests
|
||||
def test_tool_with_valid_callable_creates_base_tool(
|
||||
self, mock_func
|
||||
):
|
||||
result = tool(mock_func)
|
||||
assert isinstance(result, BaseTool)
|
||||
|
||||
def test_tool_returns_correct_function_name(self, mock_func):
|
||||
result = tool(mock_func)
|
||||
assert result.func.__name__ == "mock_function"
|
||||
|
||||
# Argument Handling Tests
|
||||
def test_tool_with_string_and_runnable(self, mock_runnable):
|
||||
result = tool("mock_runnable", mock_runnable)
|
||||
assert isinstance(result, BaseTool)
|
||||
|
||||
def test_tool_raises_error_with_invalid_arguments(self):
|
||||
with pytest.raises(ValueError):
|
||||
tool(123)
|
||||
|
||||
def test_tool_with_infer_schema_true(self, mock_func):
|
||||
tool(mock_func, infer_schema=True)
|
||||
# Assertions related to schema inference
|
||||
|
||||
# Return Direct Feature Tests
|
||||
def test_tool_with_return_direct_true(self, mock_func):
|
||||
tool(mock_func, return_direct=True)
|
||||
# Assertions for return_direct behavior
|
||||
|
||||
# Error Handling Tests
|
||||
def test_tool_raises_error_without_docstring(self):
|
||||
def no_doc_func(arg: str) -> str:
|
||||
return arg
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
tool(no_doc_func)
|
||||
|
||||
def test_tool_raises_error_runnable_without_object_schema(
|
||||
self, mock_runnable
|
||||
):
|
||||
with pytest.raises(ValueError):
|
||||
tool(mock_runnable)
|
||||
|
||||
# Decorator Behavior Tests
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_tool_function(self):
|
||||
@tool
|
||||
async def async_func(arg: str) -> str:
|
||||
return arg
|
||||
|
||||
# Assertions for async behavior
|
||||
|
||||
# Integration with StructuredTool and Tool Classes
|
||||
def test_integration_with_structured_tool(self, mock_func):
|
||||
result = tool(mock_func)
|
||||
assert isinstance(result, StructuredTool)
|
||||
|
||||
# Concurrency and Async Handling Tests
|
||||
def test_concurrency_in_tool(self, mock_func):
|
||||
# Test related to concurrency
|
||||
pass
|
||||
|
||||
# Mocking and Isolation Tests
|
||||
def test_mocking_external_dependencies(self, mocker):
|
||||
# Use mocker to mock external dependencies
|
||||
pass
|
||||
|
||||
def test_tool_with_different_return_types(self):
|
||||
@tool
|
||||
def return_int(arg: str) -> int:
|
||||
return int(arg)
|
||||
|
||||
result = return_int("123")
|
||||
assert isinstance(result, int)
|
||||
assert result == 123
|
||||
|
||||
@tool
|
||||
def return_bool(arg: str) -> bool:
|
||||
return arg.lower() in ["true", "yes"]
|
||||
|
||||
result = return_bool("true")
|
||||
assert isinstance(result, bool)
|
||||
assert result is True
|
||||
|
||||
# Test with multiple arguments
|
||||
def test_tool_with_multiple_args(self):
|
||||
@tool
|
||||
def concat_strings(a: str, b: str) -> str:
|
||||
return a + b
|
||||
|
||||
result = concat_strings("Hello", "World")
|
||||
assert result == "HelloWorld"
|
||||
|
||||
# Test handling of optional arguments
|
||||
def test_tool_with_optional_args(self):
|
||||
@tool
|
||||
def greet(name: str, greeting: str = "Hello") -> str:
|
||||
return f"{greeting} {name}"
|
||||
|
||||
assert greet("Alice") == "Hello Alice"
|
||||
assert greet("Alice", greeting="Hi") == "Hi Alice"
|
||||
|
||||
# Test with variadic arguments
|
||||
def test_tool_with_variadic_args(self):
|
||||
@tool
|
||||
def sum_numbers(*numbers: int) -> int:
|
||||
return sum(numbers)
|
||||
|
||||
assert sum_numbers(1, 2, 3) == 6
|
||||
assert sum_numbers(10, 20) == 30
|
||||
|
||||
# Test with keyword arguments
|
||||
def test_tool_with_kwargs(self):
|
||||
@tool
|
||||
def build_query(**kwargs) -> str:
|
||||
return "&".join(f"{k}={v}" for k, v in kwargs.items())
|
||||
|
||||
assert build_query(a=1, b=2) == "a=1&b=2"
|
||||
assert build_query(foo="bar") == "foo=bar"
|
||||
|
||||
# Test with mixed types of arguments
|
||||
def test_tool_with_mixed_args(self):
|
||||
@tool
|
||||
def mixed_args(a: int, b: str, *args, **kwargs) -> str:
|
||||
return f"{a}{b}{len(args)}{'-'.join(kwargs.values())}"
|
||||
|
||||
assert mixed_args(1, "b", "c", "d", x="y", z="w") == "1b2y-w"
|
||||
|
||||
# Test error handling with incorrect types
|
||||
def test_tool_error_with_incorrect_types(self):
|
||||
@tool
|
||||
def add_numbers(a: int, b: int) -> int:
|
||||
return a + b
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
add_numbers("1", "2")
|
||||
|
||||
# Test with nested tools
|
||||
def test_nested_tools(self):
|
||||
@tool
|
||||
def inner_tool(arg: str) -> str:
|
||||
return f"Inner {arg}"
|
||||
|
||||
@tool
|
||||
def outer_tool(arg: str) -> str:
|
||||
return f"Outer {inner_tool(arg)}"
|
||||
|
||||
assert outer_tool("Test") == "Outer Inner Test"
|
||||
|
||||
def test_tool_with_global_variable(self):
|
||||
@tool
|
||||
def access_global(arg: str) -> str:
|
||||
return f"{global_var} {arg}"
|
||||
|
||||
assert access_global("Var") == "global Var"
|
||||
|
||||
# Test with environment variables
|
||||
def test_tool_with_env_variables(self, monkeypatch):
|
||||
monkeypatch.setenv("TEST_VAR", "Environment")
|
||||
|
||||
@tool
|
||||
def access_env_variable(arg: str) -> str:
|
||||
import os
|
||||
|
||||
return f"{os.environ['TEST_VAR']} {arg}"
|
||||
|
||||
assert access_env_variable("Var") == "Environment Var"
|
||||
|
||||
# ... [Previous test cases] ...
|
||||
|
||||
# Test with complex data structures
|
||||
def test_tool_with_complex_data_structures(self):
|
||||
@tool
|
||||
def process_data(data: dict) -> list:
|
||||
return [data[key] for key in sorted(data.keys())]
|
||||
|
||||
result = process_data({"b": 2, "a": 1})
|
||||
assert result == [1, 2]
|
||||
|
||||
# Test handling exceptions within the tool function
|
||||
def test_tool_handling_internal_exceptions(self):
|
||||
@tool
|
||||
def function_that_raises(arg: str):
|
||||
if arg == "error":
|
||||
raise ValueError("Error occurred")
|
||||
return arg
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
function_that_raises("error")
|
||||
assert function_that_raises("ok") == "ok"
|
||||
|
||||
# Test with functions returning None
|
||||
def test_tool_with_none_return(self):
|
||||
@tool
|
||||
def return_none(arg: str):
|
||||
return None
|
||||
|
||||
assert return_none("anything") is None
|
||||
|
||||
# Test with lambda functions
|
||||
def test_tool_with_lambda(self):
|
||||
tool_lambda = tool(lambda x: x * 2)
|
||||
assert tool_lambda(3) == 6
|
||||
|
||||
# Test with class methods
|
||||
def test_tool_with_class_method(self):
|
||||
class MyClass:
|
||||
@tool
|
||||
def method(self, arg: str) -> str:
|
||||
return f"Method {arg}"
|
||||
|
||||
obj = MyClass()
|
||||
assert obj.method("test") == "Method test"
|
||||
|
||||
# Test tool function with inheritance
|
||||
def test_tool_with_inheritance(self):
|
||||
class Parent:
|
||||
@tool
|
||||
def parent_method(self, arg: str) -> str:
|
||||
return f"Parent {arg}"
|
||||
|
||||
class Child(Parent):
|
||||
@tool
|
||||
def child_method(self, arg: str) -> str:
|
||||
return f"Child {arg}"
|
||||
|
||||
child_obj = Child()
|
||||
assert child_obj.parent_method("test") == "Parent test"
|
||||
assert child_obj.child_method("test") == "Child test"
|
||||
|
||||
# Test with decorators stacking
|
||||
def test_tool_with_multiple_decorators(self):
|
||||
def another_decorator(func):
|
||||
def wrapper(*args, **kwargs):
|
||||
return f"Decorated {func(*args, **kwargs)}"
|
||||
|
||||
return wrapper
|
||||
|
||||
@tool
|
||||
@another_decorator
|
||||
def decorated_function(arg: str):
|
||||
return f"Function {arg}"
|
||||
|
||||
assert decorated_function("test") == "Decorated Function test"
|
||||
|
||||
# Test tool function when used in a multi-threaded environment
|
||||
def test_tool_in_multithreaded_environment(self):
|
||||
import threading
|
||||
|
||||
@tool
|
||||
def threaded_function(arg: int) -> int:
|
||||
return arg * 2
|
||||
|
||||
results = []
|
||||
|
||||
def thread_target():
|
||||
results.append(threaded_function(5))
|
||||
|
||||
threads = [
|
||||
threading.Thread(target=thread_target) for _ in range(10)
|
||||
]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert results == [10] * 10
|
||||
|
||||
# Test with recursive functions
|
||||
def test_tool_with_recursive_function(self):
|
||||
@tool
|
||||
def recursive_function(n: int) -> int:
|
||||
if n == 0:
|
||||
return 0
|
||||
else:
|
||||
return n + recursive_function(n - 1)
|
||||
|
||||
assert recursive_function(5) == 15
|
||||
|
||||
|
||||
# Additional tests can be added here to cover more scenarios
|
||||
Loading…
Reference in new issue