RHOAIENG-69301: fix disconnected env job submission

Author: pawelpaszki

tests/e2e/support.py

@@ -157,6 +157,160 @@ def get_setup_env_variables(**kwargs):
     return env_vars
 
 
+def _env_flag_enabled(name):
+    return os.environ.get(name, "").strip().lower() in ("1", "true", "yes")
+
+
+def _env_flag_disabled(name):
+    return os.environ.get(name, "").strip().lower() in ("0", "false", "no", "off")
+
+
+def _has_registry_mirror_configured():
+    """
+    Detect mirror-only registry layout typical of disconnected OpenShift installs.
+
+    Checks ImageDigestMirrorSet (OCP 4.14+) and ImageContentSourcePolicy (legacy).
+    """
+    try:
+        from kubernetes import client as k8s_client
+
+        custom_api = k8s_client.CustomObjectsApi()
+        mirror_types = [
+            ("config.openshift.io", "v1", "imagedigestmirrorsets"),
+            ("operator.openshift.io", "v1alpha1", "imagecontentsourcepolicies"),
+        ]
+        for group, version, plural in mirror_types:
+            try:
+                result = custom_api.list_cluster_custom_object(
+                    group=group,
+                    version=version,
+                    plural=plural,
+                )
+                if result.get("items"):
+                    return True
+            except Exception:
+                continue
+    except Exception:
+        pass
+    return False
+
+
+def _disconnected_cluster_signals():
+    """Return True when the cluster is likely disconnected / air-gapped."""
+    if _env_flag_enabled("DISCONNECTED_CLUSTER") or _env_flag_enabled(
+        "IS_DISCONNECTED_CLUSTER"
+    ):
+        return True
+    if _has_registry_mirror_configured():
+        return True
+    try:
+        server = (run_oc_command(["whoami", "--show-server=true"]) or "").lower()
+        if "-dis-" in server or "disconnected" in server:
+            return True
+    except Exception:
+        pass
+    return False
+
+
+def _upgrade_mnist_prerequisites_met():
+    """
+    Return True when full MNIST can run (pip packages + dataset reachable).
+
+    Connected labs may use public PyPI and MNIST mirrors with no extra env.
+    Disconnected labs need internal PIP_INDEX_URL and AWS_DEFAULT_ENDPOINT (MinIO).
+    """
+    if not _disconnected_cluster_signals():
+        return True
+
+    pip_url = (os.environ.get("PIP_INDEX_URL") or "").strip()
+    aws_endpoint = (os.environ.get("AWS_DEFAULT_ENDPOINT") or "").strip()
+    pip_ok = bool(pip_url) and "pypi.org" not in pip_url
+    aws_ok = bool(aws_endpoint)
+    if pip_ok and aws_ok:
+        print(
+            "Disconnected cluster with PIP mirror and S3 endpoint configured; "
+            "using full MNIST upgrade job"
+        )
+        return True
+    return False
+
+
+def use_upgrade_smoke_job():
+    """
+    Use a lightweight Ray job for upgrade tests when full MNIST is not viable.
+
+    Full MNIST pulls torch/pytorch via pip and may download datasets from the
+    public internet — both fail on disconnected clusters without mirrors.
+
+    Detection order (first match wins):
+      1. UPGRADE_USE_SMOKE_JOB=true|false (explicit override)
+      2. Full MNIST prerequisites met (connected, or disconnected with mirrors)
+      3. DISCONNECTED_CLUSTER / IS_DISCONNECTED_CLUSTER env (Jenkins)
+      4. ImageDigestMirrorSet / ImageContentSourcePolicy on cluster
+      5. API server URL heuristic (-dis- / disconnected), last resort
+    """
+    if _env_flag_enabled("UPGRADE_USE_SMOKE_JOB"):
+        print("UPGRADE_USE_SMOKE_JOB enabled; using upgrade smoke job")
+        return True
+    if _env_flag_disabled("UPGRADE_USE_SMOKE_JOB"):
+        print("UPGRADE_USE_SMOKE_JOB disabled; using full MNIST upgrade job")
+        return False
+
+    if _upgrade_mnist_prerequisites_met():
+        return False
+
+    if _env_flag_enabled("DISCONNECTED_CLUSTER") or _env_flag_enabled(
+        "IS_DISCONNECTED_CLUSTER"
+    ):
+        print(
+            "Disconnected cluster env set without PIP/S3 mirrors; "
+            "using upgrade smoke job (no pip install)"
+        )
+        return True
+
+    if _has_registry_mirror_configured():
+        print(
+            "Registry mirror detected (ImageDigestMirrorSet/ICSP) without "
+            "PIP/S3 mirrors; using upgrade smoke job (no pip install)"
+        )
+        return True
+
+    try:
+        server = (run_oc_command(["whoami", "--show-server=true"]) or "").lower()
+        if "-dis-" in server or "disconnected" in server:
+            print(
+                "Detected disconnected cluster from API server URL; "
+                "using upgrade smoke job (no pip install)"
+            )
+            return True
+    except Exception:
+        pass
+
+    return False
+
+
+def get_upgrade_job_submission_spec():
+    """
+    Return entrypoint and runtime_env for post-upgrade job submission tests.
+    """
+    if use_upgrade_smoke_job():
+        return {
+            "entrypoint": "python upgrade_job_smoke.py",
+            "runtime_env": {
+                "working_dir": "./tests/e2e/",
+                "env_vars": get_setup_env_variables(),
+            },
+        }
+    return {
+        "entrypoint": "python mnist.py",
+        "runtime_env": {
+            "working_dir": "./tests/e2e/",
+            "pip": "./tests/e2e/mnist_pip_requirements.txt",
+            "env_vars": get_setup_env_variables(),
+        },
+    }
+
+
 def random_choice():
     alphabet = string.ascii_lowercase + string.digits
     return "".join(random.choices(alphabet, k=5))

