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

"""
Behavioral tests for credential_store_base and temp_auth modules.

Tests the credential storage with TTL expiration.
"""

from freezegun import freeze_time


class TestCredentialStoreBaseInit:
    """Tests for CredentialStoreBase initialization."""

    def test_temp_auth_store_initializes(self):
        """TemporaryAuthStore can be initialized."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        assert store is not None

    def test_temp_auth_store_custom_ttl(self):
        """TemporaryAuthStore accepts custom TTL."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore(ttl_seconds=60)
        assert store.ttl == 60

    def test_default_ttl_is_10_seconds(self):
        """Default TTL is 10 seconds."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        assert store.ttl == 10


class TestCredentialStoreAuth:
    """Tests for store_auth and retrieve_auth."""

    def test_store_auth_returns_token(self):
        """store_auth returns a token."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("testuser", "testpass")
        assert token is not None
        assert isinstance(token, str)
        assert len(token) > 0

    def test_store_auth_tokens_are_unique(self):
        """Each store_auth call returns a unique token."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token1 = store.store_auth("user1", "pass1")
        token2 = store.store_auth("user2", "pass2")
        assert token1 != token2

    def test_retrieve_auth_returns_credentials(self):
        """retrieve_auth returns stored credentials."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("testuser", "testpass")
        result = store.retrieve_auth(token)
        assert result is not None
        assert result[0] == "testuser"
        assert result[1] == "testpass"

    def test_retrieve_auth_removes_entry(self):
        """retrieve_auth removes the entry after retrieval."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("testuser", "testpass")
        store.retrieve_auth(token)
        # Second retrieval should return None
        result = store.retrieve_auth(token)
        assert result is None

    def test_retrieve_auth_returns_none_for_invalid_token(self):
        """retrieve_auth returns None for invalid token."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        result = store.retrieve_auth("invalid_token")
        assert result is None


class TestCredentialStorePeek:
    """Tests for peek_auth functionality."""

    def test_peek_auth_returns_credentials(self):
        """peek_auth returns stored credentials."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("testuser", "testpass")
        result = store.peek_auth(token)
        assert result is not None
        assert result[0] == "testuser"
        assert result[1] == "testpass"

    def test_peek_auth_does_not_remove_entry(self):
        """peek_auth does not remove the entry."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("testuser", "testpass")
        store.peek_auth(token)
        # Entry should still be there
        result = store.peek_auth(token)
        assert result is not None
        assert result[0] == "testuser"

    def test_peek_auth_returns_none_for_invalid_token(self):
        """peek_auth returns None for invalid token."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        result = store.peek_auth("invalid_token")
        assert result is None


class TestCredentialStoreTTL:
    """Tests for TTL expiration."""

    def test_credentials_expire_after_ttl(self):
        """Credentials expire after TTL."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        # SUT (credential_store_base) compares time.time() to
        # entry["expires_at"], so freezegun can mock the clock fully.
        with freeze_time("2026-01-01 00:00:00") as frozen:
            store = TemporaryAuthStore(ttl_seconds=1)
            token = store.store_auth("testuser", "testpass")
            # Advance past expiration
            frozen.tick(1.5)
            result = store.retrieve_auth(token)
            assert result is None

    def test_credentials_valid_before_ttl(self):
        """Credentials are valid before TTL."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore(ttl_seconds=60)
        token = store.store_auth("testuser", "testpass")
        result = store.retrieve_auth(token)
        assert result is not None


class TestCredentialStoreClearEntry:
    """Tests for clear_entry functionality."""

    def test_clear_entry_removes_credential(self):
        """clear_entry removes a specific credential."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("testuser", "testpass")
        store.clear_entry(token)
        result = store.retrieve_auth(token)
        assert result is None

    def test_clear_entry_only_affects_specified_token(self):
        """clear_entry only removes the specified token."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token1 = store.store_auth("user1", "pass1")
        token2 = store.store_auth("user2", "pass2")
        store.clear_entry(token1)
        # token2 should still be valid
        result = store.peek_auth(token2)
        assert result is not None
        assert result[0] == "user2"

    def test_clear_entry_handles_nonexistent_token(self):
        """clear_entry handles nonexistent token gracefully."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        # Should not raise
        store.clear_entry("nonexistent_token")


