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

"""Tests for credential store base class."""

from datetime import timedelta

from freezegun import freeze_time


class ConcreteCredentialStore:
    """Concrete implementation for testing."""

    def __init__(self, ttl_seconds: int):
        from local_deep_research.database.credential_store_base import (
            CredentialStoreBase,
        )

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

            def retrieve(self, key: str):
                return self._retrieve_credentials(key)

        self._impl = _Impl(ttl_seconds)

    def store(self, key: str, username: str, password: str):
        return self._impl.store(key, username, password)

    def retrieve(self, key: str):
        return self._impl.retrieve(key)

    def clear_entry(self, key: str):
        return self._impl.clear_entry(key)


class TestCredentialStoreBase:
    def test_init(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        assert store is not None

    def test_store_and_retrieve(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        store.store("key1", "user1", "pass1")
        result = store.retrieve("key1")
        assert result == ("user1", "pass1")

    def test_retrieve_nonexistent(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        result = store.retrieve("nonexistent")
        assert result is None

    def test_clear_entry(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        store.store("key1", "user1", "pass1")
        store.clear_entry("key1")
        result = store.retrieve("key1")
        assert result is None

    def test_clear_nonexistent_entry(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        # Should not raise
        store.clear_entry("nonexistent")

    def test_multiple_entries(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        store.store("key1", "user1", "pass1")
        store.store("key2", "user2", "pass2")

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

    def test_overwrite_entry(self):
        store = ConcreteCredentialStore(ttl_seconds=3600)
        store.store("key1", "user1", "pass1")
        store.store("key1", "user1_new", "pass1_new")

        result = store.retrieve("key1")
        assert result == ("user1_new", "pass1_new")

    def test_expired_entry_returns_none(self):
        store = ConcreteCredentialStore(ttl_seconds=1)  # 1 second TTL
        with freeze_time("2024-01-01 00:00:00") as frozen:
            store.store("key1", "user1", "pass1")

            # Wait for expiration
            frozen.tick(timedelta(milliseconds=1100))

            result = store.retrieve("key1")
            assert result is None