diff --git a/CHANGELOG.md b/CHANGELOG.md index d7f29e3..0c18677 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,26 @@ # Changelog +## 1.11.2 — 2026-05-23 + +### Fixed + +- **Cross-process JWT cache.** The in-memory `_token` cache previously survived only for the lifetime of a `ColonyClient` instance — short-lived scripts and processes that recreate a client per invocation re-authenticated against `/auth/token` every time, which the server rate-limits per-IP. The SDK now persists the access token to disk so a new process for the same `(base_url, api_key)` pair reuses the cached token instead of round-tripping. + + Cache location is platform-aware: + + - **Linux / BSD / Unix**: `$XDG_CACHE_HOME/colony-sdk/` or `~/.cache/colony-sdk/` + - **macOS**: `~/Library/Caches/colony-sdk/` + - **Windows**: `%LOCALAPPDATA%\colony-sdk\Cache\` (falls back to `%APPDATA%`) + - Always overridable via `COLONY_SDK_TOKEN_CACHE_DIR` + + Filename is `.json` so the same api_key against prod vs staging gets independent cache files. Cache writes are atomic (tmpfile + rename) and mode-0600 so a co-tenant on the same host cannot read another user's token. A 60-second safety margin avoids handing out a token that's about to expire mid-request. + + Opt-out: per-client via `ColonyClient(..., cache_token=False)`, or globally via `COLONY_SDK_NO_TOKEN_CACHE=1`. + + Reads and writes are best-effort — any IO error (un-writable cache dir, corrupt cache file, disk full) silently falls through to a fresh `/auth/token` call, so cache correctness is never load-bearing on the request path. `refresh_token()`, `rotate_key()`, and the auto-401-refresh path all invalidate the on-disk cache so a stale token cannot resurrect across processes. Mirrored in `AsyncColonyClient` (shared cache file format and location for the same `(base_url, api_key)` pair). + + Regression coverage in `test_client.py::TestTokenCachePersistence` and `test_async_client.py::TestAsyncTokenCachePersistence`. A new `tests/conftest.py` autouse fixture routes the cache to a per-test `tmp_path` so existing tests don't leak token files into the developer's real cache dir. + ## 1.11.0 — 2026-05-18 ### New methods diff --git a/pyproject.toml b/pyproject.toml index bc22b4b..387145c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "colony-sdk" -version = "1.11.1" +version = "1.11.2" description = "Python SDK for The Colony (thecolony.cc) — the official Python client for the AI agent internet" readme = "README.md" license = {text = "MIT"} diff --git a/src/colony_sdk/__init__.py b/src/colony_sdk/__init__.py index f22f17e..19b6d5a 100644 --- a/src/colony_sdk/__init__.py +++ b/src/colony_sdk/__init__.py @@ -61,7 +61,7 @@ async def main(): from colony_sdk.async_client import AsyncColonyClient from colony_sdk.testing import MockColonyClient -__version__ = "1.11.1" +__version__ = "1.11.2" __all__ = [ "COLONIES", "AsyncColonyClient", diff --git a/src/colony_sdk/async_client.py b/src/colony_sdk/async_client.py index 03cb02e..a40f85b 100644 --- a/src/colony_sdk/async_client.py +++ b/src/colony_sdk/async_client.py @@ -32,6 +32,7 @@ async def main(): import asyncio import json from collections.abc import AsyncIterator +from pathlib import Path from types import TracebackType from typing import Any @@ -87,12 +88,21 @@ def __init__( client: httpx.AsyncClient | None = None, retry: RetryConfig | None = None, typed: bool = False, + cache_token: bool = True, ): self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self.retry = retry if retry is not None else RetryConfig() self.typed = typed + # `cache_token=True` (default) persists the JWT to a + # platform-specific cache directory (see + # :func:`colony_sdk.client._token_cache_dir` for resolution + # order on Linux / macOS / Windows). Shared cache file with the + # sync `ColonyClient` for the same (base_url, api_key) pair. + # Disable per-client by passing False, or globally with + # `COLONY_SDK_NO_TOKEN_CACHE=1`. + self.cache_token = cache_token self._token: str | None = None self._token_expiry: float = 0 self._client = client @@ -191,11 +201,98 @@ def _get_client(self) -> httpx.AsyncClient: # ── Auth ────────────────────────────────────────────────────────── + def _token_cache_enabled(self) -> bool: + """True if the on-disk JWT cache is active for this client. Mirrors sync.""" + from colony_sdk.client import _token_cache_disabled_via_env + + if not self.cache_token: + return False + return not _token_cache_disabled_via_env() + + def _cached_token_path(self) -> Path: + from colony_sdk.client import _token_cache_path + + return _token_cache_path(self.api_key, self.base_url) + + def _load_cached_token(self) -> bool: + """Hydrate `self._token` from the on-disk cache if a valid one exists. + + Identical contract to the sync version — see + :meth:`ColonyClient._load_cached_token`. Shared cache file so a + token written by the sync client is readable by the async client + and vice versa. + """ + import time + + from colony_sdk.client import _TOKEN_CACHE_SAFETY_MARGIN_SEC + + if not self._token_cache_enabled(): + return False + try: + path = self._cached_token_path() + if not path.exists(): + return False + with path.open("r", encoding="utf-8") as f: + data = json.load(f) + token = data.get("token") + expiry = float(data.get("expiry", 0)) + except (OSError, ValueError, TypeError, json.JSONDecodeError): + return False + if not token or expiry <= time.time() + _TOKEN_CACHE_SAFETY_MARGIN_SEC: + return False + self._token = token + self._token_expiry = expiry + return True + + def _save_cached_token(self) -> None: + """Best-effort write of the current JWT + expiry to disk.""" + import contextlib + import os + + from colony_sdk.client import _TOKEN_CACHE_SCHEMA_VERSION + + if not self._token_cache_enabled() or not self._token: + return + try: + path = self._cached_token_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump( + { + "v": _TOKEN_CACHE_SCHEMA_VERSION, + "token": self._token, + "expiry": self._token_expiry, + }, + f, + ) + except Exception: + with contextlib.suppress(OSError): + os.unlink(str(tmp)) + raise + os.replace(str(tmp), str(path)) + except OSError: + pass + + def _clear_cached_token(self) -> None: + """Remove the on-disk cache entry. Silent on failure.""" + import contextlib + + if not self._token_cache_enabled(): + return + with contextlib.suppress(OSError): + self._cached_token_path().unlink(missing_ok=True) + async def _ensure_token(self) -> None: import time if self._token and time.time() < self._token_expiry: return + # See ColonyClient._ensure_token for the cache-first rationale. + if self._load_cached_token(): + return data = await self._raw_request( "POST", "/auth/token", @@ -205,11 +302,17 @@ async def _ensure_token(self) -> None: self._token = data["access_token"] # Refresh 1 hour before expiry (tokens last 24h) self._token_expiry = time.time() + 23 * 3600 + self._save_cached_token() def refresh_token(self) -> None: - """Force a token refresh on the next request.""" + """Force a token refresh on the next request. + + Clears both the in-memory token and the on-disk cache entry + (if enabled), matching :meth:`ColonyClient.refresh_token`. + """ self._token = None self._token_expiry = 0 + self._clear_cached_token() async def rotate_key(self) -> dict: """Rotate your API key. Returns the new key and invalidates the old one. @@ -219,6 +322,9 @@ async def rotate_key(self) -> dict: """ data = await self._raw_request("POST", "/auth/rotate-key") if "api_key" in data: + # Clear the old key's on-disk cache entry BEFORE flipping + # `self.api_key` — same ordering rule as ColonyClient.rotate_key. + self._clear_cached_token() self.api_key = data["api_key"] self._token = None self._token_expiry = 0 @@ -300,6 +406,8 @@ async def _raw_request( # Auto-refresh on 401 once (separate from the configurable retry loop). if resp.status_code == 401 and not _token_refreshed and auth: + # Invalidate the disk cache too — the cached token is stale. + self._clear_cached_token() self._token = None self._token_expiry = 0 return await self._raw_request(method, path, body, auth, _retry=_retry, _token_refreshed=True) diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py index 1919128..f47f412 100644 --- a/src/colony_sdk/client.py +++ b/src/colony_sdk/client.py @@ -13,10 +13,13 @@ import hmac import json import logging +import os import re +import sys import time from collections.abc import Iterator from dataclasses import dataclass, field +from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urlencode @@ -189,6 +192,90 @@ class RetryConfig: _DEFAULT_AUTH_RETRY = RetryConfig(max_retries=6, base_delay=2.0, max_delay=60.0) +# ── On-disk JWT cache ──────────────────────────────────────────────────── +# +# The in-memory `_token` cache on `ColonyClient` survives only for the +# lifetime of the client instance. Short-lived scripts and any setup +# that constructs a fresh client per invocation pay for a `/auth/token` +# round-trip on every start — and the server rate-limits that endpoint +# per-IP, so heavy reconstruction can exhaust the budget before doing +# any real work. +# +# This file-backed cache survives across processes for the same +# (base_url, api_key) pair. The on-disk format is a small JSON envelope +# with the token, its expiry, and a schema version. Reads and writes are +# best-effort: any IO error silently falls through to a fresh fetch, so +# correctness never depends on the cache being present, readable, or +# writable. The cache file is written mode-0600 so a co-tenant on the +# same machine cannot read another user's token. + +_TOKEN_CACHE_SCHEMA_VERSION = 1 +_TOKEN_CACHE_SAFETY_MARGIN_SEC = 60.0 + + +def _token_cache_dir() -> Path: + """Resolve the JWT cache directory for the current platform. + + Resolution order: + + 1. ``COLONY_SDK_TOKEN_CACHE_DIR`` if set (tests + power users override). + 2. Platform default: + + - **Linux / BSD / other Unix**: ``$XDG_CACHE_HOME/colony-sdk`` if + set, otherwise ``~/.cache/colony-sdk`` (XDG Base Directory). + - **macOS**: ``~/Library/Caches/colony-sdk`` (Apple's File System + Programming Guide). + - **Windows**: ``%LOCALAPPDATA%/colony-sdk/Cache``, falling back + to ``%APPDATA%/colony-sdk/Cache``, and finally to + ``~/AppData/Local/colony-sdk/Cache`` if neither is set. + + If the chosen path can't be created or written at use time, the + caller silently falls through to a fresh `/auth/token` request, so + cache resolution never errors at this layer. + """ + override = os.environ.get("COLONY_SDK_TOKEN_CACHE_DIR") + if override: + return Path(override) + if sys.platform == "win32": + # Prefer LOCALAPPDATA (machine-local, not roamed) over APPDATA + # so a per-machine cache isn't synced to other machines via + # roaming profiles. + for env_var in ("LOCALAPPDATA", "APPDATA"): + base = os.environ.get(env_var) + if base: + return Path(base) / "colony-sdk" / "Cache" + return Path.home() / "AppData" / "Local" / "colony-sdk" / "Cache" + if sys.platform == "darwin": + return Path.home() / "Library" / "Caches" / "colony-sdk" + # Linux / BSD / other Unix. + xdg = os.environ.get("XDG_CACHE_HOME") + if xdg: + return Path(xdg) / "colony-sdk" + return Path.home() / ".cache" / "colony-sdk" + + +def _token_cache_path(api_key: str, base_url: str) -> Path: + """Compute the cache filename for a given (api_key, base_url) pair. + + Hashes both together so the same api_key used against multiple bases + (e.g., prod vs staging) gets independent cache files. 16 hex chars + = 64 bits — more than enough to avoid collisions for any realistic + number of (key, base) pairs on one host. + """ + fingerprint = f"{base_url}|{api_key}".encode() + digest = hashlib.sha256(fingerprint).hexdigest()[:16] + return _token_cache_dir() / f"{digest}.json" + + +def _token_cache_disabled_via_env() -> bool: + """Global opt-out via env var. Recognised values: 1/true/yes (case-insensitive).""" + return os.environ.get("COLONY_SDK_NO_TOKEN_CACHE", "").strip().lower() in ( + "1", + "true", + "yes", + ) + + def _should_retry(status: int, attempt: int, retry: RetryConfig) -> bool: """Return True if a request that returned ``status`` should be retried. @@ -434,6 +521,7 @@ def __init__( typed: bool = False, proxy: str | None = None, auth_token_retry: RetryConfig | None = None, + cache_token: bool = True, ): self.api_key = api_key self.base_url = base_url.rstrip("/") @@ -447,6 +535,16 @@ def __init__( self.auth_token_retry = auth_token_retry if auth_token_retry is not None else _DEFAULT_AUTH_RETRY self.typed = typed self.proxy = proxy + # `cache_token=True` (default) persists the JWT to a + # platform-specific cache directory (XDG on Linux, + # ~/Library/Caches on macOS, %LOCALAPPDATA% on Windows; see + # :func:`_token_cache_dir`) so it survives process restarts + # for the same (base_url, api_key) pair. Set to False to + # disable per-client. Global opt-out via the + # `COLONY_SDK_NO_TOKEN_CACHE=1` env var. The cache file is + # written mode-0600 and reads/writes are best-effort: any IO + # error silently falls through to a fresh `/auth/token` call. + self.cache_token = cache_token self._token: str | None = None self._token_expiry: float = 0 self.last_rate_limit: RateLimitInfo | None = None @@ -539,9 +637,108 @@ def clear_cache(self) -> None: # ── Auth ────────────────────────────────────────────────────────── + def _token_cache_enabled(self) -> bool: + """True if the on-disk JWT cache is active for this client. + + Both the per-client `cache_token` constructor arg and the global + `COLONY_SDK_NO_TOKEN_CACHE` env var must allow caching. The env + var takes precedence so operators can disable globally without + touching application code. + """ + if not self.cache_token: + return False + return not _token_cache_disabled_via_env() + + def _cached_token_path(self) -> Path: + """Path to this client's on-disk JWT cache file.""" + return _token_cache_path(self.api_key, self.base_url) + + def _load_cached_token(self) -> bool: + """Hydrate `self._token` from the on-disk cache if a valid one exists. + + Returns True on cache hit (token loaded), False on miss or any + read failure. Cache hits are validated against a 60-second + safety margin so a token about to expire mid-request still + triggers a refresh rather than getting handed out at the edge. + """ + if not self._token_cache_enabled(): + return False + try: + path = self._cached_token_path() + if not path.exists(): + return False + with path.open("r", encoding="utf-8") as f: + data = json.load(f) + token = data.get("token") + expiry = float(data.get("expiry", 0)) + except (OSError, ValueError, TypeError, json.JSONDecodeError): + # Corrupt file, missing field, permission denied — any IO or + # parse failure is a cache miss, never an error to the caller. + return False + if not token or expiry <= time.time() + _TOKEN_CACHE_SAFETY_MARGIN_SEC: + return False + self._token = token + self._token_expiry = expiry + return True + + def _save_cached_token(self) -> None: + """Best-effort write of the current JWT + expiry to disk. + + Writes are atomic (tmpfile + rename) and mode-0600. Any failure + is silently swallowed — the cache is a cold-start latency + optimization, not a correctness requirement. + """ + import contextlib + + if not self._token_cache_enabled() or not self._token: + return + try: + path = self._cached_token_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + # Open with 0600 from the start so the secret is never on + # disk with a wider mode (umask can otherwise widen the + # initial mode and the chmod-after-write window leaks). + fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump( + { + "v": _TOKEN_CACHE_SCHEMA_VERSION, + "token": self._token, + "expiry": self._token_expiry, + }, + f, + ) + except Exception: + # Tmp file partially written — best-effort cleanup, then + # re-raise into the outer except where the whole save + # operation is swallowed. + with contextlib.suppress(OSError): + os.unlink(str(tmp)) + raise + os.replace(str(tmp), str(path)) + except OSError: + pass + + def _clear_cached_token(self) -> None: + """Remove the on-disk cache entry. Silent on failure.""" + import contextlib + + if not self._token_cache_enabled(): + return + with contextlib.suppress(OSError): + self._cached_token_path().unlink(missing_ok=True) + def _ensure_token(self) -> None: if self._token and time.time() < self._token_expiry: return + # Try the on-disk cache before paying for a fresh /auth/token + # call. Cache is keyed by (base_url, api_key) so it survives + # process restarts and short-lived scripts that would otherwise + # re-authenticate on every invocation. + if self._load_cached_token(): + return # Use the more aggressive `auth_token_retry` config for the # /auth/token request specifically — see `_DEFAULT_AUTH_RETRY` # for budget rationale. This is the only call site that uses @@ -556,11 +753,20 @@ def _ensure_token(self) -> None: self._token = data["access_token"] # Refresh 1 hour before expiry (tokens last 24h) self._token_expiry = time.time() + 23 * 3600 + # Persist to disk so the next process for this (base_url, + # api_key) pair can skip /auth/token entirely. + self._save_cached_token() def refresh_token(self) -> None: - """Force a token refresh on the next request.""" + """Force a token refresh on the next request. + + Clears both the in-memory token and the on-disk cache entry + (if enabled) so the next call will hit `/auth/token` and write + a fresh value back. + """ self._token = None self._token_expiry = 0 + self._clear_cached_token() def rotate_key(self) -> dict: """Rotate your API key. Returns the new key and invalidates the old one. @@ -573,6 +779,10 @@ def rotate_key(self) -> dict: """ data = self._raw_request("POST", "/auth/rotate-key") if "api_key" in data: + # Clear the old key's on-disk cache entry BEFORE flipping + # `self.api_key` — otherwise `_clear_cached_token()` would + # compute the path for the new key and miss the stale file. + self._clear_cached_token() self.api_key = data["api_key"] # Force token refresh since the old key is now invalid self._token = None @@ -668,6 +878,11 @@ def _raw_request( # Auto-refresh on 401 once (separate from the configurable retry loop). if e.code == 401 and not _token_refreshed and auth: + # The token (whether in-memory or from the on-disk + # cache) was rejected. Invalidate the disk cache too, + # otherwise the next process load would re-hydrate the + # same stale token and immediately 401 again. + self._clear_cached_token() self._token = None self._token_expiry = 0 return self._raw_request( diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..90f24bf --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,35 @@ +"""Test-suite-wide fixtures. + +We default-isolate the JWT cache directory so individual tests don't +write tokens to (or read tokens from) the developer's real +``~/.cache/colony-sdk/`` location, and so cache files written by one +test cannot leak into another. Tests that need to assert specific +cache-file presence (e.g., :class:`TestTokenCachePersistence`) override +the env var via ``monkeypatch.setenv`` per test, which takes precedence. +""" + +from __future__ import annotations + +import pytest + + +@pytest.fixture(autouse=True) +def _isolate_token_cache(tmp_path, monkeypatch): + """Route the JWT cache to a per-test tmp directory by default. + + Why autouse: many tests construct ``ColonyClient`` and trigger + ``_ensure_token``, which writes a cache file. Without isolation, + those writes would land in the developer's real cache dir, where + they could leak between tests in the same suite run and across + invocations of the test suite. Per-test tmp dir solves both. + + Tests in :class:`TestTokenCachePersistence` override + ``COLONY_SDK_TOKEN_CACHE_DIR`` themselves to assert specific paths; + that monkeypatch.setenv call takes precedence over this fixture. + """ + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path / "colony-sdk-cache")) + # Defensive: also clear the global kill-switch so a stale env var + # from the developer's shell doesn't silently disable caching for + # tests that depend on it. + monkeypatch.delenv("COLONY_SDK_NO_TOKEN_CACHE", raising=False) + yield diff --git a/tests/test_async_client.py b/tests/test_async_client.py index b426617..9f44a83 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -165,6 +165,235 @@ def handler(request: httpx.Request) -> httpx.Response: assert len(me_paths) == 2 +class TestAsyncTokenCachePersistence: + """The async client persists the JWT to disk the same way the sync + client does, and the two share the cache file for matching + `(base_url, api_key)` pairs. These tests mirror the sync coverage in + `test_client.py::TestTokenCachePersistence` — sync logic is tested + in depth there; here we verify the async paths are wired up correctly. + """ + + async def test_first_async_client_writes_to_cache(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + token_calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal token_calls + if request.url.path.endswith("/auth/token"): + token_calls += 1 + return _json_response({"access_token": "jwt-async-persisted"}) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_a", client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as client: + await client.get_me() + cached = list(tmp_path.glob("*.json")) + assert len(cached) == 1 + assert token_calls == 1 + # Sync client with the same key should read this file and skip auth. + + async def test_second_async_client_reads_from_cache(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + token_calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal token_calls + if request.url.path.endswith("/auth/token"): + token_calls += 1 + return _json_response({"access_token": "jwt-once"}) + return _json_response({"ok": True}) + + # First client writes the cache. + async with AsyncColonyClient( + "col_b", client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as client_a: + await client_a.get_me() + # Second client must not hit /auth/token again. + async with AsyncColonyClient( + "col_b", client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as client_b: + await client_b.get_me() + assert token_calls == 1 + assert client_b._token == "jwt-once" + + async def test_cache_token_false_disables_disk_writes(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/auth/token"): + return _json_response({"access_token": "jwt-no-write"}) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_c", + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + cache_token=False, + ) as client: + await client.get_me() + assert list(tmp_path.glob("*.json")) == [] + + async def test_env_var_disables_async_cache(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + monkeypatch.setenv("COLONY_SDK_NO_TOKEN_CACHE", "1") + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/auth/token"): + return _json_response({"access_token": "jwt-env-off"}) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_d", client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as client: + await client.get_me() + assert list(tmp_path.glob("*.json")) == [] + + async def test_async_refresh_token_clears_disk_cache(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/auth/token"): + return _json_response({"access_token": "jwt-init"}) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_e", client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as client: + await client.get_me() + assert len(list(tmp_path.glob("*.json"))) == 1 + client.refresh_token() + assert list(tmp_path.glob("*.json")) == [] + + async def test_async_corrupt_cache_falls_through(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + # Pre-seed garbage at the expected cache path. + from colony_sdk.client import _token_cache_path + + bad_path = _token_cache_path("col_corrupt", "https://thecolony.cc/api/v1") + bad_path.parent.mkdir(parents=True, exist_ok=True) + bad_path.write_text("{not valid json") + + token_calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal token_calls + if request.url.path.endswith("/auth/token"): + token_calls += 1 + return _json_response({"access_token": "jwt-after-corrupt"}) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_corrupt", + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) as client: + await client.get_me() # MUST NOT raise + assert token_calls == 1 + assert client._token == "jwt-after-corrupt" + + async def test_async_expired_cache_triggers_fresh_auth(self, monkeypatch, tmp_path) -> None: + import time + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + from colony_sdk.client import _token_cache_path + + stale_path = _token_cache_path("col_expired", "https://thecolony.cc/api/v1") + stale_path.parent.mkdir(parents=True, exist_ok=True) + stale_path.write_text(json.dumps({"v": 1, "token": "jwt-stale", "expiry": time.time() - 1})) + + token_calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal token_calls + if request.url.path.endswith("/auth/token"): + token_calls += 1 + return _json_response({"access_token": "jwt-fresh-async"}) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_expired", + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) as client: + await client.get_me() + assert token_calls == 1 + assert client._token == "jwt-fresh-async" + + async def test_async_save_swallows_mid_write_oserror(self, monkeypatch, tmp_path) -> None: + """OSError mid json.dump in the async path is swallowed — same + contract as the sync client.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + import colony_sdk.async_client as _async_mod + + def _exploding_dump(obj, fp, **kwargs): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(_async_mod.json, "dump", _exploding_dump) + + client = AsyncColonyClient("col_async_fail") + client._token = "jwt_async_fail" + client._token_expiry = 9999999999 + client._save_cached_token() # MUST NOT raise + assert list(tmp_path.glob("*")) == [] + + async def test_async_save_swallows_outer_oserror(self, monkeypatch, tmp_path) -> None: + monkeypatch.setenv( + "COLONY_SDK_TOKEN_CACHE_DIR", + "/proc/1/root/cache-cannot-write-here", + ) + client = AsyncColonyClient("col_async_unwritable") + client._token = "jwt" + client._token_expiry = 9999999999 + client._save_cached_token() # MUST NOT raise + + async def test_async_clear_no_op_when_disabled(self, monkeypatch, tmp_path) -> None: + from colony_sdk.client import _token_cache_path + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + path = _token_cache_path("col_async_disabled", "https://thecolony.cc/api/v1") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text('{"v":1,"token":"untouched","expiry":9999999999}') + monkeypatch.setenv("COLONY_SDK_NO_TOKEN_CACHE", "1") + client = AsyncColonyClient("col_async_disabled") + client._clear_cached_token() + assert path.exists() + + async def test_async_401_invalidates_disk_cache(self, monkeypatch, tmp_path) -> None: + import time + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + from colony_sdk.client import _token_cache_path + + stale_path = _token_cache_path("col_revoked", "https://thecolony.cc/api/v1") + stale_path.parent.mkdir(parents=True, exist_ok=True) + stale_path.write_text(json.dumps({"v": 1, "token": "jwt-server-revoked", "expiry": time.time() + 86400})) + + # First /users/me with stale token returns 401, then /auth/token, + # then /users/me retry succeeds. + token_calls = 0 + me_calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal token_calls, me_calls + if request.url.path.endswith("/auth/token"): + token_calls += 1 + return _json_response({"access_token": "jwt-new-async"}) + me_calls += 1 + if me_calls == 1: + return _json_response({"detail": "stale"}, status=401) + return _json_response({"id": "u1"}) + + async with AsyncColonyClient( + "col_revoked", + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) as client: + result = await client.get_me() + assert result == {"id": "u1"} + # Cache rewritten with the new token. + cached_files = list(tmp_path.glob("*.json")) + assert len(cached_files) == 1 + cached = json.loads(next(iter(cached_files)).read_text()) + assert cached["token"] == "jwt-new-async" + + # --------------------------------------------------------------------------- # Read methods # --------------------------------------------------------------------------- diff --git a/tests/test_client.py b/tests/test_client.py index 7a1e015..63e26f9 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -599,3 +599,460 @@ def test_url_error_on_auth_token_also_retries(self, monkeypatch): raise AssertionError("expected ColonyNetworkError") except Exception as e: assert "network error" in str(e).lower() + + +class TestTokenCachePersistence: + """The JWT is persisted to disk by default so it survives process + restarts. Cross-process cache, keyed by (base_url, api_key) — the + primary win is for supervisor-rotated dogfood agents that restart + every ~20min and would otherwise re-auth every cycle, eventually + tripping the 100/hr/IP `/auth/token` rate limit. + + Tests route the cache to a temp dir via ``COLONY_SDK_TOKEN_CACHE_DIR`` + so they never touch the real ``~/.cache/colony-sdk/`` location. + """ + + def _patch(self, monkeypatch, responses): + """Same mock-urlopen shape as TestAuthTokenRetry — duplicated here + because we want focused tests on the cache layer without inheriting + an unrelated test fixture.""" + import json as _json + from io import BytesIO + from urllib.error import HTTPError + + calls = [] + iter_responses = iter(responses) + + class _FakeResponse: + def __init__(self, status, body_bytes): + self.status = status + self._body = body_bytes + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return self._body + + def getheaders(self): + return [] + + def _fake_urlopen(req, timeout=None): + calls.append({"url": req.full_url, "method": req.get_method()}) + kind, *rest = next(iter_responses) + if kind == "ok": + status, body = rest + return _FakeResponse(status, _json.dumps(body).encode()) + if kind == "http_error": + status, body = rest + body_bytes = body.encode() if isinstance(body, str) else body + raise HTTPError(req.full_url, status, "fake", {}, BytesIO(body_bytes)) + raise AssertionError(f"unknown response kind: {kind}") + + monkeypatch.setattr("colony_sdk.client.urlopen", _fake_urlopen) + monkeypatch.setattr("colony_sdk.client.time.sleep", lambda _: None) + return calls + + def test_first_client_writes_token_to_disk(self, monkeypatch, tmp_path): + """After a fresh `_ensure_token` call, the cache file exists, + contains the token, and is mode 0600.""" + import stat + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_first", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test") + c.get_me() + cached_files = list(tmp_path.glob("*.json")) + assert len(cached_files) == 1 + # File must be 0600 — protects the secret on shared hosts. + mode = cached_files[0].stat().st_mode + assert stat.S_IMODE(mode) == 0o600, f"expected 0600, got {oct(stat.S_IMODE(mode))}" + import json as _json + + data = _json.loads(cached_files[0].read_text()) + assert data["token"] == "jwt_first" + assert data["v"] == 1 + assert data["expiry"] > 0 + + def test_second_client_loads_token_from_disk(self, monkeypatch, tmp_path): + """A second `ColonyClient(api_key)` with the same key sees the + cache file and skips `/auth/token` entirely.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + # First client: writes cache + calls_a = self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_persisted", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + a = ColonyClient("col_test") + a.get_me() + first_auth_calls = sum(1 for x in calls_a if x["url"].endswith("/auth/token")) + assert first_auth_calls == 1 + + # Second client: should NOT hit /auth/token at all. + calls_b = self._patch( + monkeypatch, + [ + ("ok", 200, {"username": "colonist-one"}), # just /users/me, NO /auth/token + ], + ) + b = ColonyClient("col_test") + b.get_me() + second_auth_calls = sum(1 for x in calls_b if x["url"].endswith("/auth/token")) + assert second_auth_calls == 0 + assert b._token == "jwt_persisted" + + def test_expired_cached_token_triggers_fresh_auth(self, monkeypatch, tmp_path): + """If the cached token's expiry is in the past, the SDK ignores + the cache and fetches a fresh token (and overwrites the cache).""" + import json as _json + import time as _time + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + + # Pre-seed a stale cache file directly (don't go through the SDK) + from colony_sdk.client import _token_cache_path + + stale_path = _token_cache_path("col_test", "https://thecolony.cc/api/v1") + stale_path.parent.mkdir(parents=True, exist_ok=True) + stale_path.write_text(_json.dumps({"v": 1, "token": "jwt_stale", "expiry": _time.time() - 1})) + + calls = self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_fresh", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test") + c.get_me() + # Stale cache ignored; /auth/token called once. + assert sum(1 for x in calls if x["url"].endswith("/auth/token")) == 1 + # Cache rewritten with the fresh token. + assert c._token == "jwt_fresh" + + def test_corrupt_cache_file_falls_through_to_fresh_auth(self, monkeypatch, tmp_path): + """A garbage cache file is silently ignored and a fresh token is + fetched. Cache correctness is not load-bearing.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + + from colony_sdk.client import _token_cache_path + + path = _token_cache_path("col_test", "https://thecolony.cc/api/v1") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{not valid json at all") + + calls = self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_recovered", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test") + c.get_me() # MUST NOT raise + assert sum(1 for x in calls if x["url"].endswith("/auth/token")) == 1 + assert c._token == "jwt_recovered" + + def test_cache_token_false_per_client_disables_persistence(self, monkeypatch, tmp_path): + """When the constructor arg is False, no cache file is written — + even if the env var would otherwise enable caching.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_no_cache", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test", cache_token=False) + c.get_me() + assert list(tmp_path.glob("*.json")) == [] + + def test_env_var_disables_cache_globally(self, monkeypatch, tmp_path): + """`COLONY_SDK_NO_TOKEN_CACHE=1` disables caching even when the + per-client setting would enable it. Operator-level kill switch + without code change.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + monkeypatch.setenv("COLONY_SDK_NO_TOKEN_CACHE", "1") + self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_global_off", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test") # cache_token defaults to True + c.get_me() + assert list(tmp_path.glob("*.json")) == [] + + def test_different_api_keys_get_different_cache_files(self, monkeypatch, tmp_path): + """The cache filename is keyed by (base_url, api_key) — two clients + with different keys must not collide. Otherwise rotating an api_key + would silently re-load the old key's token until expiry.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_key_a", "expires_in": 86400}), + ("ok", 200, {"username": "alice"}), + ("ok", 200, {"access_token": "jwt_key_b", "expires_in": 86400}), + ("ok", 200, {"username": "bob"}), + ], + ) + ColonyClient("col_key_alice").get_me() + ColonyClient("col_key_bob").get_me() + assert len(list(tmp_path.glob("*.json"))) == 2 + + def test_different_base_urls_get_different_cache_files(self, monkeypatch, tmp_path): + """Same api_key against prod vs staging must get independent cache + files — same key may be valid on both bases with different tokens.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_prod", "expires_in": 86400}), + ("ok", 200, {"username": "u"}), + ("ok", 200, {"access_token": "jwt_staging", "expires_in": 86400}), + ("ok", 200, {"username": "u"}), + ], + ) + ColonyClient("col_same", base_url="https://thecolony.cc/api/v1").get_me() + ColonyClient("col_same", base_url="https://staging.example/api/v1").get_me() + assert len(list(tmp_path.glob("*.json"))) == 2 + + def test_refresh_token_removes_cache_file(self, monkeypatch, tmp_path): + """`refresh_token()` clears both in-memory and on-disk state so + the next request hits `/auth/token` even on a fresh process.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_initial", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test") + c.get_me() + assert len(list(tmp_path.glob("*.json"))) == 1 + c.refresh_token() + assert list(tmp_path.glob("*.json")) == [] + assert c._token is None + + def test_401_response_invalidates_disk_cache(self, monkeypatch, tmp_path): + """A 401 from the server means the (possibly cached) token is stale. + The disk cache must be cleared so the next process doesn't re-load + the same stale token and immediately 401 again.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + # Pre-seed a "valid-looking but server-rejected" token. + import json as _json + import time as _time + + from colony_sdk.client import _token_cache_path + + stale_path = _token_cache_path("col_test", "https://thecolony.cc/api/v1") + stale_path.parent.mkdir(parents=True, exist_ok=True) + stale_path.write_text(_json.dumps({"v": 1, "token": "jwt_revoked", "expiry": _time.time() + 86400})) + + self._patch( + monkeypatch, + [ + ("http_error", 401, '{"detail":"invalid token"}'), # /users/me with cached jwt_revoked + ("ok", 200, {"access_token": "jwt_new", "expires_in": 86400}), # /auth/token refresh + ("ok", 200, {"username": "colonist-one"}), # /users/me retry + ], + ) + c = ColonyClient("col_test") + result = c.get_me() + assert result["username"] == "colonist-one" + # Cache file rewritten with the new token (not zero — _ensure_token + # wrote the new one after fetching). + assert len(list(tmp_path.glob("*.json"))) == 1 + cached = _json.loads(next(iter(tmp_path.glob("*.json"))).read_text()) + assert cached["token"] == "jwt_new" + + def test_cache_dir_explicit_override_wins_on_every_platform(self, monkeypatch, tmp_path): + """`COLONY_SDK_TOKEN_CACHE_DIR` short-circuits all platform + detection — it's the escape hatch for tests, multi-user hosts, + and anyone who wants the cache somewhere specific.""" + from colony_sdk.client import _token_cache_dir + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path / "explicit")) + # Force a non-Linux platform — override must still win. + monkeypatch.setattr("colony_sdk.client.sys.platform", "win32") + assert _token_cache_dir() == tmp_path / "explicit" + + def test_cache_dir_linux_honors_xdg_cache_home(self, monkeypatch, tmp_path): + """Linux: `$XDG_CACHE_HOME/colony-sdk` per the XDG Base Directory Spec.""" + from colony_sdk.client import _token_cache_dir + + monkeypatch.delenv("COLONY_SDK_TOKEN_CACHE_DIR", raising=False) + monkeypatch.setattr("colony_sdk.client.sys.platform", "linux") + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + assert _token_cache_dir() == tmp_path / "colony-sdk" + + def test_cache_dir_linux_falls_back_to_home_dot_cache(self, monkeypatch, tmp_path): + """Linux without XDG_CACHE_HOME falls back to `~/.cache/colony-sdk`.""" + from pathlib import Path + + from colony_sdk.client import _token_cache_dir + + monkeypatch.delenv("COLONY_SDK_TOKEN_CACHE_DIR", raising=False) + monkeypatch.delenv("XDG_CACHE_HOME", raising=False) + monkeypatch.setattr("colony_sdk.client.sys.platform", "linux") + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + assert _token_cache_dir() == tmp_path / ".cache" / "colony-sdk" + + def test_cache_dir_macos_uses_library_caches(self, monkeypatch, tmp_path): + """macOS: `~/Library/Caches/colony-sdk` per Apple's File System + Programming Guide.""" + from pathlib import Path + + from colony_sdk.client import _token_cache_dir + + monkeypatch.delenv("COLONY_SDK_TOKEN_CACHE_DIR", raising=False) + monkeypatch.setattr("colony_sdk.client.sys.platform", "darwin") + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + # Even with XDG_CACHE_HOME set, macOS should ignore it. + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "wrong")) + assert _token_cache_dir() == tmp_path / "Library" / "Caches" / "colony-sdk" + + def test_cache_dir_windows_prefers_localappdata(self, monkeypatch, tmp_path): + """Windows: `%LOCALAPPDATA%\\colony-sdk\\Cache` — machine-local + rather than roamed. (Local cache shouldn't sync to other + machines via the roaming profile.)""" + from colony_sdk.client import _token_cache_dir + + monkeypatch.delenv("COLONY_SDK_TOKEN_CACHE_DIR", raising=False) + monkeypatch.setattr("colony_sdk.client.sys.platform", "win32") + monkeypatch.setenv("LOCALAPPDATA", str(tmp_path / "Local")) + monkeypatch.setenv("APPDATA", str(tmp_path / "Roaming")) + result = _token_cache_dir() + assert result == tmp_path / "Local" / "colony-sdk" / "Cache" + + def test_cache_dir_windows_falls_back_to_appdata(self, monkeypatch, tmp_path): + """Windows without LOCALAPPDATA falls back to APPDATA (still + better than dumping under home root).""" + from colony_sdk.client import _token_cache_dir + + monkeypatch.delenv("COLONY_SDK_TOKEN_CACHE_DIR", raising=False) + monkeypatch.setattr("colony_sdk.client.sys.platform", "win32") + monkeypatch.delenv("LOCALAPPDATA", raising=False) + monkeypatch.setenv("APPDATA", str(tmp_path / "Roaming")) + result = _token_cache_dir() + assert result == tmp_path / "Roaming" / "colony-sdk" / "Cache" + + def test_cache_dir_windows_falls_back_to_home_appdata_local(self, monkeypatch, tmp_path): + """Windows with neither env var falls back to the conventional + `~/AppData/Local/colony-sdk/Cache` path.""" + from pathlib import Path + + from colony_sdk.client import _token_cache_dir + + monkeypatch.delenv("COLONY_SDK_TOKEN_CACHE_DIR", raising=False) + monkeypatch.setattr("colony_sdk.client.sys.platform", "win32") + monkeypatch.delenv("LOCALAPPDATA", raising=False) + monkeypatch.delenv("APPDATA", raising=False) + monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) + assert _token_cache_dir() == tmp_path / "AppData" / "Local" / "colony-sdk" / "Cache" + + def test_save_swallows_mid_write_oserror(self, monkeypatch, tmp_path): + """If `json.dump` raises an OSError mid-write (disk full, broken + pipe, etc.), the tmp file is cleaned up and the outer save call + returns silently — caching never blocks a request from completing. + + Exercises the inner-except partial-write cleanup branch + outer + OSError swallow.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + + import colony_sdk.client as _client_mod + + def _exploding_dump(obj, fp, **kwargs): + # OSError-shape — same as what a real disk-full scenario raises + # mid json.dump call when the underlying fd hits ENOSPC. + raise OSError(28, "No space left on device") + + monkeypatch.setattr(_client_mod.json, "dump", _exploding_dump) + + c = ColonyClient("col_test") + c._token = "jwt_will_fail_to_save" + c._token_expiry = 9999999999 + # MUST NOT raise — _save_cached_token is best-effort under OSError. + c._save_cached_token() + # No stale tmp file left behind, no final cache file either. + assert list(tmp_path.glob("*")) == [] + + def test_save_swallows_outer_oserror(self, monkeypatch, tmp_path): + """If mkdir/open at the very top raises OSError, save returns + silently — never propagates to the caller.""" + monkeypatch.setenv( + "COLONY_SDK_TOKEN_CACHE_DIR", + "/proc/1/root/cache-cannot-write-here", # path that can't be created + ) + c = ColonyClient("col_test") + c._token = "jwt_unwritable" + c._token_expiry = 9999999999 + # MUST NOT raise. + c._save_cached_token() + + def test_clear_cached_token_no_op_when_cache_disabled(self, monkeypatch, tmp_path): + """`_clear_cached_token` early-returns without touching the + filesystem when caching is globally disabled — protects against + accidentally nuking a file someone else's process owns.""" + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + monkeypatch.setenv("COLONY_SDK_NO_TOKEN_CACHE", "1") + # Pre-seed a file at the path that WOULD be the cache target. + from colony_sdk.client import _token_cache_path + + # Compute path before the env-disable check would skip the unlink. + # We need to bypass the disable to compute the path, so check the + # path without invoking _clear_cached_token. + # Set up: cache disabled by env, but a file exists at the path + # (could be left over from a previous run). + monkeypatch.delenv("COLONY_SDK_NO_TOKEN_CACHE", raising=False) + path = _token_cache_path("col_test", "https://thecolony.cc/api/v1") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text('{"v":1,"token":"untouched","expiry":9999999999}') + # Now disable caching and call clear — file must remain. + monkeypatch.setenv("COLONY_SDK_NO_TOKEN_CACHE", "1") + c = ColonyClient("col_test") + c._clear_cached_token() + assert path.exists(), "cache disabled → clear must not touch the filesystem" + + def test_safety_margin_treats_near_expiry_as_miss(self, monkeypatch, tmp_path): + """A token whose expiry is within the 60s safety margin is treated + as a cache miss — otherwise a long request could outlive the token + and 401 mid-flight.""" + import json as _json + import time as _time + + monkeypatch.setenv("COLONY_SDK_TOKEN_CACHE_DIR", str(tmp_path)) + from colony_sdk.client import _token_cache_path + + path = _token_cache_path("col_test", "https://thecolony.cc/api/v1") + path.parent.mkdir(parents=True, exist_ok=True) + # Token "expires" in 30s — within the 60s safety margin. + path.write_text(_json.dumps({"v": 1, "token": "jwt_near_expiry", "expiry": _time.time() + 30})) + + calls = self._patch( + monkeypatch, + [ + ("ok", 200, {"access_token": "jwt_refreshed", "expires_in": 86400}), + ("ok", 200, {"username": "colonist-one"}), + ], + ) + c = ColonyClient("col_test") + c.get_me() + assert sum(1 for x in calls if x["url"].endswith("/auth/token")) == 1 + assert c._token == "jwt_refreshed"