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

"""Tests for CredentialStoreBase TTL boundary conditions and remove parameter.

Complements test_credential_store_base.py and test_credential_store_extended.py
by testing precise TTL boundaries using mocked time.time().
"""

import time as time_module
from unittest.mock import patch


from local_deep_research.database.credential_store_base import (
    CredentialStoreBase,
)


class ConcreteStore(CredentialStoreBase):
    """Minimal concrete subclass for testing base class logic."""

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

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


# ---------------------------------------------------------------------------
# _retrieve_credentials boundary conditions
# ---------------------------------------------------------------------------


class TestRetrieveCredentialsBoundary:
    """TTL boundary tests using mocked time."""

    def test_not_expired_at_exact_boundary(self):
        """time.time() == expires_at → NOT expired (uses > not >=)."""
        store = ConcreteStore(ttl_seconds=100)
        with patch.object(time_module, "time", return_value=1000.0):
            store.store("k", "user", "pass")
            # expires_at = 1000.0 + 100 = 1100.0

        with patch.object(time_module, "time", return_value=1100.0):
            result = store.retrieve("k")
            assert result == ("user", "pass")

    def test_expired_just_past_boundary(self):
        """time.time() = expires_at + epsilon → expired."""
        store = ConcreteStore(ttl_seconds=100)
        with patch.object(time_module, "time", return_value=1000.0):
            store.store("k", "user", "pass")

        with patch.object(time_module, "time", return_value=1100.001):
            result = store.retrieve("k")
            assert result is None

    def test_expired_entry_deleted_from_store(self):
        """Expired retrieval deletes the entry."""
        store = ConcreteStore(ttl_seconds=10)
        with patch.object(time_module, "time", return_value=1000.0):
            store.store("k", "user", "pass")

        with patch.object(time_module, "time", return_value=1011.0):
            store.retrieve("k")  # triggers deletion
        assert "k" not in store._store


# ---------------------------------------------------------------------------
# _cleanup_expired
# ---------------------------------------------------------------------------


class TestCleanupExpired:
    """Tests for _cleanup_expired logic with mocked time."""

    def test_some_expired(self):
        store = ConcreteStore(ttl_seconds=100)
        with patch.object(time_module, "time", return_value=1000.0):
            store.store("old", "u1", "p1")
        # Insert "new" directly, bypassing _store_credentials' implicit cleanup
        with patch.object(time_module, "time", return_value=2000.0):
            store._store["new"] = {
                "username": "u2",
                "password": "p2",
                "expires_at": time_module.time() + store.ttl,  # 2100
            }
            store._cleanup_expired()
        assert "old" not in store._store
        assert "new" in store._store

    def test_all_expired(self):
        store = ConcreteStore(ttl_seconds=10)
        with patch.object(time_module, "time", return_value=1000.0):
            store.store("a", "u1", "p1")
            store.store("b", "u2", "p2")

        with patch.object(time_module, "time", return_value=2000.0):
            store._cleanup_expired()
        assert len(store._store) == 0

    def test_empty_store_no_error(self):
        store = ConcreteStore(ttl_seconds=3600)
        store._cleanup_expired()  # should not raise
        assert len(store._store) == 0