| 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 |
| import auth_service |
|
|
|
|
| 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") |
|
|
