[BDC-488] Add helper to check for valid token

fence/job/visa_update_cronjob.py

@@ -123,6 +123,12 @@ async def producer(self, db_session, queue, chunk_idx):
             if users == None:
                 break
             for user in users:
+                if not self._has_valid_refresh_token(user, db_session):
+                    self.logger.info(
+                        f"Skipping user {user.username}: no valid refresh token"
+                    )
+                    # do not add to queue if no valid token
+                    continue
                 self.logger.info("Producer producing user {}".format(user.username))
                 await queue.put(user)
             if len(users) < chunk_size:
@@ -174,6 +180,23 @@ async def updater(self, name, updater_queue, db_session):
 
             updater_queue.task_done()
 
+    def _has_valid_refresh_token(self, user, db_session):
+        """
+        Return True if user has at least one non‑expired refresh token.
+        Deletes any expired tokens
+        """
+        valid = False
+
+        for row in list(user.upstream_refresh_tokens):
+            if row.expires > time.time():
+                valid = True
+            else:
+                # remove expired token
+                db_session.delete(row)
+                db_session.commit()
+
+        return valid
+
     def _pick_client(self, user):
         """
         Pick oidc client according to the identity provider

tests/dbgap_sync/conftest.py

@@ -29,6 +29,8 @@
     GA4GHVisaV1,
     create_user,
     User,
+    UpstreamRefreshToken,
+    query_for_user,
 )
 
 from tests.conftest import random_txn
@@ -362,6 +364,15 @@ def fake_ras_login(username, subject, email=None, db_session=None):
     logger.debug(
         f"subject: {subject}, username: {username}, actual_username: {actual_username}"
     )
+    user = query_for_user(db_session, actual_username)
+    refresh = UpstreamRefreshToken(
+        user=user,
+        refresh_token=f"fake-refresh-{subject}",
+        expires=int(time.time()) + 3600,  # valid for 1 hour
+    )
+    db_session.add(refresh)
+    db_session.commit()
+
     login_user(actual_username, provider="ras", email=None, id_from_idp=subject)
 
     # todo sub to iss table