157 lines
4.7 KiB
Python
157 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Test script for agent components."""
|
|
import sys
|
|
import os
|
|
|
|
# Add parent directory to path
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
def test_imports():
|
|
"""Test that all agent modules can be imported."""
|
|
print("Testing imports...")
|
|
|
|
try:
|
|
from agent.main import Agent
|
|
print("✓ agent.main imported successfully")
|
|
except Exception as e:
|
|
print(f"✗ Failed to import agent.main: {e}")
|
|
return False
|
|
|
|
try:
|
|
from agent.task_executor import TaskExecutor
|
|
print("✓ agent.task_executor imported successfully")
|
|
except Exception as e:
|
|
print(f"✗ Failed to import agent.task_executor: {e}")
|
|
return False
|
|
|
|
try:
|
|
from agent.handoff_logic import HandoffDecision, should_handoff
|
|
print("✓ agent.handoff_logic imported successfully")
|
|
except Exception as e:
|
|
print(f"✗ Failed to import agent.handoff_logic: {e}")
|
|
return False
|
|
|
|
try:
|
|
from agent.git_operations import GitOperations
|
|
print("✓ agent.git_operations imported successfully")
|
|
except Exception as e:
|
|
print(f"✗ Failed to import agent.git_operations: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def test_handoff_logic():
|
|
"""Test handoff decision logic."""
|
|
print("\nTesting handoff logic...")
|
|
|
|
from agent.handoff_logic import should_handoff, estimate_task_complexity
|
|
|
|
# Test low complexity task
|
|
subtask = {
|
|
"description": "Fix typo in comment",
|
|
"complexity": "low",
|
|
"estimated_time": 5,
|
|
"required_capabilities": ["general"]
|
|
}
|
|
decision = should_handoff(subtask, "agent-test")
|
|
assert not decision.should_handoff, "Low complexity task should not handoff"
|
|
print("✓ Low complexity task correctly handled")
|
|
|
|
# Test high complexity task
|
|
subtask = {
|
|
"description": "Refactor entire authentication system",
|
|
"complexity": "high",
|
|
"estimated_time": 120,
|
|
"required_capabilities": ["security", "architecture"]
|
|
}
|
|
decision = should_handoff(subtask, "agent-test")
|
|
assert decision.should_handoff, "High complexity task should handoff"
|
|
print("✓ High complexity task correctly triggers handoff")
|
|
|
|
# Test specialized capabilities
|
|
subtask = {
|
|
"description": "Optimize database queries",
|
|
"complexity": "medium",
|
|
"estimated_time": 30,
|
|
"required_capabilities": ["database", "performance"]
|
|
}
|
|
decision = should_handoff(subtask, "agent-test")
|
|
assert decision.should_handoff, "Specialized capabilities should trigger handoff"
|
|
print("✓ Specialized capabilities correctly trigger handoff")
|
|
|
|
# Test complexity estimation
|
|
assert estimate_task_complexity("Fix typo in README") == "low"
|
|
assert estimate_task_complexity("Refactor authentication system") == "high"
|
|
assert estimate_task_complexity("Add new API endpoint") == "medium"
|
|
print("✓ Complexity estimation working correctly")
|
|
|
|
return True
|
|
|
|
|
|
def test_agent_initialization():
|
|
"""Test agent can be initialized."""
|
|
print("\nTesting agent initialization...")
|
|
|
|
from agent.main import Agent
|
|
|
|
# Set dummy API key for testing
|
|
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-test-key-for-initialization-test"
|
|
|
|
try:
|
|
agent = Agent(
|
|
orchestrator_url="ws://localhost:8000",
|
|
agent_id="test-agent",
|
|
capabilities=["python", "testing"],
|
|
workspace_dir="/tmp/test-workspace"
|
|
)
|
|
print(f"✓ Agent initialized with ID: {agent.agent_id}")
|
|
print(f"✓ Agent capabilities: {agent.capabilities}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"✗ Failed to initialize agent: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Run all tests."""
|
|
print("=" * 60)
|
|
print("Agent Component Tests")
|
|
print("=" * 60)
|
|
|
|
tests = [
|
|
("Import Tests", test_imports),
|
|
("Handoff Logic Tests", test_handoff_logic),
|
|
("Agent Initialization Tests", test_agent_initialization),
|
|
]
|
|
|
|
results = []
|
|
for name, test_func in tests:
|
|
try:
|
|
result = test_func()
|
|
results.append((name, result))
|
|
except Exception as e:
|
|
print(f"\n✗ {name} failed with exception: {e}")
|
|
results.append((name, False))
|
|
|
|
print("\n" + "=" * 60)
|
|
print("Test Results")
|
|
print("=" * 60)
|
|
|
|
for name, result in results:
|
|
status = "✓ PASS" if result else "✗ FAIL"
|
|
print(f"{status}: {name}")
|
|
|
|
all_passed = all(result for _, result in results)
|
|
|
|
if all_passed:
|
|
print("\n✓ All tests passed!")
|
|
return 0
|
|
else:
|
|
print("\n✗ Some tests failed")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|