letsbe-mcp-browser/tests/test_session_manager.py

"""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