learningcircuit--local-deep-research/tests/web/auth/test_session_manager.py

"""
Tests for web/auth/session_manager.py

Tests cover:
- SessionManager class initialization with mocked security defaults
- Session creation, validation, and destruction
- Session cleanup and expiration using datetime mocking
- User session management and session info formatting
"""

import datetime
from datetime import UTC
from unittest.mock import patch

import pytest

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

MOCK_SESSION_TIMEOUT_HOURS = 2
MOCK_REMEMBER_ME_DAYS = 30

# Patch target for the security helper used inside session_manager.py
_SECURITY_DEFAULT_PATCH = (
    "local_deep_research.web.auth.session_manager.get_security_default"
)


def _security_default_side_effect(key, default):
    """Return controlled timeout values for tests."""
    mapping = {
        "security.session_timeout_hours": MOCK_SESSION_TIMEOUT_HOURS,
        "security.session_remember_me_days": MOCK_REMEMBER_ME_DAYS,
    }
    return mapping.get(key, default)


@pytest.fixture()
def manager():
    """Create a SessionManager with mocked security defaults."""
    with patch(
        _SECURITY_DEFAULT_PATCH, side_effect=_security_default_side_effect
    ):
        from local_deep_research.web.auth.session_manager import (
            SessionManager,
        )

        return SessionManager()


# ===========================================================================
# Initialization
# ===========================================================================


class TestSessionManagerInit:
    """Tests for SessionManager initialization."""

    def test_init_creates_empty_sessions_dict(self, manager):
        """Should initialize with an empty sessions dict."""
        assert manager.sessions == {}

    def test_init_sets_session_timeout_from_security_defaults(self, manager):
        """Should set session_timeout based on mocked security default."""
        assert manager.session_timeout == datetime.timedelta(
            hours=MOCK_SESSION_TIMEOUT_HOURS
        )

    def test_init_sets_remember_me_timeout_from_security_defaults(
        self, manager
    ):
        """Should set remember_me_timeout based on mocked security default."""
        assert manager.remember_me_timeout == datetime.timedelta(
            days=MOCK_REMEMBER_ME_DAYS
        )

    def test_init_calls_get_security_default(self):
        """Should call get_security_default for both timeout settings."""
        with patch(_SECURITY_DEFAULT_PATCH) as mock_get:
            mock_get.return_value = 1

            from local_deep_research.web.auth.session_manager import (
                SessionManager,
            )

            SessionManager()

            assert mock_get.call_count == 2
            call_keys = [c.args[0] for c in mock_get.call_args_list]
            assert "security.session_timeout_hours" in call_keys
            assert "security.session_remember_me_days" in call_keys

    def test_init_uses_custom_timeout_values(self):
        """Should honour non-default timeout values from security settings."""
        custom_hours = 8
        custom_days = 7

        def custom_side_effect(key, default):
            if key == "security.session_timeout_hours":
                return custom_hours
            if key == "security.session_remember_me_days":
                return custom_days
            return default

        with patch(_SECURITY_DEFAULT_PATCH, side_effect=custom_side_effect):
            from local_deep_research.web.auth.session_manager import (
                SessionManager,
            )

            mgr = SessionManager()

        assert mgr.session_timeout == datetime.timedelta(hours=custom_hours)
        assert mgr.remember_me_timeout == datetime.timedelta(days=custom_days)


# ===========================================================================
# Session creation
# ===========================================================================


class TestCreateSession:
    """Tests for create_session method."""

    def test_returns_string_token(self, manager):
        """Should return a session ID that is a non-empty string."""
        session_id = manager.create_session("alice")
        assert isinstance(session_id, str)
        assert len(session_id) > 0

    def test_returns_unique_tokens(self, manager):
        """Should generate unique session IDs across multiple calls."""
        ids = {manager.create_session(f"user{i}") for i in range(50)}
        assert len(ids) == 50

    def test_stores_session_data(self, manager):
        """Should store session entry in the sessions dict."""
        sid = manager.create_session("bob")
        assert sid in manager.sessions
        data = manager.sessions[sid]
        assert data["username"] == "bob"
        assert data["remember_me"] is False
        assert isinstance(data["created_at"], datetime.datetime)
        assert isinstance(data["last_access"], datetime.datetime)

    def test_default_remember_me_is_false(self, manager):
        """Should default remember_me to False."""
        sid = manager.create_session("carol")
        assert manager.sessions[sid]["remember_me"] is False

    def test_remember_me_true(self, manager):
        """Should store remember_me=True when requested."""
        sid = manager.create_session("dave", remember_me=True)
        assert manager.sessions[sid]["remember_me"] is True

    def test_timestamps_are_utc_aware(self, manager):
        """Should set created_at and last_access as UTC-aware datetimes."""
        sid = manager.create_session("eve")
        data = manager.sessions[sid]
        assert data["created_at"].tzinfo is not None
        assert data["last_access"].tzinfo is not None

    def test_multiple_sessions_same_user(self, manager):
        """Should allow multiple sessions for the same username."""
        sid1 = manager.create_session("frank")
        sid2 = manager.create_session("frank")
        assert sid1 != sid2
        assert sid1 in manager.sessions
        assert sid2 in manager.sessions
        assert manager.sessions[sid1]["username"] == "frank"
        assert manager.sessions[sid2]["username"] == "frank"


