backend/tests/test_refresh_store.py

import os
import sys
import tempfile
from unittest.mock import MagicMock

os.environ.setdefault("CEPHEUS_CLOUD", "1")
os.environ.setdefault("CEPHEUS_API_KEY", "test-key")
os.environ["CEPHEUS_AUTH_DEV_MODE"] = "1"
os.environ.pop("REDIS_URL", None)

for mod in ("google.adk", "google.adk.agents", "google.adk.runners", "google.adk.sessions"):
    sys.modules.setdefault(mod, MagicMock())

BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BACKEND_DIR not in sys.path:
    sys.path.insert(0, BACKEND_DIR)

import refresh_token_store as rts  # noqa: E402
import auth_service  # noqa: E402


def test_refresh_token_survives_reinit(tmp_path, monkeypatch):
    monkeypatch.setattr(rts, "_DATA_DIR", str(tmp_path))
    monkeypatch.setattr(rts, "_REFRESH_FILE", str(tmp_path / "refresh_tokens.json"))
    auth_service._refresh_tokens.clear()

    pair = auth_service.create_token_pair("admin", "admin")
    jti_count = len(auth_service._refresh_tokens)
    assert jti_count == 1

    auth_service._refresh_tokens.clear()
    auth_service.init_refresh_store()
    assert len(auth_service._refresh_tokens) == 1

    refreshed = auth_service.refresh_access_token(pair["refresh_token"])
    assert refreshed.get("access_token")