File size: 6,089 Bytes
3de42e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | #!/usr/bin/env python3
"""Async audit for Stack 2.9 tools - tests async tools properly."""
import asyncio
import sys
import time
sys.path.insert(0, '/Users/walidsobhi/stack-2.9/src')
from tools import tool_registry
def get_test_input(tool_name: str) -> dict:
"""Generate appropriate test input for each tool."""
test_inputs = {
"file_read": {"path": "/Users/walidsobhi/stack-2.9/README.md"},
"file_write": {"path": "/tmp/test_audit.txt", "content": "test content"},
"file_exists": {"path": "/Users/walidsobhi/stack-2.9/README.md"},
"file_delete": {"path": "/tmp/test_audit.txt"},
"file_edit": {"operation": "insert", "path": "/tmp/test_edit.txt", "content": "test"},
"file_edit_insert": {"path": "/tmp/test_insert.txt", "content": "test", "offset": 0},
"file_edit_delete": {"path": "/tmp/test_delete.txt", "start": 0, "end": 5},
"file_edit_replace": {"path": "/tmp/test_replace.txt", "pattern": "old", "replacement": "new"},
"glob": {"pattern": "*.py"},
"glob_list": {"pattern": "*.md"},
"grep": {"pattern": "def ", "path": "/Users/walidsobhi/stack-2.9"},
"grep_count": {"pattern": "import", "path": "/Users/walidsobhi/stack-2.9/src"},
"web_fetch": {"url": "https://example.com"},
"web_fetch_meta": {"url": "https://example.com"},
"WebSearch": {"query": "python async", "num_results": 3},
"ask_question": {"question": "What is 2+2?"},
"get_pending_questions": {},
"answer_question": {"question_id": "1", "answer": "4"},
"brief": {"content": "This is a test content for brief summarization."},
"brief_summary": {"content": "Test content here."},
"agent_spawn": {"agent_type": "general-purpose", "task": "test task"},
"agent_status": {"agent_id": "test-123"},
"agent_list": {},
"sleep": {"seconds": 0.01},
"wait_for": {"condition": "true", "timeout": 1},
"skill_list": {},
"skill_execute": {"skill_name": "test", "args": {}},
"skill_info": {"skill_name": "test"},
"skill_chain": {"skills": [], "initial_input": {}},
"skill_search": {"query": "test"},
"TaskCreate": {"subject": "Test Task", "description": "Test description"},
"TaskList": {},
"TaskUpdate": {"taskId": "1", "status": "completed"},
"TaskDelete": {"taskId": "1"},
"TodoWrite": {"subject": "Test", "content": "Test content"},
"team_create": {"name": "test-team"},
"team_delete": {"team_id": "1"},
"team_disband": {"team_id": "1"},
"team_list": {},
"team_status": {"team_id": "1"},
"team_assign": {"team_id": "1", "agent_id": "1"},
"team_leave": {"team_id": "1"},
"Config": {"operation": "get", "key": "test"},
"CronCreate": {"expression": "* * * * *", "command": "echo test"},
"CronList": {},
"CronDelete": {"id": "1"},
"EnterPlanMode": {},
"ExitPlanMode": {},
"message_send": {"channel": "test", "content": "test"},
"message_list": {"channel": "test"},
"message_channel": {"name": "test"},
"message_template": {"template": "test", "vars": {}},
"remote_add": {"name": "test", "url": "https://example.com"},
"remote_list": {},
"remote_remove": {"name": "test"},
"remote_trigger": {"action": "test", "params": {}},
"mcp_list_servers": {},
"mcp_add_server": {"name": "test", "command": "test"},
"mcp_call": {"server": "test", "tool_name": "test", "args": {}},
"read_mcp_resource": {"resource_uri": "test://resource"},
"tool_search": {"query": "file"},
"tool_list_all": {},
"tool_info": {"name": "file_read"},
"tool_capabilities": {},
"synthetic_output": {"content": "test"},
"structure_data": {"content": "test", "format": "json"},
"enter_worktree": {"path": "/tmp/test"},
"exit_worktree": {},
"list_worktrees": {},
}
return test_inputs.get(tool_name, {})
async def test_tool(tool_name: str) -> dict:
"""Test a single async tool."""
tool = tool_registry.get(tool_name)
if not tool:
return {"tool": tool_name, "status": "FAIL", "error": "Tool not found"}
test_input = get_test_input(tool_name)
start_time = time.time()
try:
result = await tool.execute(**test_input)
duration = time.time() - start_time
return {
"tool": tool_name,
"status": "PASS",
"duration": duration,
"result": str(result)[:100] if result else "None"
}
except Exception as e:
duration = time.time() - start_time
return {
"tool": tool_name,
"status": "FAIL",
"duration": duration,
"error": str(e)[:100]
}
async def main():
"""Run async audit on all tools."""
print("=" * 60)
print("STACK 2.9 ASYNC TOOLS AUDIT")
print("=" * 60)
tools = tool_registry.list()
print(f"\nFound {len(tools)} registered tools\n")
results = []
passed = 0
failed = 0
for tool_name in tools:
result = await test_tool(tool_name)
results.append(result)
status = result["status"]
if status == "PASS":
passed += 1
print(f"✅ {tool_name}: PASS ({result['duration']:.3f}s)")
else:
failed += 1
error = result.get('error', 'Unknown error')
print(f"❌ {tool_name}: FAIL - {error[:50]}")
print("\n" + "=" * 60)
print("RESULTS SUMMARY")
print("=" * 60)
print(f"Total Tools: {len(tools)}")
print(f"Passed: {passed} ({passed*100//len(tools)}%)")
print(f"Failed: {failed} ({failed*100//len(tools)}%)")
if failed > 0:
print("\nFailed tools:")
for r in results:
if r["status"] == "FAIL":
print(f" - {r['tool']}: {r.get('error', 'Unknown')[:60]}")
return results
if __name__ == "__main__":
asyncio.run(main()) |