# ===========================================================================
# Session validation
# ===========================================================================


class TestValidateSession:
    """Tests for validate_session method."""

    def test_returns_username_for_valid_session(self, manager):
        """Should return the username for a freshly created session."""
        sid = manager.create_session("grace")
        assert manager.validate_session(sid) == "grace"

    def test_returns_none_for_unknown_session_id(self, manager):
        """Should return None for a session_id that was never created."""
        assert manager.validate_session("totally-bogus-id") is None

    def test_returns_none_for_expired_regular_session(self, manager):
        """Should return None when regular session has exceeded session_timeout."""
        fixed_now = datetime.datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC)
        expired_time = fixed_now - datetime.timedelta(
            hours=MOCK_SESSION_TIMEOUT_HOURS, seconds=1
        )

        with patch(
            "local_deep_research.web.auth.session_manager.datetime"
        ) as mock_dt:
            mock_dt.datetime.now.return_value = fixed_now
            mock_dt.timedelta = datetime.timedelta
            mock_dt.UTC = UTC

            sid = manager.create_session("hank")
            # Manually set last_access to an expired time
            manager.sessions[sid]["last_access"] = expired_time

            result = manager.validate_session(sid)

        assert result is None

    def test_destroys_expired_session(self, manager):
        """Should remove the session from self.sessions when it expires."""
        sid = manager.create_session("ivy")
        manager.sessions[sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        manager.validate_session(sid)
        assert sid not in manager.sessions

    def test_returns_none_for_expired_remember_me_session(self, manager):
        """Should return None for remember_me session past remember_me_timeout."""
        sid = manager.create_session("jack", remember_me=True)
        manager.sessions[sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(days=MOCK_REMEMBER_ME_DAYS + 1)

        assert manager.validate_session(sid) is None
        assert sid not in manager.sessions

    def test_remember_me_survives_regular_timeout(self, manager):
        """A remember_me session should still be valid after regular timeout elapses."""
        sid = manager.create_session("kate", remember_me=True)
        # 3 hours ago exceeds the 2-hour regular timeout but not 30-day remember_me
        manager.sessions[sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        result = manager.validate_session(sid)
        assert result == "kate"

    def test_updates_last_access_on_valid_session(self, manager):
        """Should update last_access to current time when session is valid."""
        sid = manager.create_session("leo")
        old_access = manager.sessions[sid]["last_access"]

        # Nudge time forward slightly
        with patch(
            "local_deep_research.web.auth.session_manager.datetime"
        ) as mock_dt:
            future = old_access + datetime.timedelta(minutes=10)
            mock_dt.datetime.now.return_value = future
            mock_dt.timedelta = datetime.timedelta
            mock_dt.UTC = UTC

            result = manager.validate_session(sid)

        assert result == "leo"
        assert manager.sessions[sid]["last_access"] == future

    def test_does_not_update_last_access_for_unknown_session(self, manager):
        """Calling validate on unknown id should not create any session entry."""
        manager.validate_session("nonexistent")
        assert "nonexistent" not in manager.sessions

    def test_expired_boundary_exact_timeout(self, manager):
        """Session at exactly the timeout boundary should still be valid (uses >)."""
        fixed_now = datetime.datetime(2026, 6, 15, 12, 0, 0, tzinfo=UTC)
        boundary_time = fixed_now - datetime.timedelta(
            hours=MOCK_SESSION_TIMEOUT_HOURS
        )

        sid = manager.create_session("mia")
        manager.sessions[sid]["last_access"] = boundary_time

        with patch(
            "local_deep_research.web.auth.session_manager.datetime"
        ) as mock_dt:
            mock_dt.datetime.now.return_value = fixed_now
            mock_dt.timedelta = datetime.timedelta
            mock_dt.UTC = UTC

            # Source code checks ``now - last_access > timeout``,
            # so exactly equal should still be valid.
            result = manager.validate_session(sid)

        assert result == "mia"


# ===========================================================================
# Session destruction
# ===========================================================================


class TestDestroySession:
    """Tests for destroy_session method."""

    def test_removes_session(self, manager):
        """Should remove the session from self.sessions."""
        sid = manager.create_session("nina")
        assert sid in manager.sessions
        manager.destroy_session(sid)
        assert sid not in manager.sessions

    def test_noop_for_unknown_session_id(self, manager):
        """Should not raise when destroying an unknown session_id."""
        manager.destroy_session("does-not-exist")  # no exception

    def test_does_not_affect_other_sessions(self, manager):
        """Destroying one session should leave others intact."""
        sid1 = manager.create_session("oscar")
        sid2 = manager.create_session("oscar")
        manager.destroy_session(sid1)
        assert sid1 not in manager.sessions
        assert sid2 in manager.sessions


# ===========================================================================
# Cleanup expired sessions
# ===========================================================================


class TestCleanupExpiredSessions:
    """Tests for cleanup_expired_sessions method."""

    def test_removes_expired_regular_sessions(self, manager):
        """Should remove regular sessions that have exceeded session_timeout."""
        sid = manager.create_session("quinn")
        manager.sessions[sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        manager.cleanup_expired_sessions()
        assert sid not in manager.sessions

    def test_keeps_valid_regular_sessions(self, manager):
        """Should keep regular sessions that have not expired."""
        sid = manager.create_session("rosa")
        manager.cleanup_expired_sessions()
        assert sid in manager.sessions

    def test_removes_only_expired_sessions(self, manager):
        """Should remove expired sessions while keeping valid ones."""
        expired_sid = manager.create_session("stale_user")
        manager.sessions[expired_sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        valid_sid = manager.create_session("fresh_user")

        manager.cleanup_expired_sessions()

        assert expired_sid not in manager.sessions
        assert valid_sid in manager.sessions

    def test_removes_multiple_expired_sessions(self, manager):
        """Should remove all expired sessions in a single cleanup pass."""
        expired_ids = []
        for i in range(5):
            sid = manager.create_session(f"expired_{i}")
            manager.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)
            expired_ids.append(sid)

        valid_sid = manager.create_session("keeper")
        manager.cleanup_expired_sessions()

        for eid in expired_ids:
            assert eid not in manager.sessions
        assert valid_sid in manager.sessions

    def test_respects_remember_me_timeout(self, manager):
        """Should keep remember_me sessions that exceed regular timeout but not remember_me_timeout."""
        sid = manager.create_session("tina", remember_me=True)
        # 3 hours ago: past regular timeout but within remember_me timeout
        manager.sessions[sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        manager.cleanup_expired_sessions()
        assert sid in manager.sessions

    def test_removes_expired_remember_me_sessions(self, manager):
        """Should remove remember_me sessions that exceed remember_me_timeout."""
        sid = manager.create_session("uma", remember_me=True)
        manager.sessions[sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(days=MOCK_REMEMBER_ME_DAYS + 1)

        manager.cleanup_expired_sessions()
        assert sid not in manager.sessions

    def test_handles_empty_sessions(self, manager):
        """Should not raise when there are no sessions."""
        manager.cleanup_expired_sessions()  # no exception
        assert manager.sessions == {}

    def test_mixed_regular_and_remember_me(self, manager):
        """Should correctly handle a mix of regular and remember_me sessions."""
        # Expired regular
        s1 = manager.create_session("u1")
        manager.sessions[s1]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        # Valid remember_me (same age as s1 but has longer timeout)
        s2 = manager.create_session("u2", remember_me=True)
        manager.sessions[s2]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        # Expired remember_me
        s3 = manager.create_session("u3", remember_me=True)
        manager.sessions[s3]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(days=MOCK_REMEMBER_ME_DAYS + 1)

        # Valid regular
        s4 = manager.create_session("u4")

        manager.cleanup_expired_sessions()

        assert s1 not in manager.sessions  # expired regular
        assert s2 in manager.sessions  # still valid remember_me
        assert s3 not in manager.sessions  # expired remember_me
        assert s4 in manager.sessions  # still valid regular


# ===========================================================================
# Active sessions count
# ===========================================================================


class TestGetActiveSessionsCount:
    """Tests for get_active_sessions_count method."""

    def test_returns_zero_when_empty(self, manager):
        """Should return 0 when there are no sessions."""
        assert manager.get_active_sessions_count() == 0

    def test_returns_correct_count(self, manager):
        """Should return the number of active sessions."""
        for i in range(4):
            manager.create_session(f"user{i}")
        assert manager.get_active_sessions_count() == 4

    def test_triggers_cleanup_before_counting(self, manager):
        """Should clean up expired sessions before counting."""
        for i in range(3):
            manager.create_session(f"active_{i}")

        expired_sid = manager.create_session("stale")
        manager.sessions[expired_sid]["last_access"] = datetime.datetime.now(
            UTC
        ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        count = manager.get_active_sessions_count()
        assert count == 3
        assert expired_sid not in manager.sessions

    def test_count_after_all_expired(self, manager):
        """Should return 0 when all sessions have expired."""
        for i in range(3):
            sid = manager.create_session(f"user{i}")
            manager.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - datetime.timedelta(hours=MOCK_SESSION_TIMEOUT_HOURS + 1)

        assert manager.get_active_sessions_count() == 0


# ===========================================================================
# User sessions
# ===========================================================================


class TestGetUserSessions:
    """Tests for get_user_sessions method."""

    def test_returns_empty_list_when_no_sessions(self, manager):
        """Should return an empty list when the user has no sessions."""
        assert manager.get_user_sessions("nobody") == []

    def test_returns_only_matching_user_sessions(self, manager):
        """Should return sessions only for the specified username."""
        manager.create_session("alice")
        manager.create_session("alice")
        manager.create_session("bob")

        alice_sessions = manager.get_user_sessions("alice")
        assert len(alice_sessions) == 2

        bob_sessions = manager.get_user_sessions("bob")
        assert len(bob_sessions) == 1

    def test_does_not_return_other_users_sessions(self, manager):
        """Should never include sessions belonging to other users."""
        manager.create_session("charlie")
        manager.create_session("diana")

        result = manager.get_user_sessions("charlie")
        assert len(result) == 1

    def test_truncates_session_id(self, manager):
        """Should truncate session_id to first 8 characters followed by '...'."""
        sid = manager.create_session("eve")
        result = manager.get_user_sessions("eve")

        assert len(result) == 1
        truncated_id = result[0]["session_id"]
        assert truncated_id == sid[:8] + "..."
        assert len(truncated_id) == 11  # 8 chars + 3 dots

    def test_returns_expected_keys(self, manager):
        """Should return dicts with session_id, created_at, last_access, remember_me."""
        manager.create_session("frank", remember_me=True)
        result = manager.get_user_sessions("frank")

        assert len(result) == 1
        info = result[0]
        assert set(info.keys()) == {
            "session_id",
            "created_at",
            "last_access",
            "remember_me",
        }
        assert info["remember_me"] is True
        assert isinstance(info["created_at"], datetime.datetime)
        assert isinstance(info["last_access"], datetime.datetime)

    def test_multiple_sessions_same_user(self, manager):
        """Should list all sessions when a user has multiple active sessions."""
        sid1 = manager.create_session("grace")
        sid2 = manager.create_session("grace")

        result = manager.get_user_sessions("grace")
        assert len(result) == 2

        returned_ids = {r["session_id"] for r in result}
        assert returned_ids == {sid1[:8] + "...", sid2[:8] + "..."}

    def test_returns_remember_me_flag_per_session(self, manager):
        """Each session entry should reflect its own remember_me setting."""
        manager.create_session("hana", remember_me=False)
        manager.create_session("hana", remember_me=True)

        result = manager.get_user_sessions("hana")
        flags = {r["remember_me"] for r in result}
        assert flags == {True, False}


class TestDestroyAllUserSessions:
    """Tests for destroy_all_user_sessions."""

    def test_destroys_all_sessions_for_user(self, manager):
        """Should destroy every session belonging to the target user."""
        manager.create_session("alice", remember_me=False)
        manager.create_session("alice", remember_me=False)
        manager.create_session("bob", remember_me=False)

        count = manager.destroy_all_user_sessions("alice")

        assert count == 2
        assert manager.get_user_sessions("alice") == []
        # Bob's session should be untouched
        assert len(manager.get_user_sessions("bob")) == 1

    def test_returns_zero_for_unknown_user(self, manager):
        """Should return 0 and not raise when user has no sessions."""
        count = manager.destroy_all_user_sessions("nobody")
        assert count == 0

    def test_returns_zero_after_double_call(self, manager):
        """Calling twice should be idempotent."""
        manager.create_session("alice", remember_me=False)
        manager.destroy_all_user_sessions("alice")
        count = manager.destroy_all_user_sessions("alice")
        assert count == 0