248 lines
7.9 KiB
Python
248 lines
7.9 KiB
Python
"""Tests for session management."""
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.session_manager import BrowserSession, SessionManager, utc_now
|
|
|
|
|
|
class TestBrowserSession:
|
|
"""Tests for the BrowserSession class."""
|
|
|
|
def test_touch_updates_last_activity(self):
|
|
"""touch() should update last_activity timestamp."""
|
|
mock_context = MagicMock()
|
|
mock_page = MagicMock()
|
|
|
|
session = BrowserSession(
|
|
session_id="test-123",
|
|
allowed_domains=["example.com"],
|
|
context=mock_context,
|
|
page=mock_page,
|
|
)
|
|
|
|
original_activity = session.last_activity
|
|
session.touch()
|
|
|
|
assert session.last_activity >= original_activity
|
|
|
|
def test_increment_actions(self):
|
|
"""increment_actions() should update counter and touch."""
|
|
mock_context = MagicMock()
|
|
mock_page = MagicMock()
|
|
|
|
session = BrowserSession(
|
|
session_id="test-123",
|
|
allowed_domains=["example.com"],
|
|
context=mock_context,
|
|
page=mock_page,
|
|
)
|
|
|
|
assert session.actions_used == 0
|
|
session.increment_actions()
|
|
assert session.actions_used == 1
|
|
|
|
def test_actions_remaining(self):
|
|
"""actions_remaining should calculate correctly."""
|
|
mock_context = MagicMock()
|
|
mock_page = MagicMock()
|
|
|
|
session = BrowserSession(
|
|
session_id="test-123",
|
|
allowed_domains=["example.com"],
|
|
context=mock_context,
|
|
page=mock_page,
|
|
)
|
|
|
|
initial_remaining = session.actions_remaining
|
|
session.increment_actions()
|
|
assert session.actions_remaining == initial_remaining - 1
|
|
|
|
def test_is_expired_idle_timeout(self):
|
|
"""Session should expire after idle timeout."""
|
|
mock_context = MagicMock()
|
|
mock_page = MagicMock()
|
|
|
|
session = BrowserSession(
|
|
session_id="test-123",
|
|
allowed_domains=["example.com"],
|
|
context=mock_context,
|
|
page=mock_page,
|
|
)
|
|
|
|
# Set last_activity to past
|
|
session.last_activity = utc_now() - timedelta(seconds=400)
|
|
assert session.is_expired
|
|
|
|
def test_is_expired_max_lifetime(self):
|
|
"""Session should expire after max lifetime."""
|
|
mock_context = MagicMock()
|
|
mock_page = MagicMock()
|
|
|
|
session = BrowserSession(
|
|
session_id="test-123",
|
|
allowed_domains=["example.com"],
|
|
context=mock_context,
|
|
page=mock_page,
|
|
)
|
|
|
|
# Set created_at to past
|
|
session.created_at = utc_now() - timedelta(seconds=2000)
|
|
session.last_activity = utc_now() # Recent activity
|
|
assert session.is_expired
|
|
|
|
def test_not_expired_fresh_session(self):
|
|
"""Fresh session should not be expired."""
|
|
mock_context = MagicMock()
|
|
mock_page = MagicMock()
|
|
|
|
session = BrowserSession(
|
|
session_id="test-123",
|
|
allowed_domains=["example.com"],
|
|
context=mock_context,
|
|
page=mock_page,
|
|
)
|
|
|
|
assert not session.is_expired
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestSessionManager:
|
|
"""Tests for the SessionManager class."""
|
|
|
|
async def test_create_session_success(self):
|
|
"""create_session should create a new session."""
|
|
mock_browser = AsyncMock()
|
|
mock_context = AsyncMock()
|
|
mock_page = AsyncMock()
|
|
|
|
mock_browser.new_context.return_value = mock_context
|
|
mock_context.new_page.return_value = mock_page
|
|
|
|
manager = SessionManager(mock_browser)
|
|
session = await manager.create_session(["example.com"])
|
|
|
|
assert session is not None
|
|
assert session.session_id is not None
|
|
assert session.allowed_domains == ["example.com"]
|
|
assert manager.active_session_count == 1
|
|
|
|
async def test_create_session_empty_domains_raises(self):
|
|
"""create_session with empty domains should raise."""
|
|
mock_browser = AsyncMock()
|
|
manager = SessionManager(mock_browser)
|
|
|
|
with pytest.raises(ValueError):
|
|
await manager.create_session([])
|
|
|
|
async def test_create_session_max_limit(self):
|
|
"""create_session should fail when max sessions reached."""
|
|
mock_browser = AsyncMock()
|
|
mock_context = AsyncMock()
|
|
mock_page = AsyncMock()
|
|
|
|
mock_browser.new_context.return_value = mock_context
|
|
mock_context.new_page.return_value = mock_page
|
|
|
|
manager = SessionManager(mock_browser)
|
|
|
|
# Create max sessions
|
|
with patch("app.session_manager.settings") as mock_settings:
|
|
mock_settings.max_sessions = 2
|
|
mock_settings.max_actions_per_session = 50
|
|
mock_settings.idle_timeout_seconds = 300
|
|
mock_settings.max_session_lifetime_seconds = 1800
|
|
mock_settings.default_timeout_ms = 30000
|
|
mock_settings.navigation_timeout_ms = 60000
|
|
|
|
await manager.create_session(["example.com"])
|
|
await manager.create_session(["other.com"])
|
|
|
|
with pytest.raises(ValueError) as exc_info:
|
|
await manager.create_session(["blocked.com"])
|
|
|
|
assert "Maximum sessions" in str(exc_info.value)
|
|
|
|
async def test_get_session_returns_session(self):
|
|
"""get_session should return existing session."""
|
|
mock_browser = AsyncMock()
|
|
mock_context = AsyncMock()
|
|
mock_page = AsyncMock()
|
|
|
|
mock_browser.new_context.return_value = mock_context
|
|
mock_context.new_page.return_value = mock_page
|
|
|
|
manager = SessionManager(mock_browser)
|
|
session = await manager.create_session(["example.com"])
|
|
|
|
retrieved = await manager.get_session(session.session_id)
|
|
assert retrieved is session
|
|
|
|
async def test_get_session_returns_none_for_unknown(self):
|
|
"""get_session should return None for unknown session."""
|
|
mock_browser = AsyncMock()
|
|
manager = SessionManager(mock_browser)
|
|
|
|
retrieved = await manager.get_session("unknown-id")
|
|
assert retrieved is None
|
|
|
|
async def test_close_session_removes_session(self):
|
|
"""close_session should remove and close session."""
|
|
mock_browser = AsyncMock()
|
|
mock_context = AsyncMock()
|
|
mock_page = AsyncMock()
|
|
|
|
mock_browser.new_context.return_value = mock_context
|
|
mock_context.new_page.return_value = mock_page
|
|
|
|
manager = SessionManager(mock_browser)
|
|
session = await manager.create_session(["example.com"])
|
|
|
|
assert manager.active_session_count == 1
|
|
|
|
result = await manager.close_session(session.session_id)
|
|
|
|
assert result is True
|
|
assert manager.active_session_count == 0
|
|
mock_context.close.assert_called_once()
|
|
|
|
async def test_close_session_idempotent(self):
|
|
"""close_session should be idempotent."""
|
|
mock_browser = AsyncMock()
|
|
mock_context = AsyncMock()
|
|
mock_page = AsyncMock()
|
|
|
|
mock_browser.new_context.return_value = mock_context
|
|
mock_context.new_page.return_value = mock_page
|
|
|
|
manager = SessionManager(mock_browser)
|
|
session = await manager.create_session(["example.com"])
|
|
|
|
await manager.close_session(session.session_id)
|
|
result = await manager.close_session(session.session_id)
|
|
|
|
assert result is False # Already closed
|
|
|
|
async def test_close_all_sessions(self):
|
|
"""close_all_sessions should close all sessions."""
|
|
mock_browser = AsyncMock()
|
|
mock_context = AsyncMock()
|
|
mock_page = AsyncMock()
|
|
|
|
mock_browser.new_context.return_value = mock_context
|
|
mock_context.new_page.return_value = mock_page
|
|
|
|
manager = SessionManager(mock_browser)
|
|
|
|
await manager.create_session(["example.com"])
|
|
await manager.create_session(["other.com"])
|
|
|
|
assert manager.active_session_count == 2
|
|
|
|
await manager.close_all_sessions()
|
|
|
|
assert manager.active_session_count == 0
|