Jack1808/Claude_Code

1from unittest.mock import AsyncMock, MagicMock, patch2 3import pytest4 5 6@pytest.mark.asyncio7async def test_register_real_session_id_moves_pending_to_active_and_maps():8    from cli.manager import CLISessionManager9 10    with patch("cli.manager.CLISession") as mock_session_cls:11        mock_session = MagicMock()12        mock_session.is_busy = False13        mock_session.stop = AsyncMock(return_value=True)14        mock_session_cls.return_value = mock_session15 16        manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")17        session, temp_id, is_new = await manager.get_or_create_session()18        assert session is mock_session19        assert is_new is True20 21        ok = await manager.register_real_session_id(temp_id, "real_1")22        assert ok is True23 24        # Lookup via temp id should resolve to the real session id.25        s2, sid2, is_new2 = await manager.get_or_create_session(session_id=temp_id)26        assert s2 is mock_session27        assert sid2 == "real_1"28        assert is_new2 is False29 30 31@pytest.mark.asyncio32async def test_register_real_session_id_missing_temp_id_returns_false():33    from cli.manager import CLISessionManager34 35    manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")36    ok = await manager.register_real_session_id("missing", "real_1")37    assert ok is False38 39 40@pytest.mark.asyncio41async def test_remove_session_pending_stops_and_returns_true():42    from cli.manager import CLISessionManager43 44    with patch("cli.manager.CLISession") as mock_session_cls:45        mock_session = MagicMock()46        mock_session.is_busy = False47        mock_session.stop = AsyncMock(return_value=True)48        mock_session_cls.return_value = mock_session49 50        manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")51        _, temp_id, _ = await manager.get_or_create_session()52 53        removed = await manager.remove_session(temp_id)54        assert removed is True55        mock_session.stop.assert_awaited_once()56 57 58@pytest.mark.asyncio59async def test_remove_session_active_removes_temp_mapping():60    from cli.manager import CLISessionManager61 62    with patch("cli.manager.CLISession") as mock_session_cls:63        mock_session = MagicMock()64        mock_session.is_busy = False65        mock_session.stop = AsyncMock(return_value=True)66        mock_session_cls.return_value = mock_session67 68        manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")69        _, temp_id, _ = await manager.get_or_create_session()70        await manager.register_real_session_id(temp_id, "real_1")71 72        removed = await manager.remove_session("real_1")73        assert removed is True74 75        # Temp ID should no longer resolve to an active session after removal.76        _, sid2, is_new2 = await manager.get_or_create_session(session_id=temp_id)77        assert sid2 == temp_id78        assert is_new2 is True79 80 81@pytest.mark.asyncio82async def test_stop_all_handles_stop_exceptions():83    from cli.manager import CLISessionManager84 85    manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")86 87    s1 = MagicMock()88    s1.stop = AsyncMock(side_effect=RuntimeError("boom"))89    s1.is_busy = False90 91    s2 = MagicMock()92    s2.stop = AsyncMock(return_value=True)93    s2.is_busy = False94 95    manager._sessions["a"] = s196    manager._pending_sessions["b"] = s297 98    await manager.stop_all()99    s1.stop.assert_awaited_once()100    s2.stop.assert_awaited_once()101    assert manager.get_stats()["active_sessions"] == 0102    assert manager.get_stats()["pending_sessions"] == 0103