mirror of
https://github.com/ultraworkers/claw-code.git
synced 2026-04-24 21:28:11 +08:00
6a76cc7c08
Closes the last #160 gap: claws can now manage session lifecycle entirely through the CLI without filesystem hacks. New commands: - claw list-sessions [--directory DIR] [--output-format text|json] Enumerates stored session IDs. JSON mode emits {sessions, count}. Missing/empty directories return empty list (exit 0), not an error. - claw delete-session SESSION_ID [--directory DIR] [--output-format text|json] Idempotent: not-found is exit 0 with status='not_found' (no raise). Partial-failure: exit 1 with typed JSON error envelope: {session_id, deleted: false, error: {kind, message, retryable}} The 'session_delete_failed' kind is retryable=true so orchestrators know to retry vs escalate. Public API surface extended in src/__init__.py: - list_sessions, session_exists, delete_session - SessionNotFoundError, SessionDeleteError Tests added (tests/test_porting_workspace.py): - test_list_sessions_cli_runs: text + json modes against tempdir - test_delete_session_cli_idempotent: first call deleted=true, second call deleted=false (exit 0, status=not_found) - test_delete_session_cli_partial_failure_exit_1: permission error surfaces as exit 1 + typed JSON error with retryable=true All 43 tests pass. The session storage abstraction chapter is closed: - storage layer decoupled from claw code (#160 initial impl) - delete contract hardened + caller-audited (#160 hardening pass) - CLI wired with idempotency preserved at exit-code boundary (this commit)
336 lines
14 KiB
Python
336 lines
14 KiB
Python
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
|
|
from src.port_manifest import build_port_manifest
|
|
from src.query_engine import QueryEnginePort
|
|
from src.tools import PORTED_TOOLS
|
|
|
|
|
|
class PortingWorkspaceTests(unittest.TestCase):
|
|
def test_manifest_counts_python_files(self) -> None:
|
|
manifest = build_port_manifest()
|
|
self.assertGreaterEqual(manifest.total_python_files, 20)
|
|
self.assertTrue(manifest.top_level_modules)
|
|
|
|
def test_query_engine_summary_mentions_workspace(self) -> None:
|
|
summary = QueryEnginePort.from_workspace().render_summary()
|
|
self.assertIn('Python Porting Workspace Summary', summary)
|
|
self.assertIn('Command surface:', summary)
|
|
self.assertIn('Tool surface:', summary)
|
|
|
|
def test_cli_summary_runs(self) -> None:
|
|
result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'summary'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertIn('Python Porting Workspace Summary', result.stdout)
|
|
|
|
def test_parity_audit_runs(self) -> None:
|
|
result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'parity-audit'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertIn('Parity Audit', result.stdout)
|
|
|
|
def test_root_file_coverage_is_complete_when_local_archive_exists(self) -> None:
|
|
audit = run_parity_audit()
|
|
if audit.archive_present:
|
|
self.assertEqual(audit.root_file_coverage[0], audit.root_file_coverage[1])
|
|
self.assertGreaterEqual(audit.directory_coverage[0], 28)
|
|
self.assertGreaterEqual(audit.command_entry_ratio[0], 150)
|
|
self.assertGreaterEqual(audit.tool_entry_ratio[0], 100)
|
|
|
|
def test_command_and_tool_snapshots_are_nontrivial(self) -> None:
|
|
self.assertGreaterEqual(len(PORTED_COMMANDS), 150)
|
|
self.assertGreaterEqual(len(PORTED_TOOLS), 100)
|
|
|
|
def test_commands_and_tools_cli_run(self) -> None:
|
|
commands_result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'commands', '--limit', '5', '--query', 'review'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
tools_result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'tools', '--limit', '5', '--query', 'MCP'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertIn('Command entries:', commands_result.stdout)
|
|
self.assertIn('Tool entries:', tools_result.stdout)
|
|
|
|
def test_subsystem_packages_expose_archive_metadata(self) -> None:
|
|
from src import assistant, bridge, utils
|
|
|
|
self.assertGreater(assistant.MODULE_COUNT, 0)
|
|
self.assertGreater(bridge.MODULE_COUNT, 0)
|
|
self.assertGreater(utils.MODULE_COUNT, 100)
|
|
self.assertTrue(utils.SAMPLE_FILES)
|
|
|
|
def test_route_and_show_entry_cli_run(self) -> None:
|
|
route_result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'route', 'review MCP tool', '--limit', '5'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
show_command = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'show-command', 'review'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
show_tool = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'show-tool', 'MCPTool'],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.assertIn('review', route_result.stdout.lower())
|
|
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_list_sessions_cli_runs(self) -> None:
|
|
"""#160: list-sessions CLI enumerates stored sessions in text + json."""
|
|
import json
|
|
import tempfile
|
|
from src.session_store import StoredSession, save_session
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
for sid in ['alpha', 'bravo']:
|
|
save_session(
|
|
StoredSession(session_id=sid, messages=('hi',), input_tokens=1, output_tokens=2),
|
|
tmp_path,
|
|
)
|
|
# text mode
|
|
text_result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'list-sessions', '--directory', str(tmp_path)],
|
|
check=True, capture_output=True, text=True,
|
|
)
|
|
self.assertIn('alpha', text_result.stdout)
|
|
self.assertIn('bravo', text_result.stdout)
|
|
# json mode
|
|
json_result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'list-sessions',
|
|
'--directory', str(tmp_path), '--output-format', 'json'],
|
|
check=True, capture_output=True, text=True,
|
|
)
|
|
data = json.loads(json_result.stdout)
|
|
self.assertEqual(data, {'sessions': ['alpha', 'bravo'], 'count': 2})
|
|
|
|
def test_delete_session_cli_idempotent(self) -> None:
|
|
"""#160: delete-session CLI is idempotent (not-found is exit 0, status=not_found)."""
|
|
import json
|
|
import tempfile
|
|
from src.session_store import StoredSession, save_session
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
save_session(
|
|
StoredSession(session_id='once', messages=('hi',), input_tokens=1, output_tokens=2),
|
|
tmp_path,
|
|
)
|
|
# first delete: success
|
|
first = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'delete-session', 'once',
|
|
'--directory', str(tmp_path), '--output-format', 'json'],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(first.returncode, 0)
|
|
self.assertEqual(
|
|
json.loads(first.stdout),
|
|
{'session_id': 'once', 'deleted': True, 'status': 'deleted'},
|
|
)
|
|
# second delete: idempotent, still exit 0
|
|
second = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'delete-session', 'once',
|
|
'--directory', str(tmp_path), '--output-format', 'json'],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(second.returncode, 0)
|
|
self.assertEqual(
|
|
json.loads(second.stdout),
|
|
{'session_id': 'once', 'deleted': False, 'status': 'not_found'},
|
|
)
|
|
|
|
def test_delete_session_cli_partial_failure_exit_1(self) -> None:
|
|
"""#160: partial-failure (permission error) surfaces as exit 1 + typed JSON error."""
|
|
import json
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
bad = tmp_path / 'locked.json'
|
|
bad.mkdir()
|
|
try:
|
|
result = subprocess.run(
|
|
[sys.executable, '-m', 'src.main', 'delete-session', 'locked',
|
|
'--directory', str(tmp_path), '--output-format', 'json'],
|
|
capture_output=True, text=True,
|
|
)
|
|
self.assertEqual(result.returncode, 1)
|
|
data = json.loads(result.stdout)
|
|
self.assertFalse(data['deleted'])
|
|
self.assertEqual(data['error']['kind'], 'session_delete_failed')
|
|
self.assertTrue(data['error']['retryable'])
|
|
finally:
|
|
bad.rmdir()
|
|
|
|
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()
|