Rewriting Project Claw Code - Python port with Rust on the way

This commit is contained in:
instructkr
2026-03-31 08:03:46 -07:00
parent 507c2460b9
commit 01bf54ad15
31 changed files with 1207 additions and 111 deletions

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import subprocess
import sys
import unittest
from pathlib import Path
from src.commands import PORTED_COMMANDS
from src.parity_audit import run_parity_audit
@@ -100,6 +101,148 @@ class PortingWorkspaceTests(unittest.TestCase):
self.assertIn('review', show_command.stdout.lower())
self.assertIn('mcptool', show_tool.stdout.lower())
def test_bootstrap_cli_runs(self) -> None:
result = subprocess.run(
[sys.executable, '-m', 'src.main', 'bootstrap', 'review MCP tool', '--limit', '5'],
check=True,
capture_output=True,
text=True,
)
self.assertIn('Runtime Session', result.stdout)
self.assertIn('Startup Steps', result.stdout)
self.assertIn('Routed Matches', result.stdout)
def test_bootstrap_session_tracks_turn_state(self) -> None:
from src.runtime import PortRuntime
session = PortRuntime().bootstrap_session('review MCP tool', limit=5)
self.assertGreaterEqual(len(session.turn_result.matched_tools), 1)
self.assertIn('Prompt:', session.turn_result.output)
self.assertGreaterEqual(session.turn_result.usage.input_tokens, 1)
def test_exec_command_and_tool_cli_run(self) -> None:
command_result = subprocess.run(
[sys.executable, '-m', 'src.main', 'exec-command', 'review', 'inspect security review'],
check=True,
capture_output=True,
text=True,
)
tool_result = subprocess.run(
[sys.executable, '-m', 'src.main', 'exec-tool', 'MCPTool', 'fetch resource list'],
check=True,
capture_output=True,
text=True,
)
self.assertIn("Mirrored command 'review'", command_result.stdout)
self.assertIn("Mirrored tool 'MCPTool'", tool_result.stdout)
def test_setup_report_and_registry_filters_run(self) -> None:
setup_result = subprocess.run(
[sys.executable, '-m', 'src.main', 'setup-report'],
check=True,
capture_output=True,
text=True,
)
command_result = subprocess.run(
[sys.executable, '-m', 'src.main', 'commands', '--limit', '5', '--no-plugin-commands'],
check=True,
capture_output=True,
text=True,
)
tool_result = subprocess.run(
[sys.executable, '-m', 'src.main', 'tools', '--limit', '5', '--simple-mode', '--no-mcp'],
check=True,
capture_output=True,
text=True,
)
self.assertIn('Setup Report', setup_result.stdout)
self.assertIn('Command entries:', command_result.stdout)
self.assertIn('Tool entries:', tool_result.stdout)
def test_load_session_cli_runs(self) -> None:
from src.runtime import PortRuntime
session = PortRuntime().bootstrap_session('review MCP tool', limit=5)
session_id = Path(session.persisted_session_path).stem
result = subprocess.run(
[sys.executable, '-m', 'src.main', 'load-session', session_id],
check=True,
capture_output=True,
text=True,
)
self.assertIn(session_id, result.stdout)
self.assertIn('messages', result.stdout)
def test_tool_permission_filtering_cli_runs(self) -> None:
result = subprocess.run(
[sys.executable, '-m', 'src.main', 'tools', '--limit', '10', '--deny-prefix', 'mcp'],
check=True,
capture_output=True,
text=True,
)
self.assertIn('Tool entries:', result.stdout)
self.assertNotIn('MCPTool', result.stdout)
def test_turn_loop_cli_runs(self) -> None:
result = subprocess.run(
[sys.executable, '-m', 'src.main', 'turn-loop', 'review MCP tool', '--max-turns', '2', '--structured-output'],
check=True,
capture_output=True,
text=True,
)
self.assertIn('## Turn 1', result.stdout)
self.assertIn('stop_reason=', result.stdout)
def test_remote_mode_clis_run(self) -> None:
remote_result = subprocess.run([sys.executable, '-m', 'src.main', 'remote-mode', 'workspace'], check=True, capture_output=True, text=True)
ssh_result = subprocess.run([sys.executable, '-m', 'src.main', 'ssh-mode', 'workspace'], check=True, capture_output=True, text=True)
teleport_result = subprocess.run([sys.executable, '-m', 'src.main', 'teleport-mode', 'workspace'], check=True, capture_output=True, text=True)
self.assertIn('mode=remote', remote_result.stdout)
self.assertIn('mode=ssh', ssh_result.stdout)
self.assertIn('mode=teleport', teleport_result.stdout)
def test_flush_transcript_cli_runs(self) -> None:
result = subprocess.run(
[sys.executable, '-m', 'src.main', 'flush-transcript', 'review MCP tool'],
check=True,
capture_output=True,
text=True,
)
self.assertIn('flushed=True', result.stdout)
def test_command_graph_and_tool_pool_cli_run(self) -> None:
command_graph = subprocess.run([sys.executable, '-m', 'src.main', 'command-graph'], check=True, capture_output=True, text=True)
tool_pool = subprocess.run([sys.executable, '-m', 'src.main', 'tool-pool'], check=True, capture_output=True, text=True)
self.assertIn('Command Graph', command_graph.stdout)
self.assertIn('Tool Pool', tool_pool.stdout)
def test_setup_report_mentions_deferred_init(self) -> None:
result = subprocess.run(
[sys.executable, '-m', 'src.main', 'setup-report'],
check=True,
capture_output=True,
text=True,
)
self.assertIn('Deferred init:', result.stdout)
self.assertIn('plugin_init=True', result.stdout)
def test_execution_registry_runs(self) -> None:
from src.execution_registry import build_execution_registry
registry = build_execution_registry()
self.assertGreaterEqual(len(registry.commands), 150)
self.assertGreaterEqual(len(registry.tools), 100)
self.assertIn('Mirrored command', registry.command('review').execute('review security'))
self.assertIn('Mirrored tool', registry.tool('MCPTool').execute('fetch mcp resources'))
def test_bootstrap_graph_and_direct_modes_run(self) -> None:
graph_result = subprocess.run([sys.executable, '-m', 'src.main', 'bootstrap-graph'], check=True, capture_output=True, text=True)
direct_result = subprocess.run([sys.executable, '-m', 'src.main', 'direct-connect-mode', 'workspace'], check=True, capture_output=True, text=True)
deep_link_result = subprocess.run([sys.executable, '-m', 'src.main', 'deep-link-mode', 'workspace'], check=True, capture_output=True, text=True)
self.assertIn('Bootstrap Graph', graph_result.stdout)
self.assertIn('mode=direct-connect', direct_result.stdout)
self.assertIn('mode=deep-link', deep_link_result.stdout)
if __name__ == '__main__':
unittest.main()