Jack1808/Claude_Code
0
1"""Tests for cli/ module."""2 3import asyncio4import json5import os6from unittest.mock import AsyncMock, MagicMock, patch7 8import pytest9 10from messaging.event_parser import parse_cli_event11 12# --- Existing Parser Tests ---13 14 15class TestCLIParser:16 """Test CLI event parsing."""17 18 def test_parse_text_content(self):19 """Test parsing text content from assistant message."""20 event = {21 "type": "assistant",22 "message": {"content": [{"type": "text", "text": "Hello, world!"}]},23 }24 result = parse_cli_event(event)25 assert len(result) == 126 assert result[0]["type"] == "text_chunk"27 assert result[0]["text"] == "Hello, world!"28 29 def test_parse_thinking_content(self):30 """Test parsing thinking content."""31 event = {32 "type": "assistant",33 "message": {34 "content": [{"type": "thinking", "thinking": "Let me think..."}]35 },36 }37 result = parse_cli_event(event)38 assert len(result) == 139 assert result[0]["type"] == "thinking_chunk"40 assert (41 result[0]["text"] == "Let me think...\n"42 or result[0]["text"] == "Let me think..."43 )44 45 def test_parse_multiple_content(self):46 """Test parsing mixed content (thinking + tools)."""47 event = {48 "type": "assistant",49 "message": {50 "content": [51 {"type": "thinking", "thinking": "Thinking..."},52 {"type": "tool_use", "name": "ls", "input": {}},53 ]54 },55 }56 result = parse_cli_event(event)57 assert len(result) == 258 assert result[0]["type"] == "thinking_chunk"59 assert result[0]["text"] == "Thinking..."60 assert result[1]["type"] == "tool_use"61 62 def test_parse_tool_use(self):63 """Test parsing tool use content."""64 event = {65 "type": "assistant",66 "message": {67 "content": [68 {69 "type": "tool_use",70 "name": "read_file",71 "input": {"path": "/test"},72 }73 ]74 },75 }76 result = parse_cli_event(event)77 assert len(result) == 178 assert result[0]["type"] == "tool_use"79 assert result[0]["name"] == "read_file"80 assert result[0]["input"] == {"path": "/test"}81 82 def test_parse_text_delta(self):83 """Test parsing streaming text delta."""84 event = {85 "type": "content_block_delta",86 "index": 0,87 "delta": {"type": "text_delta", "text": "streaming text"},88 }89 result = parse_cli_event(event)90 assert len(result) == 191 assert result[0]["type"] == "text_delta"92 assert result[0]["text"] == "streaming text"93 94 def test_parse_thinking_delta(self):95 """Test parsing streaming thinking delta."""96 event = {97 "type": "content_block_delta",98 "index": 1,99 "delta": {"type": "thinking_delta", "thinking": "thinking..."},100 }101 result = parse_cli_event(event)102 assert len(result) == 1103 assert result[0]["type"] == "thinking_delta"104 assert result[0]["text"] == "thinking..."105 106 def test_parse_error(self):107 """Test parsing error event."""108 event = {"type": "error", "error": {"message": "Something went wrong"}}109 result = parse_cli_event(event)110 assert result[0]["type"] == "error"111 assert result[0]["message"] == "Something went wrong"112 113 def test_parse_exit_success(self):114 """Test parsing exit event with success."""115 event = {"type": "exit", "code": 0}116 result = parse_cli_event(event)117 assert result[0]["type"] == "complete"118 assert result[0]["status"] == "success"119 120 def test_parse_exit_failure(self):121 """Test parsing exit event with failure returns error then complete."""122 event = {"type": "exit", "code": 1}123 result = parse_cli_event(event)124 # Non-zero exit now returns error first, then complete125 assert len(result) == 2126 assert result[0]["type"] == "error"127 assert (128 "exit" in result[0]["message"].lower()129 or "code" in result[0]["message"].lower()130 )131 assert result[1]["type"] == "complete"132 assert result[1]["status"] == "failed"133 134 def test_parse_invalid_event(self):135 """Test parsing returns empty list for unrecognized event."""136 result = parse_cli_event({"type": "unknown"})137 assert result == []138 139 def test_parse_non_dict(self):140 """Test parsing returns empty list for non-dict input."""141 result = parse_cli_event("not a dict")142 assert result == []143 144 145# --- CLI Session Tests ---146 147 148class TestCLISession:149 """Test CLISession."""150 151 def test_session_init(self):152 """Test CLISession initialization."""153 from cli.session import CLISession154 155 session = CLISession(156 workspace_path="/tmp/test",157 api_url="http://localhost:8082/v1",158 allowed_dirs=["/home/user/projects"],159 )160 assert session.workspace == os.path.normpath(os.path.abspath("/tmp/test"))161 assert session.api_url == "http://localhost:8082/v1"162 assert not session.is_busy163 164 def test_session_extract_session_id(self):165 """Test session ID extraction from various event formats."""166 from cli.session import CLISession167 168 session = CLISession("/tmp", "http://localhost:8082/v1")169 170 # Direct session_id field171 assert session._extract_session_id({"session_id": "abc123"}) == "abc123"172 assert session._extract_session_id({"sessionId": "abc123"}) == "abc123"173 174 # Nested in init175 assert (176 session._extract_session_id({"init": {"session_id": "nested123"}})177 == "nested123"178 )179 180 # Nested in result181 assert (182 session._extract_session_id({"result": {"session_id": "res123"}})183 == "res123"184 )185 186 # Conversation id187 assert (188 session._extract_session_id({"conversation": {"id": "conv123"}})189 == "conv123"190 )191 192 # No session ID193 assert session._extract_session_id({"type": "message"}) is None194 assert session._extract_session_id("not a dict") is None195 196 @pytest.mark.asyncio197 async def test_start_task_basic_flow(self):198 """Test start_task running a basic command flow."""199 from cli.session import CLISession200 201 session = CLISession("/tmp", "http://localhost:8082/v1")202 203 # Mock subprocess204 mock_process = AsyncMock()205 mock_process.stdout.read.side_effect = [206 b'{"type": "message", "content": "Hello"}\n',207 b'{"session_id": "sess_1"}\n',208 b"", # EOF209 ]210 mock_process.stderr.read.return_value = b"" # No error211 mock_process.wait.return_value = 0212 mock_process.returncode = 0213 214 with patch(215 "asyncio.create_subprocess_exec", new_callable=AsyncMock216 ) as mock_exec:217 mock_exec.return_value = mock_process218 219 events = [e async for e in session.start_task("Hello")]220 221 # Verify command construction222 # Arg 1 is subprocess command223 args = mock_exec.call_args[0]224 assert args[0] == "claude"225 assert "-p" in args226 assert "Hello" in args227 228 # Verify events229 assert (230 len(events) == 4231 ) # message, session_id, session_info (synthesized), exit232 assert events[0] == {"type": "message", "content": "Hello"}233 assert events[1] == {"type": "session_info", "session_id": "sess_1"}234 # The session_info event is yielded by _handle_line_gen right after extracting ID235 assert events[2] == {"session_id": "sess_1"} # The original event236 assert events[3] == {"type": "exit", "code": 0, "stderr": None}237 238 assert session.current_session_id == "sess_1"239 240 @pytest.mark.asyncio241 async def test_start_task_with_session_resume(self):242 """Test resuming an existing session."""243 from cli.session import CLISession244 245 session = CLISession("/tmp", "http://localhost:8082/v1")246 247 mock_process = AsyncMock()248 mock_process.stdout.read.side_effect = [249 b"",250 ] # Immediate EOF251 mock_process.stderr.read.return_value = b""252 mock_process.wait.return_value = 0253 254 with patch(255 "asyncio.create_subprocess_exec", new_callable=AsyncMock256 ) as mock_exec:257 mock_exec.return_value = mock_process258 259 async for _ in session.start_task("Hello", session_id="sess_abc"):260 pass261 262 args = mock_exec.call_args[0]263 assert "--resume" in args264 assert "sess_abc" in args265 assert "--fork-session" not in args266 267 @pytest.mark.asyncio268 async def test_start_task_with_session_resume_and_fork(self):269 """Test resuming an existing session and forking."""270 from cli.session import CLISession271 272 session = CLISession("/tmp", "http://localhost:8082/v1")273 274 mock_process = AsyncMock()275 mock_process.stdout.read.side_effect = [b""] # Immediate EOF276 mock_process.stderr.read.return_value = b""277 mock_process.wait.return_value = 0278 279 with patch(280 "asyncio.create_subprocess_exec", new_callable=AsyncMock281 ) as mock_exec:282 mock_exec.return_value = mock_process283 284 async for _ in session.start_task(285 "Hello", session_id="sess_abc", fork_session=True286 ):287 pass288 289 args = mock_exec.call_args[0]290 assert "--resume" in args291 assert "sess_abc" in args292 assert "--fork-session" in args293 294 @pytest.mark.asyncio295 async def test_start_task_process_failure_with_stderr(self):296 """Test process exit with error code and stderr output."""297 from cli.session import CLISession298 299 session = CLISession("/tmp", "http://localhost:8082/v1")300 301 mock_process = AsyncMock()302 mock_process.stdout.read.side_effect = [b""] # No stdout303 mock_process.stderr.read.return_value = b"Fatal error"304 mock_process.wait.return_value = 1305 306 with patch(307 "asyncio.create_subprocess_exec", new_callable=AsyncMock308 ) as mock_exec:309 mock_exec.return_value = mock_process310 311 events = [e async for e in session.start_task("Hello")]312 313 # Should have error event from stderr, then exit event314 assert len(events) == 2315 assert events[0]["type"] == "error"316 assert events[0]["error"]["message"] == "Fatal error"317 318 assert events[1]["type"] == "exit"319 assert events[1]["code"] == 1320 assert events[1]["stderr"] == "Fatal error"321 322 @pytest.mark.asyncio323 async def test_stop_session(self):324 """Test stopping the session process."""325 from cli.session import CLISession326 327 session = CLISession("/tmp", "http://localhost:8082/v1")328 329 mock_process = MagicMock()330 mock_process.returncode = None # Running331 # Mock wait to simulate async finish332 mock_process.wait = AsyncMock(return_value=0)333 334 session.process = mock_process335 336 stopped = await session.stop()337 338 assert stopped is True339 mock_process.terminate.assert_called_once()340 mock_process.wait.assert_called()341 342 @pytest.mark.asyncio343 async def test_stop_session_timeout_force_kill(self):344 """Test force kill if terminate times out."""345 from cli.session import CLISession346 347 session = CLISession("/tmp", "http://localhost:8082/v1")348 349 mock_process = MagicMock()350 mock_process.returncode = None351 352 # First wait times out353 async def wait_side_effect():354 if not mock_process.kill.called:355 await asyncio.sleep(6) # Should be > 5.0 timeout356 return 0357 358 # We can simulate timeout by raising TimeoutError directly on first call359 mock_process.wait = AsyncMock(side_effect=[asyncio.TimeoutError, 0])360 361 session.process = mock_process362 363 stopped = await session.stop()364 365 assert stopped is True366 mock_process.terminate.assert_called()367 mock_process.kill.assert_called()368 369 @pytest.mark.asyncio370 async def test_start_task_split_buffer(self):371 """Test handling of JSON split across chunks."""372 from cli.session import CLISession373 374 session = CLISession("/tmp", "http://localhost:8082/v1")375 376 mock_process = AsyncMock()377 # Split json: {"type": "mess... age"}378 mock_process.stdout.read.side_effect = [379 b'{"type": "mess',380 b'age", "content": "Split"}\n',381 b"",382 ]383 mock_process.stderr.read.return_value = b""384 mock_process.wait.return_value = 0385 386 with patch(387 "asyncio.create_subprocess_exec", new_callable=AsyncMock388 ) as mock_exec:389 mock_exec.return_value = mock_process390 391 events = [392 e async for e in session.start_task("test") if e["type"] == "message"393 ]394 395 assert len(events) == 1396 assert events[0]["content"] == "Split"397 398 @pytest.mark.asyncio399 async def test_start_task_remnant_buffer(self):400 """Test handling of buffer remnant at EOF (no newline at end)."""401 from cli.session import CLISession402 403 session = CLISession("/tmp", "http://localhost:8082/v1")404 405 mock_process = AsyncMock()406 mock_process.stdout.read.side_effect = [407 b'{"type": "message", "content": "Remnant"}', # No newline408 b"",409 ]410 mock_process.stderr.read.return_value = b""411 mock_process.wait.return_value = 0412 413 with patch(414 "asyncio.create_subprocess_exec", new_callable=AsyncMock415 ) as mock_exec:416 mock_exec.return_value = mock_process417 418 events = [419 e async for e in session.start_task("test") if e["type"] == "message"420 ]421 422 assert len(events) == 1423 assert events[0]["content"] == "Remnant"424 425 @pytest.mark.asyncio426 async def test_start_task_non_v1_url(self):427 """Test start_task with a non-v1 URL."""428 from cli.session import CLISession429 430 # URL not ending in /v1431 session = CLISession("/tmp", "http://localhost:8082")432 433 mock_process = AsyncMock()434 mock_process.stdout.read.side_effect = [b""]435 mock_process.stderr.read.return_value = b""436 mock_process.wait.return_value = 0437 438 with patch(439 "asyncio.create_subprocess_exec", new_callable=AsyncMock440 ) as mock_exec:441 mock_exec.return_value = mock_process442 async for _ in session.start_task("test"):443 pass444 445 # Check env var446 kwargs = mock_exec.call_args[1]447 env = kwargs["env"]448 assert env["ANTHROPIC_BASE_URL"] == "http://localhost:8082"449 450 @pytest.mark.asyncio451 async def test_start_task_allowed_dirs(self):452 """Test start_task includes allowed dirs in command."""453 from cli.session import CLISession454 455 session = CLISession(456 "/tmp", "http://localhost:8082/v1", allowed_dirs=["/dir1", "/dir2"]457 )458 459 mock_process = AsyncMock()460 mock_process.stdout.read.side_effect = [b""]461 mock_process.stderr.read.return_value = b""462 mock_process.wait.return_value = 0463 464 with patch(465 "asyncio.create_subprocess_exec", new_callable=AsyncMock466 ) as mock_exec:467 mock_exec.return_value = mock_process468 async for _ in session.start_task("test"):469 pass470 471 cmd = mock_exec.call_args[0]472 assert "--add-dir" in cmd473 assert os.path.normpath("/dir1") in cmd474 assert os.path.normpath("/dir2") in cmd475 476 @pytest.mark.asyncio477 async def test_start_task_plans_directory(self):478 """Test start_task includes --settings plansDirectory when plans_directory set."""479 from cli.session import CLISession480 481 session = CLISession(482 "/tmp",483 "http://localhost:8082/v1",484 plans_directory="./agent_workspace/plans",485 )486 487 mock_process = AsyncMock()488 mock_process.stdout.read.side_effect = [b""]489 mock_process.stderr.read.return_value = b""490 mock_process.wait.return_value = 0491 492 with patch(493 "asyncio.create_subprocess_exec", new_callable=AsyncMock494 ) as mock_exec:495 mock_exec.return_value = mock_process496 async for _ in session.start_task("test"):497 pass498 499 cmd = mock_exec.call_args[0]500 assert "--settings" in cmd501 settings_idx = cmd.index("--settings")502 assert settings_idx + 1 < len(cmd)503 settings = json.loads(cmd[settings_idx + 1])504 assert settings["plansDirectory"] == "./agent_workspace/plans"505 506 @pytest.mark.asyncio507 async def test_start_task_json_error(self):508 """Test handling of non-JSON output from CLI."""509 from cli.session import CLISession510 511 session = CLISession("/tmp", "http://localhost:8082/v1")512 513 mock_process = AsyncMock()514 mock_process.stdout.read.side_effect = [b"Not valid json\n", b""]515 mock_process.stderr.read.return_value = b""516 mock_process.wait.return_value = 0517 518 with patch(519 "asyncio.create_subprocess_exec", new_callable=AsyncMock520 ) as mock_exec:521 mock_exec.return_value = mock_process522 523 events = [e async for e in session.start_task("test") if e["type"] == "raw"]524 525 assert len(events) == 1526 assert events[0]["content"] == "Not valid json"527 528 @pytest.mark.asyncio529 async def test_stop_exception(self):530 """Test exception handling during stop."""531 from cli.session import CLISession532 533 session = CLISession("/tmp", "http://localhost:8082/v1")534 535 mock_process = MagicMock()536 mock_process.returncode = None537 # Raise exception on terminate538 mock_process.terminate.side_effect = RuntimeError("Permission denied")539 540 session.process = mock_process541 542 stopped = await session.stop()543 assert stopped is False544 545 546class TestCLISessionManager:547 """Test CLISessionManager."""548 549 @pytest.mark.asyncio550 async def test_manager_create_session(self):551 """Test creating a new session."""552 from cli.manager import CLISessionManager553 554 manager = CLISessionManager(555 workspace_path="/tmp/test",556 api_url="http://localhost:8082/v1",557 )558 559 session, sid, is_new = await manager.get_or_create_session()560 assert session is not None561 assert sid.startswith("pending_")562 assert is_new is True563 564 @pytest.mark.asyncio565 async def test_manager_reuse_session(self):566 """Test reusing an existing session."""567 from cli.manager import CLISessionManager568 569 manager = CLISessionManager(570 workspace_path="/tmp/test",571 api_url="http://localhost:8082/v1",572 )573 574 # Create first session575 s1, sid1, _is_new1 = await manager.get_or_create_session()576 577 # Request same session578 s2, _sid2, is_new2 = await manager.get_or_create_session(session_id=sid1)579 580 assert s1 is s2581 assert is_new2 is False582 583 @pytest.mark.asyncio584 async def test_manager_stats(self):585 """Test manager stats."""586 from cli.manager import CLISessionManager587 588 manager = CLISessionManager(589 workspace_path="/tmp/test",590 api_url="http://localhost:8082/v1",591 )592 593 stats = manager.get_stats()594 assert stats["active_sessions"] == 0595 assert stats["pending_sessions"] == 0596