File size: 1,242 Bytes
c794b6b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 | 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")
|