Compare commits
No commits in common. "d5eb7d400b824841b7654cfd920aa3de613cb2f9" and "b119bac02435b1d8a3c058fa2b2730032827c464" have entirely different histories.
d5eb7d400b
...
b119bac024
|
|
@ -31,11 +31,3 @@ dev = ["pytest>=8.0.0", "pytest-asyncio>=0.23.0", "pytest-cov>=4.1.0", "black>=2
|
||||||
|
|
||||||
[tool.setuptools]
|
[tool.setuptools]
|
||||||
packages = ["luxx"]
|
packages = ["luxx"]
|
||||||
|
|
||||||
[tool.pytest.ini_options]
|
|
||||||
testpaths = ["tests"]
|
|
||||||
python_files = ["test_*.py"]
|
|
||||||
python_classes = ["Test*"]
|
|
||||||
python_functions = ["test_*"]
|
|
||||||
asyncio_mode = "auto"
|
|
||||||
addopts = "-v --tb=short"
|
|
||||||
|
|
|
||||||
|
|
@ -1 +0,0 @@
|
||||||
"""Test suite for Luxx project"""
|
|
||||||
|
|
@ -1,32 +0,0 @@
|
||||||
"""Pytest configuration and fixtures"""
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import pytest
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
# Add project root to path
|
|
||||||
project_root = Path(__file__).parent.parent
|
|
||||||
sys.path.insert(0, str(project_root))
|
|
||||||
|
|
||||||
# Set test environment variables
|
|
||||||
os.environ.setdefault("APP_SECRET_KEY", "test-secret-key-for-testing")
|
|
||||||
os.environ.setdefault("DEEPSEEK_API_KEY", "test-api-key")
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def sample_workspace(tmp_path):
|
|
||||||
"""Create a temporary workspace for testing"""
|
|
||||||
workspace = tmp_path / "test_workspace"
|
|
||||||
workspace.mkdir()
|
|
||||||
return workspace
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def sample_user_context():
|
|
||||||
"""Sample user context for tool testing"""
|
|
||||||
return {
|
|
||||||
"user_id": 1,
|
|
||||||
"username": "test_user",
|
|
||||||
"workspace": "/tmp/test_workspace",
|
|
||||||
"user_permission_level": 3
|
|
||||||
}
|
|
||||||
|
|
@ -1,333 +0,0 @@
|
||||||
"""Tests for LLM response parsing - MiniMax/OpenAI format"""
|
|
||||||
import json
|
|
||||||
import pytest
|
|
||||||
from luxx.services.llm_response import ParsedDelta
|
|
||||||
|
|
||||||
|
|
||||||
class TestMiniMaxParsing:
|
|
||||||
"""Tests for MiniMax/OpenAI streaming format parsing"""
|
|
||||||
|
|
||||||
def test_parse_text_chunk(self):
|
|
||||||
"""Parse text content chunk"""
|
|
||||||
chunk_str = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"Hello","role":"assistant"}}]}'
|
|
||||||
chunk = json.loads(chunk_str[6:]) # Remove "data: "
|
|
||||||
|
|
||||||
delta = chunk["choices"][0]["delta"]
|
|
||||||
assert delta["content"] == "Hello"
|
|
||||||
|
|
||||||
def test_parse_text_accumulation(self):
|
|
||||||
"""Multiple chunks should accumulate to full text"""
|
|
||||||
chunks = [
|
|
||||||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"你好","role":"assistant"}}]}',
|
|
||||||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"！","role":"assistant"}}]}',
|
|
||||||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"有什么","role":"assistant"}}]}',
|
|
||||||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"可以帮助你的吗？","role":"assistant"}}]}',
|
|
||||||
]
|
|
||||||
|
|
||||||
full_text = ""
|
|
||||||
for c in chunks:
|
|
||||||
chunk = json.loads(c)
|
|
||||||
content = chunk["choices"][0]["delta"].get("content", "")
|
|
||||||
full_text += content
|
|
||||||
|
|
||||||
assert full_text == "你好！有什么可以帮助你的吗？"
|
|
||||||
|
|
||||||
def test_parse_finish_chunk(self):
|
|
||||||
"""Parse finish_reason chunk"""
|
|
||||||
chunk_str = 'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{"role":"assistant"}}]}'
|
|
||||||
chunk = json.loads(chunk_str[6:])
|
|
||||||
|
|
||||||
finish_reason = chunk["choices"][0].get("finish_reason")
|
|
||||||
assert finish_reason == "stop"
|
|
||||||
|
|
||||||
def test_parse_usage_chunk(self):
|
|
||||||
"""Parse usage information"""
|
|
||||||
chunk_str = '{"id":"msg_001","usage":{"prompt_tokens":100,"completion_tokens":50,"total_tokens":150}}'
|
|
||||||
chunk = json.loads(chunk_str)
|
|
||||||
|
|
||||||
usage = chunk.get("usage", {})
|
|
||||||
assert usage["prompt_tokens"] == 100
|
|
||||||
assert usage["completion_tokens"] == 50
|
|
||||||
assert usage["total_tokens"] == 150
|
|
||||||
|
|
||||||
def test_parse_empty_content_chunk(self):
|
|
||||||
"""Parse chunk with empty content (just role)"""
|
|
||||||
chunk_str = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}'
|
|
||||||
chunk = json.loads(chunk_str[6:])
|
|
||||||
|
|
||||||
delta = chunk["choices"][0]["delta"]
|
|
||||||
content = delta.get("content", "")
|
|
||||||
assert content == ""
|
|
||||||
|
|
||||||
def test_parse_done_marker(self):
|
|
||||||
"""Parse [DONE] marker"""
|
|
||||||
chunk_str = "data: [DONE]"
|
|
||||||
assert chunk_str.strip().startswith("data: [DONE]")
|
|
||||||
|
|
||||||
|
|
||||||
class TestOpenAIAdapter:
|
|
||||||
"""Tests for OpenAI adapter parsing logic"""
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def adapter(self):
|
|
||||||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
|
||||||
return OpenAIAdapter()
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_text(self, adapter):
|
|
||||||
"""Should parse text content"""
|
|
||||||
chunk = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"Hello"}}]}'
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
|
|
||||||
assert len(deltas) == 1
|
|
||||||
assert deltas[0].text == "Hello"
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_finish(self, adapter):
|
|
||||||
"""Should detect completion"""
|
|
||||||
chunk = 'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{}}]}'
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
|
|
||||||
assert len(deltas) == 1
|
|
||||||
assert deltas[0].is_complete is True
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_empty_content(self, adapter):
|
|
||||||
"""Should skip empty content chunks"""
|
|
||||||
chunk = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"role":"assistant"}}]}'
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
|
|
||||||
# Empty content without finish_reason should be skipped
|
|
||||||
assert len(deltas) == 0
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_done_marker(self, adapter):
|
|
||||||
"""Should handle [DONE] marker"""
|
|
||||||
chunk = "data: [DONE]"
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
|
|
||||||
assert len(deltas) == 1
|
|
||||||
assert deltas[0].is_complete is True
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_invalid_json(self, adapter):
|
|
||||||
"""Should handle invalid JSON gracefully"""
|
|
||||||
chunk = "not valid json"
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
|
|
||||||
assert len(deltas) == 0
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_empty_chunk(self, adapter):
|
|
||||||
"""Should handle empty chunk"""
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk("")]
|
|
||||||
assert len(deltas) == 0
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_parse_stream_whitespace_chunk(self, adapter):
|
|
||||||
"""Should handle whitespace-only chunk"""
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(" \n")]
|
|
||||||
assert len(deltas) == 0
|
|
||||||
|
|
||||||
|
|
||||||
class TestBuildRequest:
|
|
||||||
"""Tests for request building"""
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def adapter(self):
|
|
||||||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
|
||||||
return OpenAIAdapter()
|
|
||||||
|
|
||||||
def test_build_request_basic(self, adapter):
|
|
||||||
"""Should build basic request"""
|
|
||||||
body, headers = adapter.build_request(
|
|
||||||
model="MiniMax-M2.5",
|
|
||||||
messages=[{"role": "user", "content": "Hello"}]
|
|
||||||
)
|
|
||||||
|
|
||||||
assert body["model"] == "MiniMax-M2.5"
|
|
||||||
assert body["messages"] == [{"role": "user", "content": "Hello"}]
|
|
||||||
assert body["stream"] is True
|
|
||||||
|
|
||||||
def test_build_request_with_tools(self, adapter):
|
|
||||||
"""Should include tools in request"""
|
|
||||||
tools = [{"type": "function", "function": {"name": "test", "parameters": {}}}]
|
|
||||||
body, _ = adapter.build_request(
|
|
||||||
model="MiniMax-M2.5",
|
|
||||||
messages=[],
|
|
||||||
tools=tools
|
|
||||||
)
|
|
||||||
|
|
||||||
assert "tools" in body
|
|
||||||
assert body["tool_choice"] == "auto"
|
|
||||||
|
|
||||||
def test_build_request_with_temperature(self, adapter):
|
|
||||||
"""Should include temperature"""
|
|
||||||
body, _ = adapter.build_request(
|
|
||||||
model="MiniMax-M2.5",
|
|
||||||
messages=[],
|
|
||||||
temperature=0.7
|
|
||||||
)
|
|
||||||
|
|
||||||
assert body["temperature"] == 0.7
|
|
||||||
|
|
||||||
def test_build_request_with_max_tokens(self, adapter):
|
|
||||||
"""Should include max_tokens"""
|
|
||||||
body, _ = adapter.build_request(
|
|
||||||
model="MiniMax-M2.5",
|
|
||||||
messages=[],
|
|
||||||
max_tokens=1000
|
|
||||||
)
|
|
||||||
|
|
||||||
assert body["max_tokens"] == 1000
|
|
||||||
|
|
||||||
|
|
||||||
class TestParsedDelta:
|
|
||||||
"""Tests for ParsedDelta dataclass"""
|
|
||||||
|
|
||||||
def test_default_values(self):
|
|
||||||
"""Should have correct defaults"""
|
|
||||||
delta = ParsedDelta()
|
|
||||||
assert delta.text == ""
|
|
||||||
assert delta.thinking == ""
|
|
||||||
assert delta.tool_call is None
|
|
||||||
assert delta.usage == {}
|
|
||||||
assert delta.is_complete is False
|
|
||||||
|
|
||||||
def test_with_text(self):
|
|
||||||
"""Should accept text content"""
|
|
||||||
delta = ParsedDelta(text="Hello world")
|
|
||||||
assert delta.text == "Hello world"
|
|
||||||
|
|
||||||
def test_with_usage(self):
|
|
||||||
"""Should accept usage dict"""
|
|
||||||
delta = ParsedDelta(usage={"prompt_tokens": 10, "completion_tokens": 5})
|
|
||||||
assert delta.usage["prompt_tokens"] == 10
|
|
||||||
|
|
||||||
def test_with_complete_flag(self):
|
|
||||||
"""Should accept is_complete flag"""
|
|
||||||
delta = ParsedDelta(is_complete=True)
|
|
||||||
assert delta.is_complete is True
|
|
||||||
|
|
||||||
|
|
||||||
class TestEndToEndStreaming:
|
|
||||||
"""End-to-end streaming simulation tests"""
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def adapter(self):
|
|
||||||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
|
||||||
return OpenAIAdapter()
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_full_text_stream(self, adapter):
|
|
||||||
"""Simulate full text response stream"""
|
|
||||||
chunks = [
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"用户","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"用中文","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"说","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\"你好\"","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"，这是","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"一个简单的","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"问候。","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\n","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"我应该","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"用中文","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"友好地","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"回应。","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\n\n","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\n\n你好！","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"有什么","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"我可以","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"帮助你的吗？","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{"role":"assistant"}}]}',
|
|
||||||
]
|
|
||||||
|
|
||||||
full_text = ""
|
|
||||||
is_complete = False
|
|
||||||
|
|
||||||
for chunk in chunks:
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
for delta in deltas:
|
|
||||||
if delta.is_complete:
|
|
||||||
is_complete = True
|
|
||||||
if delta.text:
|
|
||||||
full_text += delta.text
|
|
||||||
|
|
||||||
expected = "用户用中文说\"你好\"，这是一个简单的问候。我应该用中文友好地回应。\n\n你好！有什么我可以帮助你的吗？"
|
|
||||||
assert full_text == expected
|
|
||||||
assert is_complete is True
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_empty_stream_between_finish(self, adapter):
|
|
||||||
"""Handle empty content chunks before finish"""
|
|
||||||
chunks = [
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"Hello","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}',
|
|
||||||
'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{"role":"assistant"}}]}',
|
|
||||||
]
|
|
||||||
|
|
||||||
full_text = ""
|
|
||||||
for chunk in chunks:
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
for delta in deltas:
|
|
||||||
if delta.text:
|
|
||||||
full_text += delta.text
|
|
||||||
|
|
||||||
assert full_text == "Hello"
|
|
||||||
|
|
||||||
|
|
||||||
class TestToolCallParsing:
|
|
||||||
"""Tests for tool call parsing"""
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def adapter(self):
|
|
||||||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
|
||||||
return OpenAIAdapter()
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_tool_call_chunk(self, adapter):
|
|
||||||
"""Parse tool call chunk"""
|
|
||||||
chunk = json.dumps({
|
|
||||||
"id": "chatcmpl_001",
|
|
||||||
"choices": [{
|
|
||||||
"index": 0,
|
|
||||||
"delta": {
|
|
||||||
"tool_calls": [{
|
|
||||||
"index": 0,
|
|
||||||
"id": "call_abc123",
|
|
||||||
"type": "function",
|
|
||||||
"function": {
|
|
||||||
"name": "web_search",
|
|
||||||
"arguments": '{"query":'
|
|
||||||
}
|
|
||||||
}]
|
|
||||||
}
|
|
||||||
}]
|
|
||||||
})
|
|
||||||
|
|
||||||
# The current adapter doesn't yield tool_call deltas
|
|
||||||
# This test documents current behavior
|
|
||||||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
|
||||||
# Content is empty, no delta yielded
|
|
||||||
assert len(deltas) == 0
|
|
||||||
|
|
||||||
def test_tool_call_accumulation(self):
|
|
||||||
"""Simulate tool call argument accumulation"""
|
|
||||||
chunks = [
|
|
||||||
{"function": {"name": "python_eval", "arguments": "{"}},
|
|
||||||
{"function": {"name": "", "arguments": '"expr": '}},
|
|
||||||
{"function": {"name": "", "arguments": '"1 + 1"'}},
|
|
||||||
{"function": {"name": "", "arguments": "}"}},
|
|
||||||
]
|
|
||||||
|
|
||||||
accumulated = {"name": "", "arguments": ""}
|
|
||||||
for c in chunks:
|
|
||||||
if c["function"]["name"]:
|
|
||||||
accumulated["name"] += c["function"]["name"]
|
|
||||||
if c["function"]["arguments"]:
|
|
||||||
accumulated["arguments"] += c["function"]["arguments"]
|
|
||||||
|
|
||||||
assert accumulated["name"] == "python_eval"
|
|
||||||
assert accumulated["arguments"] == '{"expr": "1 + 1"}'
|
|
||||||
|
|
@ -1,48 +0,0 @@
|
||||||
"""Tests for config module"""
|
|
||||||
|
|
||||||
class TestConfig:
|
|
||||||
"""Tests for Config class"""
|
|
||||||
|
|
||||||
def test_config_singleton(self):
|
|
||||||
"""Should return same instance"""
|
|
||||||
from luxx.config import config, Config
|
|
||||||
config1 = Config()
|
|
||||||
config2 = Config()
|
|
||||||
assert config1 is config2
|
|
||||||
assert config is config1
|
|
||||||
|
|
||||||
def test_get_with_default(self):
|
|
||||||
"""Should return default value for missing key"""
|
|
||||||
from luxx.config import config
|
|
||||||
result = config.get("nonexistent.key", "default_value")
|
|
||||||
assert result == "default_value"
|
|
||||||
|
|
||||||
def test_get_nested_key(self):
|
|
||||||
"""Should support dot-separated keys"""
|
|
||||||
from luxx.config import config
|
|
||||||
# These should return configured or default values
|
|
||||||
secret = config.get("app.secret_key")
|
|
||||||
assert secret is not None
|
|
||||||
|
|
||||||
def test_properties_have_defaults(self):
|
|
||||||
"""All properties should have sensible defaults"""
|
|
||||||
from luxx.config import config
|
|
||||||
assert isinstance(config.debug, bool)
|
|
||||||
assert isinstance(config.app_host, str)
|
|
||||||
assert isinstance(config.app_port, int)
|
|
||||||
assert isinstance(config.database_url, str)
|
|
||||||
assert config.app_port == 8000
|
|
||||||
|
|
||||||
def test_tools_config_properties(self):
|
|
||||||
"""Tools configuration properties should work"""
|
|
||||||
from luxx.config import config
|
|
||||||
assert config.tools_enable_cache is not None
|
|
||||||
assert config.tools_cache_ttl > 0
|
|
||||||
assert config.tools_max_workers > 0
|
|
||||||
assert config.tools_max_iterations > 0
|
|
||||||
|
|
||||||
def test_llm_config_properties(self):
|
|
||||||
"""LLM configuration properties should work"""
|
|
||||||
from luxx.config import config
|
|
||||||
assert config.llm_provider is not None
|
|
||||||
assert config.llm_api_url is not None
|
|
||||||
|
|
@ -1,110 +0,0 @@
|
||||||
"""Tests for utils/helpers module"""
|
|
||||||
from luxx.utils.helpers import (
|
|
||||||
generate_id,
|
|
||||||
hash_password,
|
|
||||||
verify_password,
|
|
||||||
create_access_token,
|
|
||||||
decode_access_token,
|
|
||||||
success_response,
|
|
||||||
error_response
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestGenerateId:
|
|
||||||
"""Tests for generate_id function"""
|
|
||||||
|
|
||||||
def test_generate_id_returns_string(self):
|
|
||||||
"""Should return a string"""
|
|
||||||
result = generate_id()
|
|
||||||
assert isinstance(result, str)
|
|
||||||
|
|
||||||
def test_generate_id_with_prefix(self):
|
|
||||||
"""Should return id with prefix"""
|
|
||||||
result = generate_id("task")
|
|
||||||
assert result.startswith("task_")
|
|
||||||
|
|
||||||
def test_generate_id_unique(self):
|
|
||||||
"""Should generate unique ids"""
|
|
||||||
ids = [generate_id() for _ in range(100)]
|
|
||||||
assert len(set(ids)) == 100
|
|
||||||
|
|
||||||
|
|
||||||
class TestPasswordHashing:
|
|
||||||
"""Tests for password hashing functions"""
|
|
||||||
|
|
||||||
def test_hash_password_returns_string(self):
|
|
||||||
"""Should return a hashed string"""
|
|
||||||
password = "test_password_123"
|
|
||||||
hashed = hash_password(password)
|
|
||||||
assert isinstance(hashed, str)
|
|
||||||
assert hashed != password
|
|
||||||
|
|
||||||
def test_verify_password_correct(self):
|
|
||||||
"""Should return True for correct password"""
|
|
||||||
password = "test_password_123"
|
|
||||||
hashed = hash_password(password)
|
|
||||||
assert verify_password(password, hashed) is True
|
|
||||||
|
|
||||||
def test_verify_password_incorrect(self):
|
|
||||||
"""Should return False for incorrect password"""
|
|
||||||
password = "test_password_123"
|
|
||||||
wrong_password = "wrong_password"
|
|
||||||
hashed = hash_password(password)
|
|
||||||
assert verify_password(wrong_password, hashed) is False
|
|
||||||
|
|
||||||
|
|
||||||
class TestJWTToken:
|
|
||||||
"""Tests for JWT token functions"""
|
|
||||||
|
|
||||||
def test_create_access_token_returns_string(self):
|
|
||||||
"""Should return a JWT token string"""
|
|
||||||
token = create_access_token({"user_id": 1})
|
|
||||||
assert isinstance(token, str)
|
|
||||||
assert len(token) > 0
|
|
||||||
|
|
||||||
def test_decode_access_token_valid(self):
|
|
||||||
"""Should decode valid token"""
|
|
||||||
payload = {"user_id": 1, "username": "test"}
|
|
||||||
token = create_access_token(payload)
|
|
||||||
decoded = decode_access_token(token)
|
|
||||||
assert decoded is not None
|
|
||||||
assert decoded["user_id"] == 1
|
|
||||||
assert decoded["username"] == "test"
|
|
||||||
|
|
||||||
def test_decode_access_token_invalid(self):
|
|
||||||
"""Should return None for invalid token"""
|
|
||||||
result = decode_access_token("invalid.token.here")
|
|
||||||
assert result is None
|
|
||||||
|
|
||||||
|
|
||||||
class TestResponseWrappers:
|
|
||||||
"""Tests for response wrapper functions"""
|
|
||||||
|
|
||||||
def test_success_response_format(self):
|
|
||||||
"""Should return correct success format"""
|
|
||||||
result = success_response({"key": "value"}, "Success message")
|
|
||||||
assert result["success"] is True
|
|
||||||
assert result["message"] == "Success message"
|
|
||||||
assert result["data"] == {"key": "value"}
|
|
||||||
|
|
||||||
def test_success_response_default(self):
|
|
||||||
"""Should use default values"""
|
|
||||||
result = success_response()
|
|
||||||
assert result["success"] is True
|
|
||||||
assert result["message"] == "Success"
|
|
||||||
assert result["data"] is None
|
|
||||||
|
|
||||||
def test_error_response_format(self):
|
|
||||||
"""Should return correct error format"""
|
|
||||||
result = error_response("Error occurred", code=404)
|
|
||||||
assert result["success"] is False
|
|
||||||
assert result["message"] == "Error occurred"
|
|
||||||
assert result["code"] == 404
|
|
||||||
assert "errors" not in result
|
|
||||||
|
|
||||||
def test_error_response_with_errors(self):
|
|
||||||
"""Should include errors field"""
|
|
||||||
errors = {"field": ["required"]}
|
|
||||||
result = error_response("Validation failed", code=400, errors=errors)
|
|
||||||
assert result["success"] is False
|
|
||||||
assert result["errors"] == errors
|
|
||||||
|
|
@ -1,223 +0,0 @@
|
||||||
"""Tests for task module"""
|
|
||||||
import pytest
|
|
||||||
from luxx.services.task import (
|
|
||||||
Task,
|
|
||||||
Step,
|
|
||||||
TaskGraph,
|
|
||||||
TaskService,
|
|
||||||
TaskStatus,
|
|
||||||
StepStatus,
|
|
||||||
task_service
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestStep:
|
|
||||||
"""Tests for Step dataclass"""
|
|
||||||
|
|
||||||
def test_step_creation(self):
|
|
||||||
"""Should create step with required fields"""
|
|
||||||
step = Step(id="step_1", name="Test Step")
|
|
||||||
assert step.id == "step_1"
|
|
||||||
assert step.name == "Test Step"
|
|
||||||
assert step.status == StepStatus.PENDING
|
|
||||||
assert step.depends_on == []
|
|
||||||
|
|
||||||
def test_step_with_dependencies(self):
|
|
||||||
"""Should create step with dependencies"""
|
|
||||||
step = Step(
|
|
||||||
id="step_2",
|
|
||||||
name="Dependent Step",
|
|
||||||
depends_on=["step_1"]
|
|
||||||
)
|
|
||||||
assert "step_1" in step.depends_on
|
|
||||||
|
|
||||||
def test_step_to_dict(self):
|
|
||||||
"""Should convert step to dict"""
|
|
||||||
step = Step(id="step_1", name="Test")
|
|
||||||
result = step.to_dict()
|
|
||||||
assert isinstance(result, dict)
|
|
||||||
assert result["id"] == "step_1"
|
|
||||||
assert result["name"] == "Test"
|
|
||||||
assert result["status"] == "pending"
|
|
||||||
|
|
||||||
|
|
||||||
class TestTask:
|
|
||||||
"""Tests for Task dataclass"""
|
|
||||||
|
|
||||||
def test_task_creation(self):
|
|
||||||
"""Should create task with required fields"""
|
|
||||||
task = Task(id="task_1", name="Test Task")
|
|
||||||
assert task.id == "task_1"
|
|
||||||
assert task.name == "Test Task"
|
|
||||||
assert task.status == TaskStatus.PENDING
|
|
||||||
assert task.steps == []
|
|
||||||
|
|
||||||
def test_task_with_steps(self):
|
|
||||||
"""Should create task with steps"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
step2 = Step(id="step_2", name="Step 2")
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
assert len(task.steps) == 2
|
|
||||||
|
|
||||||
def test_task_to_dict(self):
|
|
||||||
"""Should convert task to dict"""
|
|
||||||
task = Task(id="task_1", name="Test", goal="Complete task")
|
|
||||||
result = task.to_dict()
|
|
||||||
assert result["id"] == "task_1"
|
|
||||||
assert result["goal"] == "Complete task"
|
|
||||||
assert result["status"] == "pending"
|
|
||||||
|
|
||||||
|
|
||||||
class TestTaskGraph:
|
|
||||||
"""Tests for TaskGraph class"""
|
|
||||||
|
|
||||||
def test_graph_creation(self):
|
|
||||||
"""Should create graph from task"""
|
|
||||||
task = Task(id="task_1", name="Test")
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
assert graph.task is task
|
|
||||||
|
|
||||||
def test_topological_sort_no_dependencies(self):
|
|
||||||
"""Should sort steps without dependencies"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
step2 = Step(id="step_2", name="Step 2")
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
sorted_steps = graph.topological_sort()
|
|
||||||
assert len(sorted_steps) == 2
|
|
||||||
|
|
||||||
def test_topological_sort_with_dependencies(self):
|
|
||||||
"""Should respect dependencies in sort"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
sorted_steps = graph.topological_sort()
|
|
||||||
ids = [s.id for s in sorted_steps]
|
|
||||||
assert ids.index("step_1") < ids.index("step_2")
|
|
||||||
|
|
||||||
def test_get_ready_steps(self):
|
|
||||||
"""Should return steps ready to execute"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
ready = graph.get_ready_steps([])
|
|
||||||
assert len(ready) == 1
|
|
||||||
assert ready[0].id == "step_1"
|
|
||||||
|
|
||||||
def test_get_ready_steps_after_completion(self):
|
|
||||||
"""Should return dependent steps after completion"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
ready = graph.get_ready_steps(["step_1"])
|
|
||||||
assert len(ready) == 1
|
|
||||||
assert ready[0].id == "step_2"
|
|
||||||
|
|
||||||
def test_detect_cycles_no_cycle(self):
|
|
||||||
"""Should return empty for no cycles"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
step2 = Step(id="step_2", name="Step 2")
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
cycles = graph.detect_cycles()
|
|
||||||
assert cycles == []
|
|
||||||
|
|
||||||
def test_detect_cycles_with_cycle(self):
|
|
||||||
"""Should detect circular dependency"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1", depends_on=["step_2"])
|
|
||||||
step2 = Step(id="step_2", name="Step 2", depends_on=["step_1"])
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1, step2])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
cycles = graph.detect_cycles()
|
|
||||||
assert len(cycles) > 0
|
|
||||||
|
|
||||||
def test_validate_valid_graph(self):
|
|
||||||
"""Should validate valid graph"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1")
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
is_valid, error = graph.validate()
|
|
||||||
assert is_valid is True
|
|
||||||
assert error is None
|
|
||||||
|
|
||||||
def test_validate_invalid_dependency(self):
|
|
||||||
"""Should fail validation for invalid dependency"""
|
|
||||||
step1 = Step(id="step_1", name="Step 1", depends_on=["nonexistent"])
|
|
||||||
task = Task(id="task_1", name="Test", steps=[step1])
|
|
||||||
graph = TaskGraph(task)
|
|
||||||
is_valid, error = graph.validate()
|
|
||||||
assert is_valid is False
|
|
||||||
assert error is not None
|
|
||||||
|
|
||||||
|
|
||||||
class TestTaskService:
|
|
||||||
"""Tests for TaskService class"""
|
|
||||||
|
|
||||||
def test_create_task(self):
|
|
||||||
"""Should create a new task"""
|
|
||||||
service = TaskService()
|
|
||||||
task = service.create_task(name="Test Task", goal="Complete test")
|
|
||||||
assert task is not None
|
|
||||||
assert task.name == "Test Task"
|
|
||||||
assert task.goal == "Complete test"
|
|
||||||
|
|
||||||
def test_create_task_with_steps(self):
|
|
||||||
"""Should create task with steps"""
|
|
||||||
service = TaskService()
|
|
||||||
steps = [
|
|
||||||
{"name": "Step 1", "description": "First step"},
|
|
||||||
{"name": "Step 2", "description": "Second step"}
|
|
||||||
]
|
|
||||||
task = service.create_task(name="Test", goal="Goal", steps=steps)
|
|
||||||
assert len(task.steps) == 2
|
|
||||||
|
|
||||||
def test_get_task(self):
|
|
||||||
"""Should retrieve task by id"""
|
|
||||||
service = TaskService()
|
|
||||||
created = service.create_task(name="Test", goal="Goal")
|
|
||||||
retrieved = service.get_task(created.id)
|
|
||||||
assert retrieved is not None
|
|
||||||
assert retrieved.id == created.id
|
|
||||||
|
|
||||||
def test_get_nonexistent_task(self):
|
|
||||||
"""Should return None for nonexistent task"""
|
|
||||||
service = TaskService()
|
|
||||||
result = service.get_task("nonexistent_id")
|
|
||||||
assert result is None
|
|
||||||
|
|
||||||
def test_update_task_status(self):
|
|
||||||
"""Should update task status"""
|
|
||||||
service = TaskService()
|
|
||||||
task = service.create_task(name="Test", goal="Goal")
|
|
||||||
updated = service.update_task_status(task.id, TaskStatus.RUNNING)
|
|
||||||
assert updated is not None
|
|
||||||
assert updated.status == TaskStatus.RUNNING
|
|
||||||
|
|
||||||
def test_add_steps(self):
|
|
||||||
"""Should add steps to existing task"""
|
|
||||||
service = TaskService()
|
|
||||||
task = service.create_task(name="Test", goal="Goal")
|
|
||||||
steps = [{"name": "New Step"}]
|
|
||||||
added = service.add_steps(task.id, steps)
|
|
||||||
assert added is not None
|
|
||||||
assert len(added) == 1
|
|
||||||
assert len(task.steps) == 1
|
|
||||||
|
|
||||||
def test_delete_task(self):
|
|
||||||
"""Should delete task"""
|
|
||||||
service = TaskService()
|
|
||||||
task = service.create_task(name="Test", goal="Goal")
|
|
||||||
result = service.delete_task(task.id)
|
|
||||||
assert result is True
|
|
||||||
assert service.get_task(task.id) is None
|
|
||||||
|
|
||||||
def test_build_graph(self):
|
|
||||||
"""Should build graph for task"""
|
|
||||||
service = TaskService()
|
|
||||||
task = service.create_task(name="Test", goal="Goal")
|
|
||||||
graph = service.build_graph(task.id)
|
|
||||||
assert graph is not None
|
|
||||||
assert isinstance(graph, TaskGraph)
|
|
||||||
|
|
@ -1,313 +0,0 @@
|
||||||
"""Tests for tools module"""
|
|
||||||
import pytest
|
|
||||||
from luxx.tools.core import (
|
|
||||||
ToolContext,
|
|
||||||
ToolDefinition,
|
|
||||||
ToolResult,
|
|
||||||
ToolRegistry,
|
|
||||||
CommandPermission
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestToolContext:
|
|
||||||
"""Tests for ToolContext dataclass"""
|
|
||||||
|
|
||||||
def test_tool_context_creation(self):
|
|
||||||
"""Should create context with default values"""
|
|
||||||
ctx = ToolContext()
|
|
||||||
assert ctx.workspace is None
|
|
||||||
assert ctx.user_id is None
|
|
||||||
assert ctx.username is None
|
|
||||||
assert ctx.extra == {}
|
|
||||||
|
|
||||||
def test_tool_context_with_values(self):
|
|
||||||
"""Should create context with provided values"""
|
|
||||||
ctx = ToolContext(
|
|
||||||
workspace="/workspace/test",
|
|
||||||
user_id=1,
|
|
||||||
username="testuser",
|
|
||||||
extra={"key": "value"}
|
|
||||||
)
|
|
||||||
assert ctx.workspace == "/workspace/test"
|
|
||||||
assert ctx.user_id == 1
|
|
||||||
assert ctx.username == "testuser"
|
|
||||||
assert ctx.extra["key"] == "value"
|
|
||||||
|
|
||||||
|
|
||||||
class TestToolDefinition:
|
|
||||||
"""Tests for ToolDefinition dataclass"""
|
|
||||||
|
|
||||||
def test_tool_definition_creation(self):
|
|
||||||
"""Should create tool definition"""
|
|
||||||
def handler(args):
|
|
||||||
return {"result": "ok"}
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="test_tool",
|
|
||||||
description="A test tool",
|
|
||||||
parameters={"type": "object"},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
assert tool.name == "test_tool"
|
|
||||||
assert tool.description == "A test tool"
|
|
||||||
assert tool.category == "general"
|
|
||||||
assert tool.required_permission == CommandPermission.READ_ONLY
|
|
||||||
|
|
||||||
def test_tool_definition_to_openai_format(self):
|
|
||||||
"""Should convert to OpenAI format"""
|
|
||||||
def handler(args):
|
|
||||||
return {"result": "ok"}
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="test_tool",
|
|
||||||
description="A test tool",
|
|
||||||
parameters={"type": "object", "properties": {}},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
result = tool.to_openai_format()
|
|
||||||
assert result["type"] == "function"
|
|
||||||
assert result["function"]["name"] == "test_tool"
|
|
||||||
|
|
||||||
|
|
||||||
class TestToolResult:
|
|
||||||
"""Tests for ToolResult dataclass"""
|
|
||||||
|
|
||||||
def test_tool_result_success(self):
|
|
||||||
"""Should create success result"""
|
|
||||||
result = ToolResult(success=True, data={"key": "value"})
|
|
||||||
assert result.success is True
|
|
||||||
assert result.data["key"] == "value"
|
|
||||||
assert result.error is None
|
|
||||||
|
|
||||||
def test_tool_result_failure(self):
|
|
||||||
"""Should create failure result"""
|
|
||||||
result = ToolResult(success=False, error="Something went wrong")
|
|
||||||
assert result.success is False
|
|
||||||
assert result.error == "Something went wrong"
|
|
||||||
|
|
||||||
def test_tool_result_to_dict(self):
|
|
||||||
"""Should convert to dictionary"""
|
|
||||||
result = ToolResult(success=True, data={"key": "value"})
|
|
||||||
d = result.to_dict()
|
|
||||||
assert isinstance(d, dict)
|
|
||||||
assert d["success"] is True
|
|
||||||
assert d["data"]["key"] == "value"
|
|
||||||
|
|
||||||
def test_tool_result_ok_factory(self):
|
|
||||||
"""Should use ok() factory method"""
|
|
||||||
result = ToolResult.ok({"result": "success"})
|
|
||||||
assert result.success is True
|
|
||||||
assert result.data == {"result": "success"}
|
|
||||||
|
|
||||||
def test_tool_result_fail_factory(self):
|
|
||||||
"""Should use fail() factory method"""
|
|
||||||
result = ToolResult.fail("Error occurred")
|
|
||||||
assert result.success is False
|
|
||||||
assert result.error == "Error occurred"
|
|
||||||
|
|
||||||
|
|
||||||
class TestToolRegistry:
|
|
||||||
"""Tests for ToolRegistry class"""
|
|
||||||
|
|
||||||
def test_registry_singleton(self):
|
|
||||||
"""Should return same instance"""
|
|
||||||
reg1 = ToolRegistry()
|
|
||||||
reg2 = ToolRegistry()
|
|
||||||
assert reg1 is reg2
|
|
||||||
|
|
||||||
def test_register_tool(self):
|
|
||||||
"""Should register a tool"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear() # Start fresh
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return {"result": "ok"}
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="my_tool",
|
|
||||||
description="My test tool",
|
|
||||||
parameters={},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
registry.register(tool)
|
|
||||||
assert registry.get("my_tool") is not None
|
|
||||||
assert registry.tool_count() == 1
|
|
||||||
|
|
||||||
def test_get_nonexistent_tool(self):
|
|
||||||
"""Should return None for nonexistent tool"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
assert registry.get("nonexistent") is None
|
|
||||||
|
|
||||||
def test_list_all_tools(self):
|
|
||||||
"""Should list all registered tools"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return {}
|
|
||||||
|
|
||||||
tool1 = ToolDefinition(name="tool1", description="Tool 1", parameters={}, handler=handler)
|
|
||||||
tool2 = ToolDefinition(name="tool2", description="Tool 2", parameters={}, handler=handler)
|
|
||||||
registry.register(tool1)
|
|
||||||
registry.register(tool2)
|
|
||||||
|
|
||||||
tools = registry.list_all()
|
|
||||||
assert len(tools) == 2
|
|
||||||
|
|
||||||
def test_list_by_category(self):
|
|
||||||
"""Should filter tools by category"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return {}
|
|
||||||
|
|
||||||
tool1 = ToolDefinition(
|
|
||||||
name="tool1", description="Tool 1", parameters={},
|
|
||||||
handler=handler, category="code"
|
|
||||||
)
|
|
||||||
tool2 = ToolDefinition(
|
|
||||||
name="tool2", description="Tool 2", parameters={},
|
|
||||||
handler=handler, category="file"
|
|
||||||
)
|
|
||||||
registry.register(tool1)
|
|
||||||
registry.register(tool2)
|
|
||||||
|
|
||||||
code_tools = registry.list_by_category("code")
|
|
||||||
assert len(code_tools) == 1
|
|
||||||
|
|
||||||
def test_execute_tool(self):
|
|
||||||
"""Should execute a tool"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return {"executed": True, "args": args}
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="test_tool",
|
|
||||||
description="Test tool",
|
|
||||||
parameters={},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
registry.register(tool)
|
|
||||||
|
|
||||||
result = registry.execute("test_tool", {"input": "value"})
|
|
||||||
# Direct handler returns are passed through as-is
|
|
||||||
assert result["executed"] is True
|
|
||||||
assert result["args"]["input"] == "value"
|
|
||||||
|
|
||||||
def test_execute_tool_with_tool_result(self):
|
|
||||||
"""Should return ToolResult when handler returns ToolResult"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return ToolResult.ok({"executed": True})
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="test_tool",
|
|
||||||
description="Test tool",
|
|
||||||
parameters={},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
registry.register(tool)
|
|
||||||
|
|
||||||
result = registry.execute("test_tool", {})
|
|
||||||
assert result["success"] is True
|
|
||||||
assert result["data"]["executed"] is True
|
|
||||||
|
|
||||||
def test_execute_nonexistent_tool(self):
|
|
||||||
"""Should return error for nonexistent tool"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
result = registry.execute("nonexistent", {})
|
|
||||||
assert result["success"] is False
|
|
||||||
assert "not found" in result["error"]
|
|
||||||
|
|
||||||
def test_execute_with_context(self):
|
|
||||||
"""Should pass context to handler"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
received_context = None
|
|
||||||
|
|
||||||
def handler(args, context=None):
|
|
||||||
nonlocal received_context
|
|
||||||
received_context = context
|
|
||||||
return ToolResult.ok({"received": True})
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="test_tool",
|
|
||||||
description="Test tool",
|
|
||||||
parameters={},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
registry.register(tool)
|
|
||||||
|
|
||||||
ctx = ToolContext(user_id=1, username="test")
|
|
||||||
registry.execute("test_tool", {}, context=ctx)
|
|
||||||
assert received_context is not None
|
|
||||||
assert received_context.user_id == 1
|
|
||||||
|
|
||||||
def test_permission_check(self):
|
|
||||||
"""Should check user permission"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return ToolResult.ok({"ok": True})
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="admin_tool",
|
|
||||||
description="Admin tool",
|
|
||||||
parameters={},
|
|
||||||
handler=handler,
|
|
||||||
required_permission=CommandPermission.ADMIN
|
|
||||||
)
|
|
||||||
registry.register(tool)
|
|
||||||
|
|
||||||
# User with low permission
|
|
||||||
ctx = ToolContext(
|
|
||||||
user_id=1,
|
|
||||||
extra={"user_permission_level": CommandPermission.READ_ONLY}
|
|
||||||
)
|
|
||||||
result = registry.execute("admin_tool", {}, context=ctx)
|
|
||||||
assert result["success"] is False
|
|
||||||
assert "Permission denied" in result["error"]
|
|
||||||
|
|
||||||
def test_remove_tool(self):
|
|
||||||
"""Should remove a tool"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return {}
|
|
||||||
|
|
||||||
tool = ToolDefinition(
|
|
||||||
name="removable",
|
|
||||||
description="To be removed",
|
|
||||||
parameters={},
|
|
||||||
handler=handler
|
|
||||||
)
|
|
||||||
registry.register(tool)
|
|
||||||
assert registry.get("removable") is not None
|
|
||||||
|
|
||||||
registry.remove("removable")
|
|
||||||
assert registry.get("removable") is None
|
|
||||||
|
|
||||||
def test_clear_tools(self):
|
|
||||||
"""Should clear all tools"""
|
|
||||||
registry = ToolRegistry()
|
|
||||||
registry.clear()
|
|
||||||
|
|
||||||
def handler(args):
|
|
||||||
return {}
|
|
||||||
|
|
||||||
tool = ToolDefinition(name="tool1", description="", parameters={}, handler=handler)
|
|
||||||
registry.register(tool)
|
|
||||||
assert registry.tool_count() > 0
|
|
||||||
|
|
||||||
registry.clear()
|
|
||||||
assert registry.tool_count() == 0
|
|
||||||
Loading…
Reference in New Issue