fix(tests): Add error handling and environment variable validation to test files

tests/interop/create_ci_badge.py

@@ -1,11 +1,14 @@
 import json
 import os
 import subprocess
+import sys
 from datetime import datetime
 
 from junitparser import JUnitXml
 
-oc = os.environ["HOME"] + "/oc_client/oc"
+# Use os.environ.get() with fallback to avoid KeyError
+home_dir = os.environ.get("HOME", "/tmp")
+oc = os.path.join(home_dir, "oc_client", "oc")
 
 ci_badge = {
     "schemaVersion": 1,
@@ -24,19 +27,36 @@
 
 
 def get_openshift_version():
+    """Get OpenShift version from cluster.
+
+    Returns:
+        tuple: (full_version, major_minor) on success
+        None: on any error
+    """
     try:
-        version_ret = subprocess.run([oc, "version", "-o", "json"], capture_output=True)
+        version_ret = subprocess.run(
+            [oc, "version", "-o", "json"],
+            capture_output=True,
+            check=False
+        )
+        if version_ret.returncode != 0:
+            print(f"Error running oc version: {version_ret.stderr.decode('utf-8')}")
+            return None
         version_out = version_ret.stdout.decode("utf-8")
         openshift_version = json.loads(version_out)["openshiftVersion"]
         major_minor = ".".join(openshift_version.split(".")[:-1])
         return openshift_version, major_minor
-    except KeyError as e:
-        print("KeyError:" + str(e))
+    except (KeyError, json.JSONDecodeError, UnicodeDecodeError, OSError) as e:
+        print(f"Error getting OpenShift version: {type(e).__name__}: {e}")
         return None
 
 
 if __name__ == "__main__":
     versions = get_openshift_version()
+    if versions is None:
+        print("Failed to get OpenShift version, exiting")
+        sys.exit(1)
+
     ci_badge["openshiftVersion"] = versions[0]
 
     pattern_repo = subprocess.run(
@@ -51,12 +71,20 @@ def get_openshift_version():
 
     # Check each xml file for failures
     results_dir = os.environ.get("WORKSPACE")
+    if results_dir is None:
+        print("WORKSPACE environment variable is not set, exiting")
+        sys.exit(1)
+
+    if not os.path.isdir(results_dir):
+        print(f"WORKSPACE directory does not exist: {results_dir}")
+        sys.exit(1)
+
     failures = 0
 
     for file in os.listdir(results_dir):
         if file.startswith("test_") and file.endswith(".xml"):
-            with open(os.path.join(results_dir, file), "r") as result_file:  # type: ignore
-                xml = JUnitXml.fromfile(result_file)  # type: ignore
+            with open(os.path.join(results_dir, file), "r") as result_file:
+                xml = JUnitXml.fromfile(result_file)
                 for suite in xml:
                     for case in suite:
                         if case.result:
@@ -69,15 +97,26 @@ def get_openshift_version():
     # For now we assume `message` is the same as patternBranch
     ci_badge["message"] = ci_badge["patternBranch"]
 
+    # Validate required environment variables for filename
+    pattern_shortname = os.environ.get("PATTERN_SHORTNAME")
+    infra_provider = os.environ.get("INFRA_PROVIDER")
+
+    if not pattern_shortname:
+        print("PATTERN_SHORTNAME environment variable is not set, exiting")
+        sys.exit(1)
+    if not infra_provider:
+        print("INFRA_PROVIDER environment variable is not set, exiting")
+        sys.exit(1)
+
     ci_badge_json_basename = (
-        os.environ.get("PATTERN_SHORTNAME")  # type: ignore
+        pattern_shortname
         + "-"
-        + os.environ.get("INFRA_PROVIDER")
+        + infra_provider
         + "-"
         + versions[1]
         + "-stable-badge.json"
     )
-    ci_badge_json_filename = os.path.join(results_dir, ci_badge_json_basename)  # type: ignore
+    ci_badge_json_filename = os.path.join(results_dir, ci_badge_json_basename)
     print(f"Creating CI badge file at: {ci_badge_json_filename}")
 
     with open(ci_badge_json_filename, "w") as ci_badge_file:

tests/interop/test_modify_web_content.py

@@ -14,10 +14,21 @@
 
 logger = logging.getLogger(__loggername__)
 
+# Configurable timeout settings (can be overridden via environment)
+CONTENT_UPDATE_TIMEOUT_MINUTES = int(os.environ.get("CONTENT_UPDATE_TIMEOUT_MINUTES", "10"))
+CONTENT_UPDATE_POLL_SECONDS = int(os.environ.get("CONTENT_UPDATE_POLL_SECONDS", "30"))
+
+# Configurable repository path (can be overridden via environment)
+PATTERNS_REPO_PATH = os.environ.get(
+    "PATTERNS_REPO_PATH",
+    os.path.join(os.environ.get("HOME", ""), "validated_patterns/multicloud-gitops")
+)
+
 
 @pytest.mark.modify_web_content
 def test_modify_web_content(openshift_dyn_client):
     logger.info("Find the url for the hello-world route")
+    route = None
     try:
         for route in Route.get(
             dyn_client=openshift_dyn_client,
@@ -30,15 +41,19 @@ def test_modify_web_content(openshift_dyn_client):
         logger.error(f"FAIL: {err_msg}")
         assert False, err_msg
 
+    if route is None:
+        err_msg = "No route found for hello-world in hello-world namespace"
+        logger.error(f"FAIL: {err_msg}")
+        assert False, err_msg
+
     url = "http://" + route.instance.spec.host
     response = requests.get(url)
     logger.info(f"Current page content: {response.content}")
 
     if os.getenv("EXTERNAL_TEST") != "true":
-        chart = (
-            f"{os.environ['HOME']}"
-            + "/validated_patterns/multicloud-gitops/charts/"
-            + "all/hello-world/templates/hello-world-cm.yaml"
+        chart = os.path.join(
+            PATTERNS_REPO_PATH,
+            "charts/all/hello-world/templates/hello-world-cm.yaml"
         )
     else:
         chart = "../../charts/all/hello-world/templates/hello-world-cm.yaml"
@@ -51,37 +66,52 @@ def test_modify_web_content(openshift_dyn_client):
     )
 
     logger.info("Merge the change")
-    patterns_repo = f"{os.environ['HOME']}/validated_patterns/multicloud-gitops"
+    patterns_repo = PATTERNS_REPO_PATH
     if os.getenv("EXTERNAL_TEST") != "true":
-        subprocess.run(["git", "add", chart], cwd=f"{patterns_repo}")
-        subprocess.run(
-            ["git", "commit", "-m", "Updating 'hello-world'"], cwd=f"{patterns_repo}"
+        git_add = subprocess.run(["git", "add", chart], cwd=patterns_repo, capture_output=True, text=True)
+        if git_add.returncode != 0:
+            logger.error(f"git add failed: {git_add.stderr}")
+
+        git_commit = subprocess.run(
+            ["git", "commit", "-m", "Updating 'hello-world'"], cwd=patterns_repo, capture_output=True, text=True
         )
+        if git_commit.returncode != 0:
+            logger.warning(f"git commit returned non-zero: {git_commit.stderr}")
+
         push = subprocess.run(
-            ["git", "push"], cwd=f"{patterns_repo}", capture_output=True, text=True
+            ["git", "push"], cwd=patterns_repo, capture_output=True, text=True
         )
     else:
-        subprocess.run(["git", "add", chart])
-        subprocess.run(["git", "commit", "-m", "Updating 'hello-world'"])
+        git_add = subprocess.run(["git", "add", chart], capture_output=True, text=True)
+        if git_add.returncode != 0:
+            logger.error(f"git add failed: {git_add.stderr}")
+
+        git_commit = subprocess.run(["git", "commit", "-m", "Updating 'hello-world'"], capture_output=True, text=True)
+        if git_commit.returncode != 0:
+            logger.warning(f"git commit returned non-zero: {git_commit.stderr}")
+
         push = subprocess.run(["git", "push"], capture_output=True, text=True)
+
+    if push.returncode != 0:
+        logger.error(f"git push failed with return code {push.returncode}")
     logger.info(push.stdout)
     logger.info(push.stderr)
 
     logger.info("Checking for updated page content")
-    timeout = time.time() + 60 * 10
+    timeout = time.time() + 60 * CONTENT_UPDATE_TIMEOUT_MINUTES
+    new_content = None
     while time.time() < timeout:
-        time.sleep(30)
+        time.sleep(CONTENT_UPDATE_POLL_SECONDS)
         response = requests.get(url)
         logger.info(response.content)
 
         new_content = re.search(new_heading, str(response.content))
 
         logger.info(new_content)
-        if (new_content is None) or (new_content.group() != new_heading):
-            continue
-        break
+        if new_content is not None and new_content.group() == new_heading:
+            break
 
-    if (new_content is None) or (new_content.group() != new_heading):
+    if new_content is None or new_content.group() != new_heading:
         err_msg = "Did not find updated page content"
         logger.error(f"FAIL: {err_msg}")
         assert False, err_msg