learningcircuit--local-deep-research/tests/database/test_credential_store_extended.py

"""
Extended tests for credential store base class.

Tests cover:
- TTL expiration behavior
- Concurrent access patterns
- Thread safety
- Edge cases and error conditions
- Memory management
- Multiple credentials handling
"""

import time
import threading
from datetime import timedelta

import pytest
from freezegun import freeze_time

from local_deep_research.database.credential_store_base import (
    CredentialStoreBase,
)


class ConcreteCredentialStore(CredentialStoreBase):
    """Concrete implementation for testing."""

    def store(self, key: str, username: str, password: str):
        self._store_credentials(
            key, {"username": username, "password": password}
        )

    def retrieve(self, key: str, remove: bool = False):
        return self._retrieve_credentials(key, remove=remove)


@pytest.fixture
def store():
    """Create a credential store with 1 hour TTL."""
    return ConcreteCredentialStore(ttl_seconds=3600)


@pytest.fixture
def short_ttl_store():
    """Create a credential store with very short TTL."""
    return ConcreteCredentialStore(ttl_seconds=1)


class TestCredentialStoreInitialization:
    """Tests for credential store initialization."""

    def test_store_initializes_with_ttl(self):
        """Store should initialize with given TTL."""
        store = ConcreteCredentialStore(ttl_seconds=7200)
        assert store.ttl == 7200

    def test_store_initializes_empty(self):
        """Store should start empty."""
        store = ConcreteCredentialStore(ttl_seconds=3600)
        assert len(store._store) == 0

    def test_store_has_lock(self):
        """Store should have a threading lock."""
        store = ConcreteCredentialStore(ttl_seconds=3600)
        assert hasattr(store, "_lock")

    def test_zero_ttl_store(self):
        """Store with zero TTL should immediately expire entries."""
        store = ConcreteCredentialStore(ttl_seconds=0)
        with freeze_time("2024-01-01 00:00:00") as frozen:
            store.store("key1", "user", "pass")
            # Any forward progress past expires_at causes expiration
            frozen.tick(timedelta(microseconds=1))
            assert store.retrieve("key1") is None


class TestCredentialStorage:
    """Tests for credential storage operations."""

    def test_store_single_credential(self, store):
        """Should store a single credential."""
        store.store("key1", "user1", "pass1")
        result = store.retrieve("key1")
        assert result == ("user1", "pass1")

    def test_store_multiple_credentials(self, store):
        """Should store multiple credentials."""
        store.store("key1", "user1", "pass1")
        store.store("key2", "user2", "pass2")
        store.store("key3", "user3", "pass3")

        assert store.retrieve("key1") == ("user1", "pass1")
        assert store.retrieve("key2") == ("user2", "pass2")
        assert store.retrieve("key3") == ("user3", "pass3")

    def test_store_overwrites_existing(self, store):
        """Storing with same key should overwrite."""
        store.store("key1", "user1", "pass1")
        store.store("key1", "user2", "pass2")

        result = store.retrieve("key1")
        assert result == ("user2", "pass2")

    def test_store_with_empty_username(self, store):
        """Should handle empty username."""
        store.store("key1", "", "pass1")
        result = store.retrieve("key1")
        assert result == ("", "pass1")

    def test_store_with_empty_password(self, store):
        """Should handle empty password."""
        store.store("key1", "user1", "")
        result = store.retrieve("key1")
        assert result == ("user1", "")

    def test_store_with_unicode_credentials(self, store):
        """Should handle unicode credentials."""
        store.store("key1", "用户名", "密码")
        result = store.retrieve("key1")
        assert result == ("用户名", "密码")

    def test_store_with_special_characters(self, store):
        """Should handle special characters."""
        store.store("key1", "user@domain.com", "p@ss!word#123$")
        result = store.retrieve("key1")
        assert result == ("user@domain.com", "p@ss!word#123$")


class TestCredentialRetrieval:
    """Tests for credential retrieval operations."""

    def test_retrieve_nonexistent_key(self, store):
        """Should return None for nonexistent key."""
        assert store.retrieve("nonexistent") is None

    def test_retrieve_without_remove(self, store):
        """Retrieve without remove should preserve entry."""
        store.store("key1", "user1", "pass1")
        store.retrieve("key1", remove=False)
        # Should still be retrievable
        assert store.retrieve("key1") == ("user1", "pass1")

    def test_retrieve_with_remove(self, store):
        """Retrieve with remove should delete entry."""
        store.store("key1", "user1", "pass1")
        result = store.retrieve("key1", remove=True)
        assert result == ("user1", "pass1")
        # Should be gone now
        assert store.retrieve("key1") is None

    def test_retrieve_multiple_times(self, store):
        """Should be able to retrieve multiple times without remove."""
        store.store("key1", "user1", "pass1")

        for _ in range(10):
            result = store.retrieve("key1")
            assert result == ("user1", "pass1")


