File size: 4,375 Bytes
f51dab6 133255a f51dab6 0d0f0f9 f51dab6 0d0f0f9 f51dab6 4a8663f 2fca8d0 568f343 f51dab6 568f343 f51dab6 0d0f0f9 f51dab6 2fca8d0 f51dab6 0d0f0f9 f51dab6 0d0f0f9 f51dab6 4a8663f 2fca8d0 4a8663f f51dab6 0d0f0f9 f51dab6 0d0f0f9 f51dab6 4a8663f 2fca8d0 4a8663f f51dab6 e0e03f1 f51dab6 01b2c65 0d0f0f9 f51dab6 2fca8d0 f51dab6 01b2c65 f51dab6 01b2c65 f51dab6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@pytest.mark.asyncio
async def test_register_real_session_id_moves_pending_to_active_and_maps():
from free_claude_code.cli.managed.manager import ManagedClaudeSessionManager
with patch(
"free_claude_code.cli.managed.manager.ManagedClaudeSession"
) as mock_session_cls:
mock_session = MagicMock()
mock_session.is_busy = False
mock_session.stop = AsyncMock(return_value=True)
mock_session_cls.return_value = mock_session
manager = ManagedClaudeSessionManager(
workspace_path="/tmp",
proxy_root_url="http://x",
auth_token="proxy-token",
)
session, temp_id, is_new = await manager.get_or_create_session()
assert session is mock_session
assert is_new is True
mock_session_cls.assert_called_once()
assert mock_session_cls.call_args.kwargs["auth_token"] == "proxy-token"
ok = await manager.register_real_session_id(temp_id, "real_1")
assert ok is True
# Lookup via temp id should resolve to the real session id.
s2, sid2, is_new2 = await manager.get_or_create_session(session_id=temp_id)
assert s2 is mock_session
assert sid2 == "real_1"
assert is_new2 is False
@pytest.mark.asyncio
async def test_register_real_session_id_missing_temp_id_returns_false():
from free_claude_code.cli.managed.manager import ManagedClaudeSessionManager
manager = ManagedClaudeSessionManager(
workspace_path="/tmp", proxy_root_url="http://x"
)
ok = await manager.register_real_session_id("missing", "real_1")
assert ok is False
@pytest.mark.asyncio
async def test_remove_session_pending_stops_and_returns_true():
from free_claude_code.cli.managed.manager import ManagedClaudeSessionManager
with patch(
"free_claude_code.cli.managed.manager.ManagedClaudeSession"
) as mock_session_cls:
mock_session = MagicMock()
mock_session.is_busy = False
mock_session.stop = AsyncMock(return_value=True)
mock_session_cls.return_value = mock_session
manager = ManagedClaudeSessionManager(
workspace_path="/tmp", proxy_root_url="http://x"
)
_, temp_id, _ = await manager.get_or_create_session()
removed = await manager.remove_session(temp_id)
assert removed is True
mock_session.stop.assert_awaited_once()
@pytest.mark.asyncio
async def test_remove_session_active_removes_temp_mapping():
from free_claude_code.cli.managed.manager import ManagedClaudeSessionManager
with patch(
"free_claude_code.cli.managed.manager.ManagedClaudeSession"
) as mock_session_cls:
mock_session = MagicMock()
mock_session.is_busy = False
mock_session.stop = AsyncMock(return_value=True)
mock_session_cls.return_value = mock_session
manager = ManagedClaudeSessionManager(
workspace_path="/tmp", proxy_root_url="http://x"
)
_, temp_id, _ = await manager.get_or_create_session()
await manager.register_real_session_id(temp_id, "real_1")
removed = await manager.remove_session("real_1")
assert removed is True
# Temp ID should no longer resolve to an active session after removal.
_, sid2, is_new2 = await manager.get_or_create_session(session_id=temp_id)
assert sid2 == temp_id
assert is_new2 is True
@pytest.mark.asyncio
async def test_stop_all_reports_and_retains_stop_exceptions():
from free_claude_code.cli.managed.manager import ManagedClaudeSessionManager
manager = ManagedClaudeSessionManager(
workspace_path="/tmp", proxy_root_url="http://x"
)
s1 = MagicMock()
s1.stop = AsyncMock(side_effect=RuntimeError("boom"))
s1.is_busy = False
s2 = MagicMock()
s2.stop = AsyncMock(return_value=True)
s2.is_busy = False
manager._sessions["a"] = s1
manager._pending_sessions["b"] = s2
with pytest.raises(
RuntimeError,
match=r"^Managed Claude session shutdown failures: 1\.$",
):
await manager.stop_all()
s1.stop.assert_awaited_once()
s2.stop.assert_awaited_once()
assert manager.get_stats()["active_sessions"] == 1
assert manager.get_stats()["pending_sessions"] == 0
|