mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
zeekay
7374804405
The default surface was 34 tools, 61 KB, ~15,300 tokens of schema in every
prompt of every agent. The mode system already names what a session should
carry, but only the tools register_all_tools enumerates obeyed it: zsh,
curl, playwright and the helpers (ps, npx, uvx, jq, wget, cdp, open) stayed
on in every mode. A mode now decides every discovered tool, the `hanzo`
mode lists what a coding agent calls every few steps (18 tools, 32 KB,
~8,000 tokens), and the rest is a name away: HANZO_MCP_ENABLED_TOOLS=a,b
or =all, `tool enable`, or `mode`.
hanzo-tools-shell 0.6.6: run_shell kept every byte a foreground command
wrote until the command ended. `yes` for thirty seconds is gigabytes. It
keeps the first 16 MiB per stream and drains the rest.
The tests' session-wide patch of asyncio.create_subprocess_exec handed
every shell command a stream that never reached EOF; the first test to run
one read it in a loop until the machine was out of memory (22 GB, killed).
The patch is now opt-in and its process has exited.
bench/latency.py measures the cold start, the tools/list bill and the common
calls through a real MCP client over stdio. Interleaved rounds, wall and CPU
seconds, p50 over 6 cold starts:
cold wall cold CPU tools bytes ~tokens
installed 0.15.18 9.34 s 10.46 s 30 52,601 13,150
main 0.15.20 1.52 s 1.13 s 34 61,287 15,321
this 0.96 s 1.06 s 18 32,222 8,055
The installed environment's 9 s is its hanzoai 8.5.156, whose package init
imports the whole generated cloud client; hanzo-mcp >= 0.15.20 requires
hanzoai >= 8.5.624, which loads it lazily. Per-call latency is spawn-bound
(zsh 17.6 ms, git status 33 ms, fs.read 2.5 ms at p50) and moves with the
machine's load; no call path changed.
205 lines
6.3 KiB
Python
205 lines
6.3 KiB
Python
"""Test LSP tool functionality."""
|
|
|
|
import asyncio
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from hanzo_tools.lsp import create_lsp_tool
|
|
|
|
pytestmark = pytest.mark.usefixtures("mock_slow_operations")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_status():
|
|
"""Test LSP tool status check with mocking."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
# Create a test Go file
|
|
with tempfile.NamedTemporaryFile(suffix=".go", mode="w", delete=False) as f:
|
|
f.write("""package main
|
|
|
|
import "fmt"
|
|
|
|
func main() {
|
|
fmt.Println("Hello, World!")
|
|
}
|
|
|
|
func greet(name string) string {
|
|
return fmt.Sprintf("Hello, %s!", name)
|
|
}
|
|
""")
|
|
go_file = f.name
|
|
|
|
try:
|
|
# Mock the subprocess calls to prevent hanging
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Mock successful check for gopls
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
# Check status for Go
|
|
result = await lsp_tool.run(action="status", file=go_file)
|
|
|
|
assert result.data is not None
|
|
assert "language" in result.data
|
|
assert result.data["language"] == "go"
|
|
assert "lsp_server" in result.data
|
|
assert result.data["lsp_server"] == "gopls"
|
|
assert "capabilities" in result.data
|
|
assert "definition" in result.data["capabilities"]
|
|
|
|
finally:
|
|
Path(go_file).unlink()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_unsupported_file():
|
|
"""Test LSP tool with unsupported file type."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
# Test with unsupported file
|
|
result = await lsp_tool.run(action="status", file="test.unknown")
|
|
|
|
assert result.data is not None
|
|
assert "error" in result.data
|
|
assert "Unsupported file type" in result.data["error"]
|
|
assert "supported_languages" in result.data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_invalid_action():
|
|
"""Test LSP tool with invalid action."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
# Test with invalid action
|
|
result = await lsp_tool.run(action="invalid_action", file="test.py")
|
|
|
|
assert result.data is not None
|
|
assert "error" in result.data
|
|
assert "Invalid action" in result.data["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_definition_placeholder():
|
|
"""Test LSP tool definition action (placeholder) with mocking."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
with tempfile.NamedTemporaryFile(suffix=".py", mode="w", delete=False) as f:
|
|
f.write("""def hello():
|
|
return "Hello"
|
|
|
|
result = hello()
|
|
""")
|
|
py_file = f.name
|
|
|
|
try:
|
|
# Mock the subprocess calls
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Mock successful check for pylsp
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
# Test definition lookup
|
|
result = await lsp_tool.run(
|
|
action="definition",
|
|
file=py_file,
|
|
line=4,
|
|
character=9, # Position of 'hello' in 'hello()'
|
|
)
|
|
|
|
assert result.data is not None
|
|
assert "action" in result.data
|
|
assert result.data["action"] == "definition"
|
|
assert "note" in result.data
|
|
assert "fallback" in result.data
|
|
|
|
finally:
|
|
Path(py_file).unlink()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_python_status():
|
|
"""Test LSP status for Python files with mocking."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
# Mock the subprocess calls
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Mock successful check for pylsp
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
# Check Python LSP status
|
|
result = await lsp_tool.run(action="status", file="test.py")
|
|
|
|
assert result.data is not None
|
|
assert result.data["language"] == "python"
|
|
assert result.data["lsp_server"] == "pylsp"
|
|
assert "definition" in result.data["capabilities"]
|
|
assert "hover" in result.data["capabilities"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_typescript_status():
|
|
"""Test LSP status for TypeScript files with mocking."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
# Mock the subprocess calls
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Mock successful check for typescript-language-server
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
# Check TypeScript LSP status
|
|
result = await lsp_tool.run(action="status", file="app.ts")
|
|
|
|
assert result.data is not None
|
|
assert result.data["language"] == "typescript"
|
|
assert result.data["lsp_server"] == "typescript-language-server"
|
|
assert "completion" in result.data["capabilities"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lsp_tool_rename_placeholder():
|
|
"""Test LSP rename action (placeholder) with mocking."""
|
|
lsp_tool = create_lsp_tool()
|
|
|
|
# Mock the subprocess calls
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Mock successful check
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
# Test rename
|
|
result = await lsp_tool.run(
|
|
action="rename",
|
|
file="test.go",
|
|
line=10,
|
|
character=5,
|
|
new_name="newFunctionName",
|
|
)
|
|
|
|
assert result.data is not None
|
|
assert result.data["action"] == "rename"
|
|
assert result.data["new_name"] == "newFunctionName"
|
|
assert "fallback" in result.data
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Run basic tests
|
|
asyncio.run(test_lsp_tool_status())
|
|
asyncio.run(test_lsp_tool_unsupported_file())
|
|
asyncio.run(test_lsp_tool_invalid_action())
|
|
print("LSP tool tests passed!")
|