fix: Skip GCS configuration download in GitHub Actions environment (#279)

During GitHub Actions (CI) pipeline execution, native Airflow methods and cloud-dependent functions cause blocking errors due to the lack of an active database connection or cloud permissions.

Author: Jim-0518

dags/common/quarantined_tests.py

@@ -54,22 +54,36 @@ def match_quarantine_patterns(
   return False
 
 
+def is_ci_environment() -> bool:
+  """Checks if running in a CI environment.
+
+  Identifies environments where framework-specific runtime dependencies do not
+  exist. For example, GitHub Actions lacks the actual Airflow environment and
+  its components (such as database context or cloud credentials), requiring code
+  to bypass these dependencies and use mocks instead.
+
+  Returns:
+    bool: True if in CI (GitHub Actions), False otherwise.
+  """
+  return os.getenv("GITHUB_ACTIONS", "false").lower() == "true"
+
+
 def safe_get_from_variable(key: str, default_var: str):
   """
-  Check whether the current runtime is GitHub Actions. Skip retrieving variables in GitHub Actions to avoid excessive log output.
+  Check whether the current runtime is GitHub Actions. Skip retrieving variables
+  in GitHub Actions to avoid excessive log output.
   """
   value = default_var
-  is_ci_env = os.getenv("GITHUB_ACTIONS", "false").lower() == "true"
-  if is_ci_env:
+  if is_ci_environment():
     logging.info("In GitHub Actions, skip getting variables")
   else:
     value = Variable.get(key, default_var=default_var)
   return value
 
 
 """
-The quarantine list is defined by a set of UNIX Shell Glob Patterns. 
-These patterns are used to match and quarantine tests. 
+The quarantine list is defined by a set of UNIX Shell Glob Patterns.
+These patterns are used to match and quarantine tests.
 
 The patterns are stored in the Airflow Variable named 'quarantine_patterns'.
 
@@ -90,9 +104,10 @@ def is_quarantined(test_name) -> bool:
     """
     Checks if a test is quarantined using both legacy and current methods.
 
-    The legacy method checks if `test_name` is present in `QuarantineTests.tests`.
-    The current method checks against a runtime quarantine list fetched from Airflow Variables
-    (key: 'quarantine_list') using `is_in_runtime_quarantine_list()`.
+    The legacy method checks if `test_name` is present in
+    `QuarantineTests.tests`. The current method checks against a runtime
+    quarantine list fetched from Airflow Variables (key: 'quarantine_list')
+    using `is_in_runtime_quarantine_list()`.
 
     The test is considered quarantined if it's found by either method.
     """

dags/tpu_observability/utils/jobset_util.py

@@ -282,18 +282,18 @@ class JobSet:
       recreating pods. Defaults to False.
   """
 
-  namespace: str
-  max_restarts: int
-  replicated_job_name: str
-  replicas: int
-  backoff_limit: int
-  completions: int
-  parallelism: int
-  tpu_accelerator_type: str
-  tpu_topology: str
-  container_name: str
-  image: str
-  tpu_cores_per_pod: int
+  namespace: str = ""
+  max_restarts: int = -1
+  replicated_job_name: str = ""
+  replicas: int = -1
+  backoff_limit: int = -1
+  completions: int = -1
+  parallelism: int = -1
+  tpu_accelerator_type: str = ""
+  tpu_topology: str = ""
+  container_name: str = ""
+  image: str = ""
+  tpu_cores_per_pod: int = -1
   privileged: bool = False
   dag_id_prefix: str = ""
   delay_recovery: bool = False

xlml/apis/gcs.py

@@ -14,16 +14,17 @@
 
 """Functions for GCS Bucket"""
 
+from absl import logging
 import os
 import re
 import tempfile
 from typing import List
+import yaml
 
-from absl import logging
 from airflow.decorators import task
 from airflow.hooks.subprocess import SubprocessHook
 from airflow.providers.google.cloud.operators.gcs import GCSHook
-import yaml
+from dags.common.quarantined_tests import is_ci_environment
 
 
 def obtain_file_list(gcs_path: str) -> List[str]:
@@ -97,6 +98,14 @@ def load_yaml_from_gcs(gcs_path: str) -> dict:
   """Loads and parses the DAG configuration YAML file from GCS."""
   logging.info(f"Attempting to load config from: {gcs_path}")
 
+  # Workflows triggered by fork PRs do not have access to upstream secrets.
+  # Bypass GCS loading strictly to allow the CI workflow to pass successfully.
+  if is_ci_environment():
+    logging.info(
+        "In GitHub Actions, skip load_yaml_from_gcs and return an empty dict."
+    )
+    return {}
+
   if not gcs_path.startswith("gs://"):
     raise ValueError(
         f"Invalid GCS path: '{gcs_path}'. Path must start with 'gs://'."
@@ -106,8 +115,8 @@ def load_yaml_from_gcs(gcs_path: str) -> dict:
       gcs_path.lower().endswith(".yaml") or gcs_path.lower().endswith(".yml")
   ):
     logging.warning(
-        f"GCS path '{gcs_path}' does not have a typical YAML extension (.yaml or .yml). "
-        "Proceeding, but be aware this might not be a YAML file."
+        f"GCS path '{gcs_path}' does not have a typical YAML extension (.yaml "
+        "or .yml). Proceeding, but be aware this might not be a YAML file."
     )
 
   with tempfile.TemporaryDirectory() as tmpdir:
@@ -123,8 +132,9 @@ def load_yaml_from_gcs(gcs_path: str) -> dict:
 
     if not os.path.exists(temp_file_path):
       logging.error(
-          f"gcloud storage cp command completed, but '{temp_file_path}' was not created. "
-          "This often means the copy failed. Check gcloud storage stdout/stderr in logs."
+          f"gcloud storage cp command completed, but '{temp_file_path}' "
+          "was not created. This often means the copy failed. "
+          "Check gcloud storage stdout/stderr in logs."
       )
       raise FileNotFoundError(
           f"[Errno 2] Failed to download file from GCS path: {gcs_path}"