mirror of
https://github.com/ultraworkers/claw-code.git
synced 2026-04-24 21:20:48 +08:00
290ab7e41f
Completes the coverage → enforcement → documentation → alignment cycle.
Every clawable command now emits the canonical JSON envelope per SCHEMAS.md:
Common fields (now real in output):
- timestamp (ISO 8601 UTC)
- command (argv[1])
- exit_code (0/1/2)
- output_format ('json')
- schema_version ('1.0')
13 commands wrapped:
- list-sessions, delete-session, load-session, flush-transcript
- show-command, show-tool
- exec-command, exec-tool, route, bootstrap
- command-graph, tool-pool, bootstrap-graph
Implementation:
- Added wrap_json_envelope() helper in src/main.py
- Wrapped all 18 JSON output paths (13 success + 5 error paths)
- Applied exit_code=1 to error/not-found envelopes
- Kept text mode byte-identical (backward compat preserved)
Test updates:
- 3 skipped common-field tests now pass automatically
- 3 existing tests updated to verify common envelope fields while preserving command-specific field checks
- test_list_sessions_cli_runs, test_delete_session_cli_idempotent,
test_load_session_cli::test_json_mode_on_success
Full suite: 179 → 182 passing (+3 activated from skipped), zero regression.
Loop completion:
Coverage (#167-#170) ✅ All 13 commands accept --output-format
Enforcement (#171) ✅ CI blocks new commands without --output-format
Documentation (#172) ✅ SCHEMAS.md defines envelope contract
Alignment (#173 this) ✅ Actual output matches SCHEMAS.md contract
Example output now:
$ claw list-sessions --output-format json
{
"timestamp": "2026-04-22T10:34:12Z",
"command": "list-sessions",
"exit_code": 0,
"output_format": "json",
"schema_version": "1.0",
"sessions": ["alpha", "bravo"],
"count": 2
}
Closes ROADMAP #173. Protocol is now documented AND real.
Claws can build ONE error handler, ONE timestamp parser, ONE version check
instead of 13 special cases.
348 lines
15 KiB
Python
348 lines
15 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)
|
|
# Verify common envelope fields (SCHEMAS.md contract)
|
|
self.assertIn('timestamp', data)
|
|
self.assertEqual(data['command'], 'list-sessions')
|
|
self.assertEqual(data['schema_version'], '1.0')
|
|
# Verify command-specific fields
|
|
self.assertEqual(data['sessions'], ['alpha', 'bravo'])
|
|
self.assertEqual(data['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)
|
|
envelope_first = json.loads(first.stdout)
|
|
# Verify common envelope fields (SCHEMAS.md contract)
|
|
self.assertIn('timestamp', envelope_first)
|
|
self.assertEqual(envelope_first['command'], 'delete-session')
|
|
self.assertEqual(envelope_first['exit_code'], 0)
|
|
self.assertEqual(envelope_first['schema_version'], '1.0')
|
|
# Verify command-specific fields
|
|
self.assertEqual(envelope_first['session_id'], 'once')
|
|
self.assertEqual(envelope_first['deleted'], True)
|
|
self.assertEqual(envelope_first['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)
|
|
envelope_second = json.loads(second.stdout)
|
|
self.assertEqual(envelope_second['session_id'], 'once')
|
|
self.assertEqual(envelope_second['deleted'], False)
|
|
self.assertEqual(envelope_second['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()
|