Add option to validate if a token will expire in the near future

snakemake_executor_plugin_auth_tes/__init__.py

@@ -202,7 +202,7 @@ def tes_access_token(self):
         if not self.do_oidc_auth:
             return self.workflow.executor_settings.token
 
-        if self.auth_client.is_token_expired(self._access_token):
+        if self.auth_client.is_token_expired(self._access_token, 300):
             refresh_result = self.auth_client.refresh_access_token(self._refresh_token)
 
             self._access_token = refresh_result["access_token"]

snakemake_executor_plugin_auth_tes/auth.py

@@ -21,13 +21,17 @@ def __init__(self, client_id, client_secret, oidc_url):
             self.client_id, self.client_secret
         )
 
-    def is_token_expired(self, token):
+    def is_token_expired(self, token, time_offset=0):
         jwks_client = jwt.PyJWKClient(self.jwks_url)
         header = jwt.get_unverified_header(token)
         key = jwks_client.get_signing_key(header["kid"]).key
 
         try:
-            jwt.decode(token, key, [header["alg"]], options={"verify_aud": False})
+            data = jwt.decode(
+                token, key, [header["alg"]], options={"verify_aud": False}
+            )
+            if data["exp"] - time_offset:
+                return True
         except jwt.ExpiredSignatureError:
             return True