File size: 3,577 Bytes
6172a47 | 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 | 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 cli.manager import CLISessionManager
with patch("cli.manager.CLISession") 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 = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")
session, temp_id, is_new = await manager.get_or_create_session()
assert session is mock_session
assert is_new is True
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 cli.manager import CLISessionManager
manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")
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 cli.manager import CLISessionManager
with patch("cli.manager.CLISession") 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 = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")
_, 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 cli.manager import CLISessionManager
with patch("cli.manager.CLISession") 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 = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")
_, 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_handles_stop_exceptions():
from cli.manager import CLISessionManager
manager = CLISessionManager(workspace_path="/tmp", api_url="http://x/v1")
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
await manager.stop_all()
s1.stop.assert_awaited_once()
s2.stop.assert_awaited_once()
assert manager.get_stats()["active_sessions"] == 0
assert manager.get_stats()["pending_sessions"] == 0
|