Consolidate duplicate functionality (#12)

* Use get_pr_labels function from deploy.py
* Use common code for getting timeout
* Flush output for consistent displaying
* Handle display behavior in a single place
* Ignore rule recommending ternary statement
* Return empty set if there is no PR number
* Use display for error message

files/bin/deploy-iqe-cji.py

@@ -3,46 +3,21 @@
 import argparse
 import json
 import os
-import shlex
 import sys
 import typing as t
-import urllib.request
 
 from functools import cached_property
 from itertools import chain
-from urllib.request import HTTPError
 
-import fuzzydate
 import sh
 
+from deploy import display
+from deploy import get_pr_labels
+from deploy import get_timeout
 from sh import bonfire
 from sh import oc
 
 
-def get_pr_labels(
-    pr_number: str,
-    owner: str = "project-koku",
-    repo: str = "koku",
-) -> set[str]:
-    if not pr_number:
-        set()
-
-    url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}"
-    try:
-        with urllib.request.urlopen(url) as response:
-            data = json.loads(response.read())
-    except HTTPError as exc:
-        sys.exit(f"Error {exc.code} retrieving {exc.url}.")
-
-    labels = {item["name"] for item in data["labels"]}
-
-    return labels
-
-
-def ran(command) -> str:
-    return " ".join(shlex.quote(str(arg)) for arg in command)
-
-
 class IQERunner:
     def __init__(
         self,
@@ -91,7 +66,7 @@ def build_number(self) -> str:
         try:
             build_number = check_run_id[:5]
         except TypeError:
-            print("There was a probelem with {check_run_id=}. Using default value of 1.", flush=True)
+            display("There was a probelem with {check_run_id=}. Using default value of 1.")
             build_number = "1"
 
         return build_number
@@ -154,16 +129,7 @@ def iqe_marker_expression(self) -> str:
 
     @cached_property
     def iqe_cji_timeout(self) -> int:
-        try:
-            timeout = fuzzydate.to_seconds(os.environ.get("IQE_CJI_TIMEOUT", "2h"))
-        except (TypeError, ValueError) as exc:
-            print(f"{exc}. Using default value of 2h")
-            timeout = 2 * 60 * 60
-
-        if "full-run-smoke-tests" in self.pr_labels:
-            timeout = 5 * 60 * 60
-
-        return int(timeout)
+        return get_timeout("IQE_CJI_TIMEOUT", self.pr_labels)
 
     @cached_property
     def pr_labels(self) -> set[str]:
@@ -199,7 +165,7 @@ def run_pod(self) -> str:
             *self.iqe_env_vars_arg,
             "--namespace", self.namespace,
         ]  # fmt: off
-        print(ran(["bonfire"] + command), flush=True)
+        display(["bonfire"] + command)
 
         if self.check:
             sys.exit()
@@ -225,14 +191,14 @@ def check_cji_jobs(self) -> None:
         cji = json.loads(data)
         job_map = cji["status"]["jobMap"]
         if not all(v == "Complete" for v in job_map.values()):
-            print(f"\nSome jobs failed: {job_map}")
+            print(f"\nSome jobs failed: {job_map}", flush=True)
             sys.exit(1)
 
-        print(f"\nAll jobs succeeded: {job_map}")
+        print(f"\nAll jobs succeeded: {job_map}", flush=True)
 
     def run(self) -> None:
         if "ok-to-skip-smokes" in self.pr_labels:
-            print("PR labeled to skip smoke tests")
+            display("PR labeled to skip smoke tests")
             return
 
         if "smokes-required" in self.pr_labels and not any(label.endswith("smoke-tests") for label in self.pr_labels):
@@ -243,11 +209,11 @@ def run(self) -> None:
         try:
             self.follow_logs()
         except sh.TimeoutException:
-            print(f"Test exceeded timeout {self.iqe_cji_timeout}")
+            display(f"Test exceeded timeout {self.iqe_cji_timeout}")
             oc.delete.pod(self.pod, namespace=self.namespace, _ok_code=[0, 1])
         except sh.ErrorReturnCode as exc:
-            print("Test command failed")
-            print(exc)
+            display("Test command failed")
+            display(str(exc))
 
         oc([
             "wait", "--timeout", f"{self.iqe_cji_timeout}s",

files/bin/deploy.py

@@ -102,12 +102,15 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-def get_deploy_timeout(labels: set[str]) -> int:
+def get_timeout(env_var: str, labels: set[str] | None = None) -> int:
     try:
-        timeout = fuzzydate.to_seconds(os.environ.get("DEPLOY_TIMEOUT", "30min"))
+        timeout = fuzzydate.to_seconds(os.environ.get(env_var, "2h"))
     except (TypeError, ValueError) as exc:
-        print(f"{exc}. Using default value of 30min")
-        timeout = 30 * 60
+        print(f"{exc}. Using default value of 2h")
+        timeout = 2 * 60 * 60
+
+    if labels and "full-run-smoke-tests" in labels:
+        timeout = 5 * 60 * 60
 
     return int(timeout)
 
@@ -118,7 +121,7 @@ def get_pr_labels(
     repo: str = "koku",
 ) -> set[str]:
     if not pr_number:
-        set()
+        return set()
 
     url = f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}"
     try:
@@ -132,10 +135,18 @@ def get_pr_labels(
     return labels
 
 
-def log_command(command: t.Sequence[t.Any], no_log_values: t.Sequence[t.Any]) -> None:
+def display(command: str | t.Sequence[t.Any], no_log_values: t.Sequence[t.Any] | None = None) -> None:
+    if isinstance(command, str):
+        quoted = [command]
+    else:
+        quoted = [shlex.quote(str(arg)) for arg in command]
+
+    if no_log_values is None:
+        print(" ".join(quoted), flush=True)
+        return
+
     sanitized = []
     redacted = "*" * 8
-    quoted = [shlex.quote(str(arg)) for arg in command]
     for arg in quoted:
         for value in no_log_values:
             if value in arg:
@@ -160,7 +171,7 @@ def main() -> None:
     components_with_resources = os.environ.get("COMPONENTS_W_RESOURCES", "").split()
     components_with_resources_arg = chain.from_iterable(("--no-remove-resources", component) for component in components_with_resources)
     deploy_frontends = os.environ.get("DEPLOY_FRONTENDS") or "false"
-    deploy_timeout = get_deploy_timeout(labels)
+    deploy_timeout = get_timeout("DEPLOY_TIMEOUT", labels)
     extra_deploy_args = os.environ.get("EXTRA_DEPLOY_ARGS", "")
     optional_deps_method = os.environ.get("OPTIONAL_DEPS_METHOD", "hybrid")
     ref_env = os.environ.get("REF_ENV", "insights-production")
@@ -212,7 +223,7 @@ def main() -> None:
         print("PR labeled to skip smoke tests")
         return
 
-    log_command(command, no_log_values)
+    display(command, no_log_values)
 
     subprocess.check_call(command, env=os.environ | {"BONFIRE_NS_REQUESTER": requester})
 

ruff.toml

@@ -24,13 +24,14 @@ select = [
     ]
 ignore = [
     # "E501",     # line-too-long
+    "E203",
+    "PERF203",  # try-except-in-loop
     "PLR0913",  # pylint.max-args
     "PLR0914",  # pylint.max-locals
     "PLR0915",  # pylint.max-statements
     "PLR2004",  # pylint.magic-value-comparison
     "PLW1510",  # pylint.subprocess-run-without-check
-    "E203",
-    "PERF203",  # try-except-in-loop
+    "SIM108",   # if-else-block-instead-of-if-exp
     "T201",     # print
     ]