class TestTTLExpiration:
    """Tests for TTL expiration behavior."""

    def test_entry_expires_after_ttl(self, short_ttl_store):
        """Entry should expire after TTL.

        This test deliberately uses a real ``time.sleep`` instead of
        ``freeze_time`` so that the rest of the freezegun-based suite is
        validated against the actual ``time.time()`` clock at least once.
        Don't migrate this one without keeping another integration anchor.
        """
        short_ttl_store.store("key1", "user1", "pass1")
        time.sleep(1.5)  # allow: unmarked-sleep  # Wait for TTL + buffer
        assert short_ttl_store.retrieve("key1") is None

    def test_entry_valid_before_ttl(self, short_ttl_store):
        """Entry should be valid before TTL expires."""
        with freeze_time("2024-01-01 00:00:00") as frozen:
            short_ttl_store.store("key1", "user1", "pass1")
            frozen.tick(timedelta(milliseconds=500))  # Half of 1s TTL
            assert short_ttl_store.retrieve("key1") == ("user1", "pass1")

    def test_each_entry_has_own_ttl(self):
        """Each entry should have its own expiration time."""
        store = ConcreteCredentialStore(ttl_seconds=2)
        with freeze_time("2024-01-01 00:00:00") as frozen:
            store.store("key1", "user1", "pass1")
            frozen.tick(timedelta(seconds=1))
            store.store("key2", "user2", "pass2")  # Added 1s later
            frozen.tick(timedelta(milliseconds=1500))

            # key1 should be expired (2.5s old)
            # key2 should still be valid (1.5s old)
            assert store.retrieve("key1") is None
            assert store.retrieve("key2") == ("user2", "pass2")

    def test_overwrite_resets_ttl(self):
        """Overwriting an entry should reset its TTL."""
        store = ConcreteCredentialStore(ttl_seconds=1)
        with freeze_time("2024-01-01 00:00:00") as frozen:
            store.store("key1", "user1", "pass1")
            frozen.tick(timedelta(milliseconds=700))
            store.store("key1", "user1", "pass1")  # Reset TTL
            frozen.tick(timedelta(milliseconds=700))

            # Should still be valid (0.7s since reset, TTL is 1s)
            assert store.retrieve("key1") == ("user1", "pass1")


class TestCleanupExpired:
    """Tests for cleanup of expired entries."""

    def test_cleanup_removes_expired(self):
        """Cleanup should remove expired entries."""
        store = ConcreteCredentialStore(ttl_seconds=1)
        with freeze_time("2024-01-01 00:00:00") as frozen:
            store.store("key1", "user1", "pass1")
            store.store("key2", "user2", "pass2")
            frozen.tick(timedelta(milliseconds=1500))

            # Trigger cleanup by storing new entry
            store.store("key3", "user3", "pass3")

            # Old entries should be cleaned up
            assert store.retrieve("key1") is None
            assert store.retrieve("key2") is None
            assert store.retrieve("key3") == ("user3", "pass3")

    def test_cleanup_preserves_valid_entries(self, store):
        """Cleanup should preserve non-expired entries."""
        store.store("key1", "user1", "pass1")
        store.store("key2", "user2", "pass2")

        # Trigger cleanup
        store._cleanup_expired()

        # All should still be valid (TTL is 1 hour)
        assert store.retrieve("key1") == ("user1", "pass1")
        assert store.retrieve("key2") == ("user2", "pass2")


class TestClearEntry:
    """Tests for clear_entry method."""

    def test_clear_existing_entry(self, store):
        """Should clear an existing entry."""
        store.store("key1", "user1", "pass1")
        store.clear_entry("key1")
        assert store.retrieve("key1") is None

    def test_clear_nonexistent_entry(self, store):
        """Should handle clearing nonexistent entry."""
        # Should not raise
        store.clear_entry("nonexistent")

    def test_clear_does_not_affect_other_entries(self, store):
        """Clearing one entry should not affect others."""
        store.store("key1", "user1", "pass1")
        store.store("key2", "user2", "pass2")

        store.clear_entry("key1")

        assert store.retrieve("key1") is None
        assert store.retrieve("key2") == ("user2", "pass2")


