fix: when EJR tokens are invalid a refresh should be attempted

openeo_driver/jobregistry.py

@@ -398,6 +398,14 @@ def _do_request(
                     )
                 else:
                     response = do_request()
+
+                if response.status_code == 401 and use_auth:
+                    # Cached token was likely invalidated server-side; discard it and retry once with a fresh token.
+                    self._log.warning("EJR request got 401; invalidating token cache and retrying with fresh token")
+                    self._access_token_helper.invalidate_cache()
+                    headers["Authorization"] = f"Bearer {self._access_token_helper.get_access_token()}"
+                    response = do_request()
+                    self._log.debug(f"EJR retry response on `{method} {path}`: {response.status_code!r}")
             except Exception as e:
                 self._log.exception(f"Failed to do EJR API request `{method} {url}`: {e!r}")
                 raise EjrApiError(f"Failed to do EJR API request `{method} {url}`") from e

openeo_driver/util/auth.py

@@ -139,6 +139,10 @@ def get_access_token(self) -> str:
             self._cache = _AccessTokenCache(access_token, self._get_access_token_expiry_time(access_token_response))
         return self._cache.access_token
 
+    def invalidate_cache(self) -> None:
+        """Invalidate the cached access token, forcing a fresh fetch on the next call to ``get_access_token``."""
+        self._cache = _AccessTokenCache("", 0)
+
     def _get_access_token_expiry_time(self, access_token_response: AccessTokenResult) -> float:
         if access_token_response.expires_in is None:
             return time.time() + self._default_ttl

tests/test_jobregistry.py

@@ -1178,3 +1178,36 @@ def post_jobs_search(request, context):
                     _ = ejr.list_user_jobs(user_id="john")
 
         assert sleep.call_count > 0
+
+    def test_401_invalidates_token_cache_and_retries(self, requests_mock, oidc_mock, ejr, caplog):
+        """A 401 response should invalidate the token cache and retry the request once with a fresh token."""
+        call_count = {"n": 0}
+
+        def post_jobs_search(request, context):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                # First call: simulate a stale/revoked token by returning 401.
+                context.status_code = 401
+                return {"error": "Unauthorized"}
+            # Second call (after token refresh): succeed.
+            return [DUMMY_PROCESS]
+
+        requests_mock.post(f"{self.EJR_API_URL}/jobs/search", json=post_jobs_search)
+
+        token_requests_before = len(oidc_mock.get_request_history(url="/token"))
+        result = ejr.list_user_jobs(user_id="john")
+        assert result == [DUMMY_PROCESS]
+        # Two HTTP calls to the search endpoint: original + retry.
+        assert call_count["n"] == 2
+        # A new token must have been fetched for the retry.
+        assert len(oidc_mock.get_request_history(url="/token")) == token_requests_before + 2
+        assert "invalidating token cache" in caplog.text
+
+    def test_persistent_401_raises_error(self, requests_mock, oidc_mock, ejr):
+        """If the retry after token refresh also returns 401, EjrApiResponseError should be raised."""
+        requests_mock.post(f"{self.EJR_API_URL}/jobs/search", status_code=401)
+
+        with pytest.raises(EjrApiResponseError) as exc_info:
+            ejr.list_user_jobs(user_id="john")
+
+        assert exc_info.value.status_code == 401

tests/util/test_auth.py

@@ -141,6 +141,23 @@ def test_caching(
                 expected_chache_misses += 1
             assert oidc_mock.mocks["token_endpoint"].call_count == expected_chache_misses
 
+    def test_invalidate_cache(self, credentials, oidc_mock: OidcMock):
+        """invalidate_cache() forces a fresh token fetch on the next get_access_token() call."""
+        helper = ClientCredentialsAccessTokenHelper(credentials=credentials)
+
+        token1 = helper.get_access_token()
+        assert oidc_mock.mocks["token_endpoint"].call_count == 1
+
+        # Still cached — no new OIDC request.
+        assert helper.get_access_token() == token1
+        assert oidc_mock.mocks["token_endpoint"].call_count == 1
+
+        # Invalidate; next call must fetch a fresh token.
+        helper.invalidate_cache()
+        token2 = helper.get_access_token()
+        assert oidc_mock.mocks["token_endpoint"].call_count == 2
+        assert token2 == oidc_mock.state["access_token"]
+
     @pytest.mark.skip(reason="Logging was removed for eu-cdse/openeo-cdse-infra#476")
     def test_secret_logging(self, credentials, oidc_mock: OidcMock, caplog):
         """Check that secret is not logged"""