class TestCredentialStoreAliases:
    """Tests for store/retrieve aliases."""

    def test_store_alias_works(self):
        """store() method works as alias for store_auth()."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store("testuser", "testpass")
        assert token is not None

    def test_retrieve_alias_works(self):
        """retrieve() method works as alias for retrieve_auth()."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store("testuser", "testpass")
        result = store.retrieve(token)
        assert result is not None
        assert result[0] == "testuser"


class TestCredentialStoreMultipleEntries:
    """Tests for multiple credential entries."""

    def test_multiple_users_stored_separately(self):
        """Multiple users are stored separately."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token1 = store.store_auth("user1", "pass1")
        token2 = store.store_auth("user2", "pass2")
        token3 = store.store_auth("user3", "pass3")

        result1 = store.peek_auth(token1)
        result2 = store.peek_auth(token2)
        result3 = store.peek_auth(token3)

        assert result1[0] == "user1"
        assert result2[0] == "user2"
        assert result3[0] == "user3"

    def test_same_user_can_have_multiple_tokens(self):
        """Same user can have multiple tokens."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token1 = store.store_auth("sameuser", "pass1")
        token2 = store.store_auth("sameuser", "pass2")

        assert token1 != token2
        result1 = store.peek_auth(token1)
        result2 = store.peek_auth(token2)
        assert result1 is not None
        assert result2 is not None


class TestCredentialStoreThreadSafety:
    """Tests for thread-safety features."""

    def test_store_has_lock(self):
        """Store has a lock for thread safety."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        assert hasattr(store, "_lock")

    def test_concurrent_stores_work(self):
        """Concurrent stores work correctly."""
        import threading
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        tokens = []
        errors = []

        def store_auth():
            try:
                token = store.store_auth(
                    f"user_{threading.current_thread().name}", "pass"
                )
                tokens.append(token)
            except Exception as e:
                errors.append(e)

        threads = [threading.Thread(target=store_auth) for _ in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert len(errors) == 0
        assert len(tokens) == 10
        assert len(set(tokens)) == 10  # All unique


class TestGlobalTempAuthStore:
    """Tests for the global temp_auth_store instance."""

    def test_global_instance_is_temporary_auth_store(self):
        """Global instance is TemporaryAuthStore."""
        from local_deep_research.database.temp_auth import (
            TemporaryAuthStore,
            temp_auth_store,
        )

        assert isinstance(temp_auth_store, TemporaryAuthStore)


class TestCredentialStoreCleanup:
    """Tests for automatic cleanup of expired entries."""

    def test_expired_entries_cleaned_on_store(self):
        """Expired entries are cleaned when storing new credentials."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        with freeze_time("2026-01-01 00:00:00") as frozen:
            store = TemporaryAuthStore(ttl_seconds=1)
            token1 = store.store_auth("user1", "pass1")
            frozen.tick(1.5)  # Advance past expiration
            # Storing new entry should trigger cleanup
            store.store_auth("user2", "pass2")
            # token1 should be expired and cleaned
            result = store.peek_auth(token1)
            assert result is None


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

    def test_empty_username(self):
        """Handles empty username."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("", "password")
        result = store.retrieve_auth(token)
        assert result is not None
        assert result[0] == ""

    def test_empty_password(self):
        """Handles empty password."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        token = store.store_auth("username", "")
        result = store.retrieve_auth(token)
        assert result is not None
        assert result[1] == ""

    def test_special_characters_in_credentials(self):
        """Handles special characters in credentials."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        special_user = "user@example.com"
        special_pass = "p@ss!w0rd#$%"
        token = store.store_auth(special_user, special_pass)
        result = store.retrieve_auth(token)
        assert result is not None
        assert result[0] == special_user
        assert result[1] == special_pass

    def test_unicode_in_credentials(self):
        """Handles unicode in credentials."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        unicode_user = "用户名"
        unicode_pass = "密码🔐"
        token = store.store_auth(unicode_user, unicode_pass)
        result = store.retrieve_auth(token)
        assert result is not None
        assert result[0] == unicode_user
        assert result[1] == unicode_pass

    def test_long_credentials(self):
        """Handles long credentials."""
        from local_deep_research.database.temp_auth import TemporaryAuthStore

        store = TemporaryAuthStore()
        long_user = "a" * 1000
        long_pass = "b" * 1000
        token = store.store_auth(long_user, long_pass)
        result = store.retrieve_auth(token)
        assert result is not None
        assert result[0] == long_user
        assert result[1] == long_pass