tests/e2e/upgrade_job_smoke.py

@@ -0,0 +1,16 @@
+# Copyright 2024 IBM, Red Hat
+#
+# Minimal Ray job for upgrade qualification on disconnected clusters.
+# Validates job submission and execution without pip installs or external datasets.
+
+import sys
+
+
+def main() -> int:
+    print("upgrade-job-smoke: job started")
+    print("upgrade-job-smoke: job finished successfully")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/upgrade/01_raycluster_sdk_upgrade_test.py

@@ -166,13 +166,10 @@ def assert_jobsubmit_withoutLogin(self, cluster):
         else:
             api_url = dashboard_url + "/api/jobs/"
 
+        job_spec = get_upgrade_job_submission_spec()
         jobdata = {
-            "entrypoint": "python mnist.py",
-            "runtime_env": {
-                "working_dir": "./tests/e2e/",
-                "pip": "./tests/e2e/mnist_pip_requirements.txt",
-                "env_vars": get_setup_env_variables(),
-            },
+            "entrypoint": job_spec["entrypoint"],
+            "runtime_env": job_spec["runtime_env"],
         }
 
         response = requests.post(
@@ -251,14 +248,13 @@ def assert_jobsubmit_withlogin(self, cluster):
         header = {"Authorization": f"Bearer {auth_token}"}
         client = RayJobClient(address=ray_dashboard, headers=header, verify=False)
 
+        job_spec = get_upgrade_job_submission_spec()
+        print(f"Submitting upgrade job: {job_spec['entrypoint']}")
+
         # Submit the job
         submission_id = client.submit_job(
-            entrypoint="python mnist.py",
-            runtime_env={
-                "working_dir": "./tests/e2e/",
-                "pip": "./tests/e2e/mnist_pip_requirements.txt",
-                "env_vars": get_setup_env_variables(),
-            },
+            entrypoint=job_spec["entrypoint"],
+            runtime_env=job_spec["runtime_env"],
         )
         print(f"Submitted job with ID: {submission_id}")
         done = False
@@ -283,9 +279,10 @@ def assert_jobsubmit_withlogin(self, cluster):
         client.delete_job(submission_id)
 
     def assert_job_completion(self, status):
-        if status == "SUCCEEDED":
-            print(f"Job has completed: '{status}'")
+        status_value = getattr(status, "value", status)
+        if status_value == "SUCCEEDED":
+            print(f"Job has completed: '{status_value}'")
             assert True
         else:
-            print(f"Job has completed: '{status}'")
+            print(f"Job has completed: '{status_value}'")
             assert False

tests/upgrade/README.md

@@ -49,6 +49,23 @@ Post-upgrade job tests use `is_byoidc_cluster_detected()` from `tests/e2e/suppor
 
 On **htpasswd/LDAP** QE clusters (e.g. ods-qe-psi-17), job tests use `oc whoami --show-token=true` with `OCP_ADMIN_USER_*` credentials — not Keycloak password grant.
 
+## Disconnected clusters (post-upgrade job submission)
+
+Full MNIST installs torch via pip and may fetch datasets from the public internet. On disconnected clusters that fails unless mirrors are configured.
+
+Post-upgrade job tests use `get_upgrade_job_submission_spec()` from `tests/e2e/support.py`:
+
+| Job | When |
+|-----|------|
+| **`upgrade_job_smoke.py`** (no pip) | Disconnected / mirror-only cluster without PyPI + S3 env |
+| **`mnist.py`** (full pip + training) | Connected cluster, or disconnected with `PIP_INDEX_URL` (non-pypi.org) **and** `AWS_DEFAULT_ENDPOINT` set |
+
+**Detection order:** `UPGRADE_USE_SMOKE_JOB` override → MNIST prerequisites → `DISCONNECTED_CLUSTER` / `IS_DISCONNECTED_CLUSTER` (Jenkins) → `ImageDigestMirrorSet` / `ImageContentSourcePolicy` → API URL `-dis-` heuristic (last resort).
+
+Set `UPGRADE_USE_SMOKE_JOB=false` to force full MNIST when mirrors are configured. Set `DISCONNECTED_CLUSTER=true` in the test env file for explicit lab config (aligns with Jenkins `IS_DISCONNECTED_CLUSTER`).
+
+**Porting:** apply the same helpers (`use_upgrade_smoke_job`, `upgrade_job_smoke.py`, `get_upgrade_job_submission_spec` usage) to **`main`** (3.5 / `2.x-3.x-upgrade-tests-post-upgrade`) and **`3.3`** upgrade branches.
+
 ## Migration script
 
 `scripts/migration/ray_cluster_migration.py` (keep in sync with [rhoai-upgrade-helpers](https://github.com/red-hat-data-services/rhoai-upgrade-helpers) `main`).