class TestThreadSafety:
    """Tests for thread safety."""

    def test_concurrent_stores(self, store):
        """Concurrent stores should be thread-safe."""
        results = {"errors": []}

        def store_entry(key, username, password):
            try:
                store.store(key, username, password)
            except Exception as e:
                results["errors"].append(str(e))

        threads = [
            threading.Thread(
                target=store_entry, args=(f"key{i}", f"user{i}", f"pass{i}")
            )
            for i in range(100)
        ]

        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert len(results["errors"]) == 0

    def test_concurrent_retrieves(self, store):
        """Concurrent retrieves should be thread-safe."""
        store.store("shared_key", "user", "pass")
        results = []
        lock = threading.Lock()

        def retrieve_entry():
            result = store.retrieve("shared_key")
            with lock:
                results.append(result)

        threads = [threading.Thread(target=retrieve_entry) for _ in range(100)]

        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert all(r == ("user", "pass") for r in results)

    def test_concurrent_store_and_retrieve(self, store):
        """Concurrent stores and retrieves should be thread-safe."""
        results = {"errors": [], "retrievals": []}
        lock = threading.Lock()

        def store_entry():
            try:
                store.store("key1", "user", "pass")
            except Exception as e:
                with lock:
                    results["errors"].append(str(e))

        def retrieve_entry():
            try:
                result = store.retrieve("key1")
                with lock:
                    results["retrievals"].append(result)
            except Exception as e:
                with lock:
                    results["errors"].append(str(e))

        # First store, then concurrent operations
        store.store("key1", "user", "pass")

        threads = []
        for i in range(50):
            threads.append(threading.Thread(target=store_entry))
            threads.append(threading.Thread(target=retrieve_entry))

        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert len(results["errors"]) == 0

    def test_concurrent_clear_and_retrieve(self, store):
        """Concurrent clears and retrieves should be thread-safe."""
        errors = []
        lock = threading.Lock()

        def clear_and_retrieve():
            try:
                store.store("key", "user", "pass")
                store.clear_entry("key")
                store.retrieve("key")
            except Exception as e:
                with lock:
                    errors.append(str(e))

        threads = [
            threading.Thread(target=clear_and_retrieve) for _ in range(50)
        ]

        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert len(errors) == 0


class TestMemoryManagement:
    """Tests for memory management."""

    def test_many_entries_stored(self, store):
        """Should handle many entries."""
        for i in range(1000):
            store.store(f"key{i}", f"user{i}", f"pass{i}")

        # Spot check some entries
        assert store.retrieve("key0") == ("user0", "pass0")
        assert store.retrieve("key500") == ("user500", "pass500")
        assert store.retrieve("key999") == ("user999", "pass999")

    def test_entries_cleaned_up_over_time(self):
        """Old entries should be cleaned up."""
        store = ConcreteCredentialStore(ttl_seconds=1)
        with freeze_time("2024-01-01 00:00:00") as frozen:
            # Add many entries
            for i in range(100):
                store.store(f"key{i}", f"user{i}", f"pass{i}")

            frozen.tick(timedelta(milliseconds=1500))

            # Add new entry to trigger cleanup
            store.store("new_key", "new_user", "new_pass")

            # Old entries should be gone
            assert store.retrieve("key0") is None
            assert store.retrieve("new_key") == ("new_user", "new_pass")


class TestEdgeCases:
    """Tests for edge cases."""

    def test_very_long_key(self, store):
        """Should handle very long keys."""
        long_key = "k" * 10000
        store.store(long_key, "user", "pass")
        assert store.retrieve(long_key) == ("user", "pass")

    def test_very_long_credentials(self, store):
        """Should handle very long credentials."""
        long_username = "u" * 10000
        long_password = "p" * 10000
        store.store("key1", long_username, long_password)
        assert store.retrieve("key1") == (long_username, long_password)

    def test_empty_key(self, store):
        """Should handle empty key."""
        store.store("", "user", "pass")
        assert store.retrieve("") == ("user", "pass")

    def test_key_with_null_bytes(self, store):
        """Should handle keys with null bytes."""
        key = "key\x00with\x00nulls"
        store.store(key, "user", "pass")
        assert store.retrieve(key) == ("user", "pass")

    def test_credentials_with_newlines(self, store):
        """Should handle credentials with newlines."""
        store.store("key1", "user\nwith\nnewlines", "pass\nwith\nnewlines")
        assert store.retrieve("key1") == (
            "user\nwith\nnewlines",
            "pass\nwith\nnewlines",
        )

    def test_whitespace_key(self, store):
        """Should handle whitespace-only key."""
        store.store("   ", "user", "pass")
        assert store.retrieve("   ") == ("user", "pass")
        assert store.retrieve("") is None  # Different key


class TestAbstractMethods:
    """Tests for abstract method enforcement."""

    def test_cannot_instantiate_base_class(self):
        """Should not be able to instantiate abstract base class."""
        with pytest.raises(TypeError):
            CredentialStoreBase(ttl_seconds=3600)

    def test_must_implement_store(self):
        """Subclass must implement store method."""

        class IncompleteStore(CredentialStoreBase):
            def retrieve(self, key):
                pass

        with pytest.raises(TypeError):
            IncompleteStore(ttl_seconds=3600)

    def test_must_implement_retrieve(self):
        """Subclass must implement retrieve method."""

        class IncompleteStore(CredentialStoreBase):
            def store(self, key, username, password):
                pass

        with pytest.raises(TypeError):
            IncompleteStore(ttl_seconds=3600)