From d10121e22aef01b96809de0088be48254e158a1a Mon Sep 17 00:00:00 2001 From: Tim Huff Date: Wed, 29 Jul 2026 10:56:46 -0700 Subject: [PATCH 1/5] Allow disabling automatic API token rotation Add enable_token_rotation to Groundlight/TokenManager so proxies like Edge Endpoint can forward a caller's token without minting or revoking against that identity's rotation chain. Defaults to True to preserve existing behavior. Co-authored-by: Cursor --- src/groundlight/client.py | 9 ++++++- src/groundlight/experimental_api.py | 12 +++++++-- src/groundlight/token_manager.py | 34 ++++++++++++++++++++------ test/unit/test_token_manager.py | 28 ++++++++++++++++++++- test/unit/test_token_refresh_client.py | 13 ++++++++++ 5 files changed, 85 insertions(+), 11 deletions(-) diff --git a/src/groundlight/client.py b/src/groundlight/client.py index 72cb74de..11df212b 100644 --- a/src/groundlight/client.py +++ b/src/groundlight/client.py @@ -125,6 +125,9 @@ class Groundlight: # pylint: disable=too-many-instance-attributes,too-many-publ Warning: Only disable verification when connecting to a Groundlight Edge Endpoint using self-signed certificates. For security, always keep verification enabled when using the Groundlight cloud service. + :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a + non-null Token TTL. Set False for proxies that forward a caller's token without owning + rotation (for example, Groundlight Edge Endpoint request handling). :return: Groundlight client instance """ @@ -141,6 +144,7 @@ def __init__( api_token: Optional[str] = None, disable_tls_verification: Optional[bool] = None, http_transport_retries: Optional[Union[int, Retry]] = None, + enable_token_rotation: bool = True, ): """ Initialize a new Groundlight client instance. @@ -156,6 +160,8 @@ def __init__( certificates. For security, always keep verification enabled when using the Groundlight cloud service. :param http_transport_retries: Overrides urllib3 `PoolManager` retry policy for HTTP/HTTPS (forwarded to `Configuration.retries`). Not the same as SDK 5xx retries handled by `RequestsRetryDecorator`. + :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a + non-null Token TTL. Set False when forwarding a caller's token without owning its rotation chain. :return: Groundlight client """ @@ -199,6 +205,7 @@ def __init__( configured_token=api_token, configuration=self.configuration, request_timeout=DEFAULT_REQUEST_TIMEOUT, + enable_token_rotation=enable_token_rotation, ) except TokenManagerError as exc: raise ApiTokenError(str(exc)) from exc @@ -211,7 +218,7 @@ def __init__( self.month_to_date_api = MonthToDateAccountInfoApi(self.api_client) self.logged_in_user = "(not-logged-in)" self._verify_connectivity() - # No-op when the working token has no identity Token TTL. + # No-op when rotation is disabled or the working token has no identity Token TTL. self._token_manager.start() def __repr__(self) -> str: diff --git a/src/groundlight/experimental_api.py b/src/groundlight/experimental_api.py index 8deb93e0..fb670be0 100644 --- a/src/groundlight/experimental_api.py +++ b/src/groundlight/experimental_api.py @@ -52,6 +52,7 @@ def __init__( endpoint: Union[str, None] = None, api_token: Union[str, None] = None, disable_tls_verification: Optional[bool] = None, + enable_token_rotation: bool = True, ): """ Constructs an experimental Groundlight client. @@ -83,8 +84,15 @@ def __init__( Warning: Only disable verification when connecting to a Groundlight Edge Endpoint using self-signed certificates. For security, always keep verification enabled when using the Groundlight cloud service. - """ - super().__init__(endpoint=endpoint, api_token=api_token, disable_tls_verification=disable_tls_verification) + :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a + non-null Token TTL. Set False when forwarding a caller's token without owning its rotation chain. + """ + super().__init__( + endpoint=endpoint, + api_token=api_token, + disable_tls_verification=disable_tls_verification, + enable_token_rotation=enable_token_rotation, + ) self.notes_api = NotesApi(self.api_client) self.detector_group_api = DetectorGroupsApi(self.api_client) self.detector_reset_api = DetectorResetApi(self.api_client) diff --git a/src/groundlight/token_manager.py b/src/groundlight/token_manager.py index da6adc96..cf4bc7b4 100644 --- a/src/groundlight/token_manager.py +++ b/src/groundlight/token_manager.py @@ -167,8 +167,14 @@ def __init__( configuration: Configuration, request_timeout: float, token_dir: Optional[Path] = None, + enable_token_rotation: bool = True, ): - """Initialize the cache slot and select or mint a working API token.""" + """Initialize the cache slot and select or mint a working API token. + + When enable_token_rotation is False, use the configured token as-is with no by-snippet + lookup, on-disk cache, or background refresh. Intended for proxies (e.g. Edge Endpoint) + that forward a caller's token without owning its rotation chain. + """ self._configured_token = configured_token self._configured_snippet = configured_token[:TOKEN_SNIPPET_LENGTH] if len(self._configured_snippet) != TOKEN_SNIPPET_LENGTH or not re.fullmatch( @@ -179,14 +185,22 @@ def __init__( ) self._configuration = configuration self._request_timeout = request_timeout - self._token_dir = token_dir or self._default_token_dir() - self._slot_path = self._token_dir / f"{self._configured_snippet}.json" - self._lock_path = self._token_dir / f"{self._configured_snippet}.lock" - self._lock = FileLock(str(self._lock_path), timeout=LOCK_TIMEOUT_SECONDS, mode=0o600) + self._rotation_enabled = enable_token_rotation self._stop_event = threading.Event() self._thread: Optional[threading.Thread] = None self._current: Optional[CurrentToken] = None self._available = True + self._rotation_client: Optional[GroundlightApiClient] = None + self._api_tokens: Optional[ApiTokensApi] = None + self._token_dir = token_dir or self._default_token_dir() + self._slot_path = self._token_dir / f"{self._configured_snippet}.json" + self._lock_path = self._token_dir / f"{self._configured_snippet}.lock" + self._lock = FileLock(str(self._lock_path), timeout=LOCK_TIMEOUT_SECONDS, mode=0o600) + + if not self._rotation_enabled: + self._available = False + self._set_api_token(self._configured_token) + return self._ensure_token_dir() self._rotation_client = GroundlightApiClient(configuration) @@ -274,7 +288,7 @@ def _is_usable_cached_token(token: CurrentToken) -> bool: def start(self) -> None: """Start background refresh when the working token has a finite Token TTL.""" - if not self._available or self._thread is not None: + if not self._rotation_enabled or not self._available or self._thread is not None: return if self._current is None or self._current.token_ttl is None: return @@ -290,7 +304,8 @@ def close(self) -> None: self._stop_event.set() if self._thread is not None: self._thread.join() - self._rotation_client.close() + if self._rotation_client is not None: + self._rotation_client.close() def refresh(self) -> bool: """Use the cached token if it is still fresh; otherwise rotate under the file lock. @@ -299,6 +314,8 @@ def refresh(self) -> bool: current to previous, and mint a new current. Returns False only when rotation could not run (lock timeout, mint failure, or token API unavailable). """ + if not self._rotation_enabled: + return True try: with self._lock: slot = self._load_slot() @@ -404,6 +421,7 @@ def _write_slot(self, slot: TokenSlot) -> None: def _mint_replacement(self, base_name: str, previous: Optional[PreviousToken]) -> CurrentToken: """Mint a new token, persist the updated slot, and activate the new credential.""" + assert self._api_tokens is not None # only called when rotation is enabled new_name = self._new_token_name(base_name) minted_at = _utc_now() # Omit expires_at so the server applies the identity's token lifetime policy. @@ -418,12 +436,14 @@ def _mint_replacement(self, base_name: str, previous: Optional[PreviousToken]) - def _get_token_by_snippet(self, snippet: str) -> ApiToken: """Retrieve token metadata by snippet via the dedicated API endpoint.""" + assert self._api_tokens is not None # only called when rotation is enabled return self._api_tokens.get_api_token_by_snippet(snippet, _request_timeout=self._request_timeout) def _revoke_previous(self, previous: Optional[PreviousToken]) -> None: """Best-effort revoke of the demoted previous token before it is replaced in the slot.""" if previous is None: return + assert self._api_tokens is not None # only called when rotation is enabled try: self._api_tokens.delete_api_token(previous.name, _request_timeout=self._request_timeout) except NotFoundException: diff --git a/test/unit/test_token_manager.py b/test/unit/test_token_manager.py index 02c8d664..d44f4855 100644 --- a/test/unit/test_token_manager.py +++ b/test/unit/test_token_manager.py @@ -62,7 +62,7 @@ def _created_token( ) -def _manager(mocker, tmp_path, api, now=NOW) -> TokenManager: +def _manager(mocker, tmp_path, api, now=NOW, *, enable_token_rotation: bool = True) -> TokenManager: """Create a token manager with deterministic API and time dependencies.""" mocker.patch.object(token_manager, "ApiTokensApi", return_value=api) mocker.patch.object(token_manager, "_utc_now", return_value=now) @@ -73,6 +73,7 @@ def _manager(mocker, tmp_path, api, now=NOW) -> TokenManager: configuration=configuration, request_timeout=1, token_dir=tmp_path, + enable_token_rotation=enable_token_rotation, ) @@ -126,6 +127,31 @@ def test_initialization_uses_never_expire_configured_token_as_is(mocker, tmp_pat api.create_api_token.assert_not_called() +def test_initialization_skips_rotation_when_disabled(mocker, tmp_path): + """Disabling rotation uses the configured token with no token API, cache, or refresh thread.""" + api_tokens_cls = mocker.patch.object(token_manager, "ApiTokensApi") + mocker.patch.object(token_manager, "_utc_now", return_value=NOW) + configuration = Configuration(host="https://example.com/device-api") + configuration.api_key["ApiToken"] = CONFIGURED_TOKEN + + manager = TokenManager( + configured_token=CONFIGURED_TOKEN, + configuration=configuration, + request_timeout=1, + token_dir=tmp_path, + enable_token_rotation=False, + ) + manager.start() + assert manager.refresh() is True + + assert manager._configuration.api_key["ApiToken"] == CONFIGURED_TOKEN + assert manager._current is None + assert manager._thread is None + assert manager._rotation_client is None + assert not manager._slot_path.exists() + api_tokens_cls.assert_not_called() + + def test_initialization_hard_deadline_without_token_ttl_does_not_rotate(mocker, tmp_path): """Hard-deadline-only identities (token_ttl null, expires_at set) do not rotate.""" api = Mock() diff --git a/test/unit/test_token_refresh_client.py b/test/unit/test_token_refresh_client.py index 7c58c0e2..5cb9fd4a 100644 --- a/test/unit/test_token_refresh_client.py +++ b/test/unit/test_token_refresh_client.py @@ -14,7 +14,20 @@ def test_groundlight_starts_and_closes_token_manager(mocker): with client as entered_client: assert entered_client is client token_manager_class.assert_called_once() + assert token_manager_class.call_args.kwargs["enable_token_rotation"] is True manager.start.assert_called_once() manager.close.assert_called_once() api_client_close.assert_called_once() + + +def test_groundlight_forwards_enable_token_rotation(mocker): + """Groundlight passes enable_token_rotation through to TokenManager.""" + manager = Mock() + token_manager_class = mocker.patch("groundlight.client.TokenManager", return_value=manager) + mocker.patch.object(Groundlight, "_verify_connectivity") + + client = Groundlight(api_token="api_bootstrap_token_value_long_enough", enable_token_rotation=False) + client.close() + + assert token_manager_class.call_args.kwargs["enable_token_rotation"] is False From 10c69573183646574f3646a740b62ab2db1a6267 Mon Sep 17 00:00:00 2001 From: Tim Huff Date: Wed, 29 Jul 2026 11:07:19 -0700 Subject: [PATCH 2/5] Address PR review: fix lint and tighten disabled-rotation path Silence PLR0913 on the new constructor args, drop the unreachable refresh guard and redundant _available flag, defer cache/lock setup until rotation is enabled, and cover close() plus ExperimentalApi forwarding. Co-authored-by: Cursor --- src/groundlight/client.py | 2 +- src/groundlight/token_manager.py | 13 +++++-------- test/unit/test_token_manager.py | 4 ++-- test/unit/test_token_refresh_client.py | 13 +++++++++++++ 4 files changed, 21 insertions(+), 11 deletions(-) diff --git a/src/groundlight/client.py b/src/groundlight/client.py index 11df212b..1a84c1a5 100644 --- a/src/groundlight/client.py +++ b/src/groundlight/client.py @@ -138,7 +138,7 @@ class Groundlight: # pylint: disable=too-many-instance-attributes,too-many-publ POLLING_EXPONENTIAL_BACKOFF = 1.3 # This still has the nice backoff property that the max number of requests # is O(log(time)), but with 1.3 the guarantee is that the call will return no more than 30% late - def __init__( + def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments self, endpoint: Optional[str] = None, api_token: Optional[str] = None, diff --git a/src/groundlight/token_manager.py b/src/groundlight/token_manager.py index cf4bc7b4..e544533f 100644 --- a/src/groundlight/token_manager.py +++ b/src/groundlight/token_manager.py @@ -161,7 +161,7 @@ def to_dict(self) -> Dict[str, Any]: class TokenManager: # pylint: disable=too-many-instance-attributes """Manage cached API tokens and coordinate their automatic rotation.""" - def __init__( + def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments self, configured_token: str, configuration: Configuration, @@ -192,16 +192,15 @@ def __init__( self._available = True self._rotation_client: Optional[GroundlightApiClient] = None self._api_tokens: Optional[ApiTokensApi] = None - self._token_dir = token_dir or self._default_token_dir() - self._slot_path = self._token_dir / f"{self._configured_snippet}.json" - self._lock_path = self._token_dir / f"{self._configured_snippet}.lock" - self._lock = FileLock(str(self._lock_path), timeout=LOCK_TIMEOUT_SECONDS, mode=0o600) if not self._rotation_enabled: - self._available = False self._set_api_token(self._configured_token) return + self._token_dir = token_dir or self._default_token_dir() + self._slot_path = self._token_dir / f"{self._configured_snippet}.json" + self._lock_path = self._token_dir / f"{self._configured_snippet}.lock" + self._lock = FileLock(str(self._lock_path), timeout=LOCK_TIMEOUT_SECONDS, mode=0o600) self._ensure_token_dir() self._rotation_client = GroundlightApiClient(configuration) self._api_tokens = ApiTokensApi(self._rotation_client) @@ -314,8 +313,6 @@ def refresh(self) -> bool: current to previous, and mint a new current. Returns False only when rotation could not run (lock timeout, mint failure, or token API unavailable). """ - if not self._rotation_enabled: - return True try: with self._lock: slot = self._load_slot() diff --git a/test/unit/test_token_manager.py b/test/unit/test_token_manager.py index d44f4855..2b670440 100644 --- a/test/unit/test_token_manager.py +++ b/test/unit/test_token_manager.py @@ -142,13 +142,13 @@ def test_initialization_skips_rotation_when_disabled(mocker, tmp_path): enable_token_rotation=False, ) manager.start() - assert manager.refresh() is True + manager.close() assert manager._configuration.api_key["ApiToken"] == CONFIGURED_TOKEN assert manager._current is None assert manager._thread is None assert manager._rotation_client is None - assert not manager._slot_path.exists() + assert list(tmp_path.iterdir()) == [] api_tokens_cls.assert_not_called() diff --git a/test/unit/test_token_refresh_client.py b/test/unit/test_token_refresh_client.py index 5cb9fd4a..9df59eab 100644 --- a/test/unit/test_token_refresh_client.py +++ b/test/unit/test_token_refresh_client.py @@ -1,6 +1,7 @@ from unittest.mock import Mock from groundlight.client import Groundlight +from groundlight.experimental_api import ExperimentalApi def test_groundlight_starts_and_closes_token_manager(mocker): @@ -31,3 +32,15 @@ def test_groundlight_forwards_enable_token_rotation(mocker): client.close() assert token_manager_class.call_args.kwargs["enable_token_rotation"] is False + + +def test_experimental_api_forwards_enable_token_rotation(mocker): + """ExperimentalApi passes enable_token_rotation through to TokenManager.""" + manager = Mock() + token_manager_class = mocker.patch("groundlight.client.TokenManager", return_value=manager) + mocker.patch.object(Groundlight, "_verify_connectivity") + + client = ExperimentalApi(api_token="api_bootstrap_token_value_long_enough", enable_token_rotation=False) + client.close() + + assert token_manager_class.call_args.kwargs["enable_token_rotation"] is False From d262c05221aac6cc9b6d420e2d33e920a9eb2697 Mon Sep 17 00:00:00 2001 From: Tim Huff Date: Wed, 29 Jul 2026 12:14:18 -0700 Subject: [PATCH 3/5] Drop proxy-specific wording from token rotation docs Keep the behavioral description of enable_token_rotation without naming Edge Endpoint or other call-site examples. Co-authored-by: Cursor --- src/groundlight/client.py | 5 ++--- src/groundlight/experimental_api.py | 2 +- src/groundlight/token_manager.py | 3 +-- 3 files changed, 4 insertions(+), 6 deletions(-) diff --git a/src/groundlight/client.py b/src/groundlight/client.py index 1a84c1a5..e8280aef 100644 --- a/src/groundlight/client.py +++ b/src/groundlight/client.py @@ -126,8 +126,7 @@ class Groundlight: # pylint: disable=too-many-instance-attributes,too-many-publ self-signed certificates. For security, always keep verification enabled when using the Groundlight cloud service. :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a - non-null Token TTL. Set False for proxies that forward a caller's token without owning - rotation (for example, Groundlight Edge Endpoint request handling). + non-null Token TTL. :return: Groundlight client instance """ @@ -161,7 +160,7 @@ def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments :param http_transport_retries: Overrides urllib3 `PoolManager` retry policy for HTTP/HTTPS (forwarded to `Configuration.retries`). Not the same as SDK 5xx retries handled by `RequestsRetryDecorator`. :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a - non-null Token TTL. Set False when forwarding a caller's token without owning its rotation chain. + non-null Token TTL. :return: Groundlight client """ diff --git a/src/groundlight/experimental_api.py b/src/groundlight/experimental_api.py index fb670be0..09258c6f 100644 --- a/src/groundlight/experimental_api.py +++ b/src/groundlight/experimental_api.py @@ -85,7 +85,7 @@ def __init__( self-signed certificates. For security, always keep verification enabled when using the Groundlight cloud service. :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a - non-null Token TTL. Set False when forwarding a caller's token without owning its rotation chain. + non-null Token TTL. """ super().__init__( endpoint=endpoint, diff --git a/src/groundlight/token_manager.py b/src/groundlight/token_manager.py index e544533f..ddce1a9e 100644 --- a/src/groundlight/token_manager.py +++ b/src/groundlight/token_manager.py @@ -172,8 +172,7 @@ def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments """Initialize the cache slot and select or mint a working API token. When enable_token_rotation is False, use the configured token as-is with no by-snippet - lookup, on-disk cache, or background refresh. Intended for proxies (e.g. Edge Endpoint) - that forward a caller's token without owning its rotation chain. + lookup, on-disk cache, or background refresh. """ self._configured_token = configured_token self._configured_snippet = configured_token[:TOKEN_SNIPPET_LENGTH] From a3906444e2d639758bb8b5a89a72d7bec6bb4561 Mon Sep 17 00:00:00 2001 From: Tim Huff Date: Wed, 29 Jul 2026 12:19:04 -0700 Subject: [PATCH 4/5] Gate TokenManager construction in Groundlight instead of TokenManager When enable_token_rotation is False, skip creating TokenManager entirely so the configured token is used as-is. TokenManager stays a pure rotation collaborator with no disabled mode. Co-authored-by: Cursor --- src/groundlight/client.py | 27 ++++++++++++---------- src/groundlight/token_manager.py | 32 +++++++------------------- test/unit/test_token_manager.py | 28 +--------------------- test/unit/test_token_refresh_client.py | 24 ++++++++++--------- 4 files changed, 37 insertions(+), 74 deletions(-) diff --git a/src/groundlight/client.py b/src/groundlight/client.py index e8280aef..c988c8f1 100644 --- a/src/groundlight/client.py +++ b/src/groundlight/client.py @@ -199,15 +199,16 @@ def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments self.configuration.api_key["ApiToken"] = api_token self.api_client = GroundlightApiClient(self.configuration) - try: - self._token_manager = TokenManager( - configured_token=api_token, - configuration=self.configuration, - request_timeout=DEFAULT_REQUEST_TIMEOUT, - enable_token_rotation=enable_token_rotation, - ) - except TokenManagerError as exc: - raise ApiTokenError(str(exc)) from exc + self._token_manager: Optional[TokenManager] = None + if enable_token_rotation: + try: + self._token_manager = TokenManager( + configured_token=api_token, + configuration=self.configuration, + request_timeout=DEFAULT_REQUEST_TIMEOUT, + ) + except TokenManagerError as exc: + raise ApiTokenError(str(exc)) from exc self.detectors_api = DetectorsApi(self.api_client) self.detector_group_api = DetectorGroupsApi(self.api_client) self.images_api = ImageQueriesApi(self.api_client) @@ -217,8 +218,9 @@ def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments self.month_to_date_api = MonthToDateAccountInfoApi(self.api_client) self.logged_in_user = "(not-logged-in)" self._verify_connectivity() - # No-op when rotation is disabled or the working token has no identity Token TTL. - self._token_manager.start() + # No-op when the working token has no identity Token TTL. + if self._token_manager is not None: + self._token_manager.start() def __repr__(self) -> str: # Don't call the API here because that can get us stuck in a loop rendering exception strings @@ -234,7 +236,8 @@ def __exit__(self, exc_type, exc_value, traceback) -> None: def close(self) -> None: """Stop the token refresh thread and close the HTTP client.""" - self._token_manager.close() + if self._token_manager is not None: + self._token_manager.close() self.api_client.close() def _verify_connectivity(self) -> None: diff --git a/src/groundlight/token_manager.py b/src/groundlight/token_manager.py index ddce1a9e..da6adc96 100644 --- a/src/groundlight/token_manager.py +++ b/src/groundlight/token_manager.py @@ -161,19 +161,14 @@ def to_dict(self) -> Dict[str, Any]: class TokenManager: # pylint: disable=too-many-instance-attributes """Manage cached API tokens and coordinate their automatic rotation.""" - def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments + def __init__( self, configured_token: str, configuration: Configuration, request_timeout: float, token_dir: Optional[Path] = None, - enable_token_rotation: bool = True, ): - """Initialize the cache slot and select or mint a working API token. - - When enable_token_rotation is False, use the configured token as-is with no by-snippet - lookup, on-disk cache, or background refresh. - """ + """Initialize the cache slot and select or mint a working API token.""" self._configured_token = configured_token self._configured_snippet = configured_token[:TOKEN_SNIPPET_LENGTH] if len(self._configured_snippet) != TOKEN_SNIPPET_LENGTH or not re.fullmatch( @@ -184,22 +179,15 @@ def __init__( # noqa: PLR0913 # pylint: disable=too-many-arguments ) self._configuration = configuration self._request_timeout = request_timeout - self._rotation_enabled = enable_token_rotation + self._token_dir = token_dir or self._default_token_dir() + self._slot_path = self._token_dir / f"{self._configured_snippet}.json" + self._lock_path = self._token_dir / f"{self._configured_snippet}.lock" + self._lock = FileLock(str(self._lock_path), timeout=LOCK_TIMEOUT_SECONDS, mode=0o600) self._stop_event = threading.Event() self._thread: Optional[threading.Thread] = None self._current: Optional[CurrentToken] = None self._available = True - self._rotation_client: Optional[GroundlightApiClient] = None - self._api_tokens: Optional[ApiTokensApi] = None - if not self._rotation_enabled: - self._set_api_token(self._configured_token) - return - - self._token_dir = token_dir or self._default_token_dir() - self._slot_path = self._token_dir / f"{self._configured_snippet}.json" - self._lock_path = self._token_dir / f"{self._configured_snippet}.lock" - self._lock = FileLock(str(self._lock_path), timeout=LOCK_TIMEOUT_SECONDS, mode=0o600) self._ensure_token_dir() self._rotation_client = GroundlightApiClient(configuration) self._api_tokens = ApiTokensApi(self._rotation_client) @@ -286,7 +274,7 @@ def _is_usable_cached_token(token: CurrentToken) -> bool: def start(self) -> None: """Start background refresh when the working token has a finite Token TTL.""" - if not self._rotation_enabled or not self._available or self._thread is not None: + if not self._available or self._thread is not None: return if self._current is None or self._current.token_ttl is None: return @@ -302,8 +290,7 @@ def close(self) -> None: self._stop_event.set() if self._thread is not None: self._thread.join() - if self._rotation_client is not None: - self._rotation_client.close() + self._rotation_client.close() def refresh(self) -> bool: """Use the cached token if it is still fresh; otherwise rotate under the file lock. @@ -417,7 +404,6 @@ def _write_slot(self, slot: TokenSlot) -> None: def _mint_replacement(self, base_name: str, previous: Optional[PreviousToken]) -> CurrentToken: """Mint a new token, persist the updated slot, and activate the new credential.""" - assert self._api_tokens is not None # only called when rotation is enabled new_name = self._new_token_name(base_name) minted_at = _utc_now() # Omit expires_at so the server applies the identity's token lifetime policy. @@ -432,14 +418,12 @@ def _mint_replacement(self, base_name: str, previous: Optional[PreviousToken]) - def _get_token_by_snippet(self, snippet: str) -> ApiToken: """Retrieve token metadata by snippet via the dedicated API endpoint.""" - assert self._api_tokens is not None # only called when rotation is enabled return self._api_tokens.get_api_token_by_snippet(snippet, _request_timeout=self._request_timeout) def _revoke_previous(self, previous: Optional[PreviousToken]) -> None: """Best-effort revoke of the demoted previous token before it is replaced in the slot.""" if previous is None: return - assert self._api_tokens is not None # only called when rotation is enabled try: self._api_tokens.delete_api_token(previous.name, _request_timeout=self._request_timeout) except NotFoundException: diff --git a/test/unit/test_token_manager.py b/test/unit/test_token_manager.py index 2b670440..02c8d664 100644 --- a/test/unit/test_token_manager.py +++ b/test/unit/test_token_manager.py @@ -62,7 +62,7 @@ def _created_token( ) -def _manager(mocker, tmp_path, api, now=NOW, *, enable_token_rotation: bool = True) -> TokenManager: +def _manager(mocker, tmp_path, api, now=NOW) -> TokenManager: """Create a token manager with deterministic API and time dependencies.""" mocker.patch.object(token_manager, "ApiTokensApi", return_value=api) mocker.patch.object(token_manager, "_utc_now", return_value=now) @@ -73,7 +73,6 @@ def _manager(mocker, tmp_path, api, now=NOW, *, enable_token_rotation: bool = Tr configuration=configuration, request_timeout=1, token_dir=tmp_path, - enable_token_rotation=enable_token_rotation, ) @@ -127,31 +126,6 @@ def test_initialization_uses_never_expire_configured_token_as_is(mocker, tmp_pat api.create_api_token.assert_not_called() -def test_initialization_skips_rotation_when_disabled(mocker, tmp_path): - """Disabling rotation uses the configured token with no token API, cache, or refresh thread.""" - api_tokens_cls = mocker.patch.object(token_manager, "ApiTokensApi") - mocker.patch.object(token_manager, "_utc_now", return_value=NOW) - configuration = Configuration(host="https://example.com/device-api") - configuration.api_key["ApiToken"] = CONFIGURED_TOKEN - - manager = TokenManager( - configured_token=CONFIGURED_TOKEN, - configuration=configuration, - request_timeout=1, - token_dir=tmp_path, - enable_token_rotation=False, - ) - manager.start() - manager.close() - - assert manager._configuration.api_key["ApiToken"] == CONFIGURED_TOKEN - assert manager._current is None - assert manager._thread is None - assert manager._rotation_client is None - assert list(tmp_path.iterdir()) == [] - api_tokens_cls.assert_not_called() - - def test_initialization_hard_deadline_without_token_ttl_does_not_rotate(mocker, tmp_path): """Hard-deadline-only identities (token_ttl null, expires_at set) do not rotate.""" api = Mock() diff --git a/test/unit/test_token_refresh_client.py b/test/unit/test_token_refresh_client.py index 9df59eab..d52a1bf1 100644 --- a/test/unit/test_token_refresh_client.py +++ b/test/unit/test_token_refresh_client.py @@ -15,32 +15,34 @@ def test_groundlight_starts_and_closes_token_manager(mocker): with client as entered_client: assert entered_client is client token_manager_class.assert_called_once() - assert token_manager_class.call_args.kwargs["enable_token_rotation"] is True manager.start.assert_called_once() manager.close.assert_called_once() api_client_close.assert_called_once() -def test_groundlight_forwards_enable_token_rotation(mocker): - """Groundlight passes enable_token_rotation through to TokenManager.""" - manager = Mock() - token_manager_class = mocker.patch("groundlight.client.TokenManager", return_value=manager) +def test_groundlight_skips_token_manager_when_rotation_disabled(mocker): + """Disabling rotation leaves the configured token in place with no TokenManager.""" + token_manager_class = mocker.patch("groundlight.client.TokenManager") mocker.patch.object(Groundlight, "_verify_connectivity") client = Groundlight(api_token="api_bootstrap_token_value_long_enough", enable_token_rotation=False) + api_client_close = mocker.patch.object(client.api_client, "close") client.close() - assert token_manager_class.call_args.kwargs["enable_token_rotation"] is False + token_manager_class.assert_not_called() + assert client._token_manager is None + assert client.configuration.api_key["ApiToken"] == "api_bootstrap_token_value_long_enough" + api_client_close.assert_called_once() -def test_experimental_api_forwards_enable_token_rotation(mocker): - """ExperimentalApi passes enable_token_rotation through to TokenManager.""" - manager = Mock() - token_manager_class = mocker.patch("groundlight.client.TokenManager", return_value=manager) +def test_experimental_api_skips_token_manager_when_rotation_disabled(mocker): + """ExperimentalApi forwards enable_token_rotation=False and skips TokenManager.""" + token_manager_class = mocker.patch("groundlight.client.TokenManager") mocker.patch.object(Groundlight, "_verify_connectivity") client = ExperimentalApi(api_token="api_bootstrap_token_value_long_enough", enable_token_rotation=False) client.close() - assert token_manager_class.call_args.kwargs["enable_token_rotation"] is False + token_manager_class.assert_not_called() + assert client._token_manager is None From 59706c7dbffaca715d915bdbe8d989758f03cee0 Mon Sep 17 00:00:00 2001 From: Tim Huff Date: Wed, 29 Jul 2026 12:24:41 -0700 Subject: [PATCH 5/5] Drop ExperimentalApi-specific token rotation unit test Constructor forwarding for the subclass follows the existing pattern and is already covered by the Groundlight client test. Co-authored-by: Cursor --- test/unit/test_token_refresh_client.py | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/test/unit/test_token_refresh_client.py b/test/unit/test_token_refresh_client.py index d52a1bf1..07068415 100644 --- a/test/unit/test_token_refresh_client.py +++ b/test/unit/test_token_refresh_client.py @@ -1,7 +1,6 @@ from unittest.mock import Mock from groundlight.client import Groundlight -from groundlight.experimental_api import ExperimentalApi def test_groundlight_starts_and_closes_token_manager(mocker): @@ -34,15 +33,3 @@ def test_groundlight_skips_token_manager_when_rotation_disabled(mocker): assert client._token_manager is None assert client.configuration.api_key["ApiToken"] == "api_bootstrap_token_value_long_enough" api_client_close.assert_called_once() - - -def test_experimental_api_skips_token_manager_when_rotation_disabled(mocker): - """ExperimentalApi forwards enable_token_rotation=False and skips TokenManager.""" - token_manager_class = mocker.patch("groundlight.client.TokenManager") - mocker.patch.object(Groundlight, "_verify_connectivity") - - client = ExperimentalApi(api_token="api_bootstrap_token_value_long_enough", enable_token_rotation=False) - client.close() - - token_manager_class.assert_not_called() - assert client._token_manager is None