learningcircuit--local-deep-research/tests/security/test_account_lockout.py

"""Tests for per-user account lockout."""

from datetime import datetime, timedelta, timezone
from unittest.mock import patch

from local_deep_research.security.account_lockout import AccountLockoutManager


class TestAccountLockout:
    """Unit tests for AccountLockoutManager."""

    def _make_manager(self, threshold=3, lockout_minutes=15):
        return AccountLockoutManager(
            threshold=threshold, lockout_minutes=lockout_minutes
        )

    def test_not_locked_initially(self):
        mgr = self._make_manager()
        assert mgr.is_locked("alice") is False

    def test_not_locked_below_threshold(self):
        mgr = self._make_manager(threshold=3)
        mgr.record_failure("alice")
        mgr.record_failure("alice")
        assert mgr.is_locked("alice") is False

    def test_locked_at_threshold(self):
        mgr = self._make_manager(threshold=3)
        for _ in range(3):
            mgr.record_failure("alice")
        assert mgr.is_locked("alice") is True

    def test_success_resets_lockout(self):
        mgr = self._make_manager(threshold=3)
        for _ in range(3):
            mgr.record_failure("alice")
        assert mgr.is_locked("alice") is True
        mgr.record_success("alice")
        assert mgr.is_locked("alice") is False

    def test_counter_resets_after_success(self):
        mgr = self._make_manager(threshold=10)
        for _ in range(9):
            mgr.record_failure("alice")
        mgr.record_success("alice")
        # After reset, 9 more failures should not lock (threshold is 10)
        for _ in range(9):
            mgr.record_failure("alice")
        assert mgr.is_locked("alice") is False

    def test_independent_users(self):
        mgr = self._make_manager(threshold=3)
        for _ in range(3):
            mgr.record_failure("alice")
        assert mgr.is_locked("alice") is True
        assert mgr.is_locked("bob") is False

    def test_success_on_nonexistent_user_is_noop(self):
        mgr = self._make_manager()
        # Should not raise
        mgr.record_success("nonexistent")
        assert mgr.is_locked("nonexistent") is False

    def test_lockout_expires_after_duration(self):
        mgr = self._make_manager(threshold=3, lockout_minutes=15)
        now = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)

        with patch(
            "local_deep_research.security.account_lockout.datetime"
        ) as mock_dt:
            mock_dt.now.return_value = now
            mock_dt.side_effect = lambda *a, **kw: datetime(*a, **kw)
            for _ in range(3):
                mgr.record_failure("alice")

        # Still locked 14 minutes later
        with patch(
            "local_deep_research.security.account_lockout.datetime"
        ) as mock_dt:
            mock_dt.now.return_value = now + timedelta(minutes=14)
            assert mgr.is_locked("alice") is True

        # Unlocked at exactly 15 minutes
        with patch(
            "local_deep_research.security.account_lockout.datetime"
        ) as mock_dt:
            mock_dt.now.return_value = now + timedelta(minutes=15)
            assert mgr.is_locked("alice") is False

    def test_eviction_removes_unlocked_entries(self):
        """Eviction should remove unlocked/expired entries, not blanket clear."""
        mgr = self._make_manager(threshold=5)
        mgr._MAX_STATE_ENTRIES = 100
        now = datetime.now(timezone.utc)

        # Add unlocked entries (count < threshold, no locked_until)
        for i in range(90):
            mgr._state[f"unlocked_{i}"] = {"count": 1, "locked_until": None}

        # Add expired locked entries
        for i in range(10):
            mgr._state[f"expired_{i}"] = {
                "count": 5,
                "locked_until": now - timedelta(minutes=1),
            }

        # Add actively locked entries
        for i in range(5):
            mgr._state[f"locked_{i}"] = {
                "count": 5,
                "locked_until": now + timedelta(minutes=10),
            }

        assert len(mgr._state) == 105

        # Trigger eviction via record_failure
        mgr.record_failure("attacker")

        # Actively locked entries should survive
        for i in range(5):
            assert f"locked_{i}" in mgr._state
        # New entry should exist
        assert "attacker" in mgr._state
        # Unlocked/expired entries should be gone
        assert not any(k.startswith("unlocked_") for k in mgr._state)
        assert not any(k.startswith("expired_") for k in mgr._state)

    def test_blanket_clear_as_last_resort(self):
        """If eviction can't reduce below limit, blanket clear."""
        mgr = self._make_manager(threshold=3)
        mgr._MAX_STATE_ENTRIES = 10
        now = datetime.now(timezone.utc)

        # Fill with actively locked entries (can't be evicted)
        for i in range(15):
            mgr._state[f"locked_{i}"] = {
                "count": 5,
                "locked_until": now + timedelta(hours=1),
            }

        mgr.record_failure("attacker")
        # Should have blanket-cleared then added new entry
        assert len(mgr._state) == 1
        assert "attacker" in mgr._state