Merge pull request #581 from okurz/fix/style_trigger_bisect

fix(openqa-trigger-bisect-jobs): refactor for lint compliance

Author: mergify[bot]

openqa-trigger-bisect-jobs

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # Copyright SUSE LLC
-# ruff: noqa: ANN001, ANN201, B005, C901, G001, G002, G003, G004, PLR0911, PLR0916, PLW1510, PYI063, S113, S324, S404, S603, TRY401, FBT002, SIM112, UP031
-
+# ruff: noqa: ANN001, ANN201, B005, PLW1510, PYI063, S113, S324, S404, S603, TRY401, FBT002, SIM112
 """Trigger openQA tests bisecting maintenance updates.
 
 Determine which pending operating system maintenance update, called "incident",
@@ -10,6 +9,8 @@ Can be called manually and is also called by openQA hook scripts, see
 http://open.qa/docs/#_enable_custom_hook_scripts_on_job_done_based_on_result.
 """
 
+from __future__ import annotations
+
 import argparse
 import hashlib
 import json
@@ -38,12 +39,16 @@ class CustomFormatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescri
 
 @total_ordering
 class Incident:
+    """Maintenance incident representation."""
+
     def __init__(self, inc: str) -> None:
+        """Initialize incident from string."""
         self.incident = inc
         self._incident_id = None
 
     @property
-    def incident_id(self):
+    def incident_id(self) -> str:
+        """Return the incident ID."""
         if self._incident_id:
             return self._incident_id
 
@@ -56,22 +61,32 @@ class Incident:
         return self._incident_id
 
     def __str__(self) -> str:
+        """Return the incident string."""
         return self.incident
 
-    def __eq__(self, __o) -> bool:
+    def __eq__(self, __o: object) -> bool:
+        """Return whether two incidents are equal."""
+        if not isinstance(__o, Incident):
+            return NotImplemented
         return self.incident_id == __o.incident_id
 
-    def __gt__(self, __o) -> bool:
+    def __gt__(self, __o: object) -> bool:
+        """Return whether this incident ID is greater than the other."""
+        if not isinstance(__o, Incident):
+            return NotImplemented
         return int(self.incident_id) > int(__o.incident_id)
 
     def __hash__(self) -> int:
+        """Return the hash of the incident."""
         return int(hashlib.md5(self.incident.encode()).hexdigest(), base=16)
 
     def __repr__(self) -> str:
+        """Return the representation of the incident."""
         return f"<Incident -> {self.incident}"
 
 
 def parse_args():
+    """Parse command line arguments."""
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=CustomFormatter)
     parser.add_argument(
         "-v",
@@ -99,7 +114,8 @@ def parse_args():
         3: logging.INFO,
         4: logging.DEBUG,
     }
-    logging_level = logging.DEBUG if args.verbose > 4 else verbose_to_log[args.verbose]
+    max_verbose = 4
+    logging_level = logging.DEBUG if args.verbose > max_verbose else verbose_to_log[args.verbose]
     log.setLevel(logging_level)
     return args
 
@@ -113,20 +129,23 @@ client_args = [
 
 
 def call(cmds, dry_run=False):
-    log.debug("call: {}".format(cmds))
+    """Call a command and return stdout."""
+    log.debug("call: %s", cmds)
     res = subprocess.run((["echo", "Simulating: "] if dry_run else []) + cmds, capture_output=True)
     if len(res.stderr):
-        log.warning(f"call() {cmds[0]} stderr: {res.stderr}")
+        log.warning("call() %s stderr: %s", cmds[0], res.stderr)
     res.check_returncode()
     return res.stdout.decode("utf-8")
 
 
 def openqa_comment(job, host, comment, dry_run):
+    """Post a comment to an openQA job."""
     args = [*client_args, "--host", host, "-X", "POST", "jobs/" + str(job) + "/comments", "text=" + comment]
     return call(args, dry_run)
 
 
 def openqa_set_job_prio(job_id, host, prio, dry_run):
+    """Set the priority of an openQA job."""
     prio_json = json.dumps({"priority": prio})
     args = [*client_args, "--host", host, "--json", "--data", prio_json, "-X", "PUT", "jobs/" + str(job_id)]
     return call(args, dry_run)
@@ -138,6 +157,7 @@ def openqa_clone(
     default_opts=None,
     default_cmds=None,
 ):
+    """Clone an openQA job."""
     if default_cmds is None:
         default_cmds = ["_GROUP=0"]
     if default_opts is None:
@@ -146,23 +166,25 @@ def openqa_clone(
 
 
 def fetch_url(url, request_type="text"):
+    """Fetch content from URL."""
     try:
         content = requests.get(url, headers={"User-Agent": USER_AGENT})
         content.raise_for_status()
     except requests.exceptions.RequestException as e:
-        log.exception("Error while fetching {}: {}".format(url, str(e)))
+        log.exception("Error while fetching %s: %s", url, e)
         raise
     raw = content.content
     if request_type == "json":
         try:
             content = content.json()
         except json.decoder.JSONDecodeError as e:
-            log.exception("Error while decoding JSON from {} -> >>{}<<: {}".format(url, raw, str(e)))
+            log.exception("Error while decoding JSON from %s -> >>%s<<: %s", url, raw, e)
             raise
     return content
 
 
 def find_changed_issues(investigation):
+    """Find changed issues in investigation text."""
     changes = {}
     pattern = re.compile(r"(?P<diff>[+-])\s+\"(?P<key>[A-Z]+_TEST_(?:ISSUES|REPOS))\"\s*:\s*\"(?P<var>[^\"]*)\",")
 
@@ -178,6 +200,11 @@ def find_changed_issues(investigation):
         if not changes[key].get(BAD) or not changes[key].get(GOOD):
             del changes[key]
             continue
+        removed_key, added_key = (
+            list(changes[key][GOOD] - changes[key][BAD]),
+            list(changes[key][BAD] - changes[key][GOOD]),
+        )
+        log.debug("[%s] removed: %s, added: %s", key, removed_key, added_key)
         if len(changes[key][BAD]) <= 1:
             # no value in triggering single-incident bisections
             del changes[key]
@@ -188,119 +215,134 @@ def find_changed_issues(investigation):
     return changes_repos or changes
 
 
-def main(args) -> None:
-    parsed_url = urlparse(args.url)
-    base_url = urlunparse((parsed_url.scheme, parsed_url.netloc, "", "", "", ""))
-    job_id = parsed_url.path.lstrip("/tests/")
-    test_url = f"{base_url}/api/v1/jobs/{job_id}"
-    log.debug("Retrieving job data from {}".format(test_url))
-    test_data = fetch_url(test_url, request_type="json")
-    job = test_data["job"]
+def _check_skip_bisection(job: dict) -> bool:
+    """Return whether bisection should be skipped for this job."""
     if job["result"] == "passed":
-        log.info("Job %d (%s) is passed, skipping bisection" % (job["id"], job["test"]))
-        return
-    search = re.search(r":investigate:", job["test"])
-    if search:
-        log.info("Job %d (%s) is already an investigation, skipping bisection" % (job["id"], job["test"]))
-        return
+        log.info("Job %d (%s) is passed, skipping bisection", job["id"], job["test"])
+        return True
+    if r":investigate:" in job["test"]:
+        log.info("Job %d (%s) is already an investigation, skipping bisection", job["id"], job["test"])
+        return True
     if job.get("clone_id") is not None:
-        log.info("Job %d already has a clone, skipping bisection" % job["id"])
-        return
+        log.info("Job %d already has a clone, skipping bisection", job["id"])
+        return True
 
-    children = job.get("children", [])
-    parents = job.get("parents", [])
-    if (
+    children = job.get("children", {})
+    parents = job.get("parents", {})
+    return bool(
         ("Parallel" in children and len(children["Parallel"]))
         or ("Directly chained" in children and len(children["Directly chained"]))
         or ("Parallel" in parents and len(parents["Parallel"]))
         or ("Directly chained" in parents and len(parents["Directly chained"]))
-    ):
-        return
+    )
 
-    investigation_url = f"{base_url}/tests/{job_id}/investigation_ajax"
-    log.debug("Retrieving investigation info from {}".format(investigation_url))
-    investigation = fetch_url(investigation_url, request_type="json")
-    log.debug("Received investigation info: {}".format(investigation))
-    if "diff_to_last_good" not in investigation:
-        return
-    all_changes = find_changed_issues(investigation["diff_to_last_good"])
-
-    if not all_changes:
-        return
 
+def _check_exclude_regex(job: dict) -> bool:
+    """Return whether the job matches exclude regex from environment."""
     exclude_group_regex = os.environ.get("exclude_group_regex", "")
     if len(exclude_group_regex) > 0:
         full_group = job.get("group", "")
         if "parent_group" in job:
             full_group = "{} / {}".format(job["parent_group"], full_group)
         if re.search(exclude_group_regex, full_group):
-            log.debug("job group '{}' matches 'exclude_group_regex', skipping".format(full_group))
-            return
+            log.debug("job group '%s' matches 'exclude_group_regex', skipping", full_group)
+            return True
 
     exclude_name_regex = os.environ.get("exclude_name_regex", "")
     if len(exclude_name_regex) > 0 and re.search(exclude_name_regex, job["test"]):
-        log.debug("job name '{}' matches 'exclude_name_regex', skipping".format(job["test"]))
-        return
+        log.debug("job name '%s' matches 'exclude_name_regex', skipping", job["test"])
+        return True
+    return False
+
+
+def _trigger_bisect_job(args, base_url, job_id, test_name, line) -> str:
+    """Trigger a single bisection job."""
+    params = (
+        [args.url]
+        + [f"{k}={v}" for k, v in line.items()]
+        + [
+            f"TEST={test_name}",
+            f"OPENQA_INVESTIGATE_ORIGIN={args.url}",
+            "MAINT_TEST_REPO=",
+        ]
+    )
 
-    log.debug("Received job data: {}".format(test_data))
-    test = job["settings"]["TEST"]
-    prio = int(job["priority"]) + args.priority_add
-    log.debug("Found test name '{}'".format(test))
+    try:
+        out = openqa_clone(params, args.dry_run)
+    except subprocess.SubprocessError as err:
+        if "the repositories for the below updates are unavailable" in str(err.stderr):
+            extra_comment = "Not triggering any bisect jobs because: "
+            openqa_comment(job_id, base_url, f"{extra_comment}{err.stderr}", args.dry_run)
+            sys.exit(0)
+        raise
+    return out
 
-    created = ""
+
+def _process_bisections(args, base_url, job_id, all_changes, test, prio) -> str:  # noqa: PLR0913, PLR0917
+    """Process all potential bisections."""
     added = []
     for key in all_changes:
-        changes = all_changes[key]
-        removed_key, added_key = list(changes[GOOD] - changes[BAD]), list(changes[BAD] - changes[GOOD])
-        log.debug("[{}] removed: {}, added: {}".format(key, removed_key, added_key))
-        added += added_key
+        added += list(all_changes[key][BAD] - all_changes[key][GOOD])
 
+    created = ""
     # whole sort is to simplify testability of code
     for issue in sorted({i.incident_id for i in added}, key=int):
         line = {}
-        log.info("Triggering one bisection job without issue '{}'".format(issue))
+        log.info("Triggering one bisection job without issue '%s'", issue)
         for key in all_changes:
             # use only VARS where is incident present in BAD
             if [i for i in all_changes[key][BAD] if i.incident_id == issue]:
                 line[key] = ",".join(str(i) for i in sorted(all_changes[key][BAD]) if i.incident_id != issue)
-                log.debug("New set of {}='{}'".format(key, line[key]))
-
-        test_name = test + ":investigate:bisect_without_{}".format(issue)
-        params = (
-            [args.url]
-            + [k + "=" + v for k, v in line.items()]
-            + [
-                "TEST=" + test_name,
-                "OPENQA_INVESTIGATE_ORIGIN=" + args.url,
-                "MAINT_TEST_REPO=",
-            ]
-        )
+                log.debug("New set of %s='%s'", key, line[key])
 
-        try:
-            out = openqa_clone(
-                params,
-                args.dry_run,
-            )
-        except subprocess.SubprocessError as err:
-            if "the repositories for the below updates are unavailable" in str(err.stderr):
-                extra_comment = "Not triggering any bisect jobs because: "
-                openqa_comment(job_id, base_url, f"{extra_comment}{err.stderr}", args.dry_run)
-                sys.exit(0)
-            else:
-                raise
+        test_name = f"{test}:investigate:bisect_without_{issue}"
+        out = _trigger_bisect_job(args, base_url, job_id, test_name, line)
 
         created_job_ids = []
         try:
             created_job_ids = json.loads(out).values()
         except Exception:
-            log.exception("openqa-clone-job returned non-JSON output: " + out)
-        for job_id in sorted(created_job_ids):
-            log.info("Created %s", job_id)
-            created += f"* **{test_name}**: {base_url}/t{job_id}\n"
-            openqa_set_job_prio(job_id, args.url, prio, args.dry_run)
-
-    if len(created):
-        comment = "Automatic bisect jobs:\n\n" + created
+            log.exception("openqa-clone-job returned non-JSON output: %s", out)
+        for created_id in sorted(created_job_ids):
+            log.info("Created %s", created_id)
+            created += f"* **{test_name}**: {base_url}/t{created_id}\n"
+            openqa_set_job_prio(created_id, args.url, prio, args.dry_run)
+    return created
+
+
+def main(args) -> None:
+    """Execute main function."""
+    parsed_url = urlparse(args.url)
+    base_url = urlunparse((parsed_url.scheme, parsed_url.netloc, "", "", "", ""))
+    job_id = parsed_url.path.lstrip("/tests/")
+    test_url = f"{base_url}/api/v1/jobs/{job_id}"
+    log.debug("Retrieving job data from %s", test_url)
+    test_data = fetch_url(test_url, request_type="json")
+    job = test_data["job"]
+
+    if _check_skip_bisection(job) or _check_exclude_regex(job):
+        return
+
+    investigation_url = f"{base_url}/tests/{job_id}/investigation_ajax"
+    log.debug("Retrieving investigation info from %s", investigation_url)
+    investigation = fetch_url(investigation_url, request_type="json")
+    log.debug("Received investigation info: %s", investigation)
+    if "diff_to_last_good" not in investigation:
+        return
+    all_changes = find_changed_issues(investigation["diff_to_last_good"])
+
+    if not all_changes:
+        return
+
+    log.debug("Received job data: %s", test_data)
+    test = job["settings"]["TEST"]
+    prio = int(job["priority"]) + args.priority_add
+    log.debug("Found test name '%s'", test)
+
+    created = _process_bisections(args, base_url, job_id, all_changes, test, prio)
+
+    if created:
+        comment = f"Automatic bisect jobs:\n\n{created}"
         openqa_comment(job["id"], base_url, comment, args.dry_run)
 
 

tests/test_trigger_bisect_jobs.py

@@ -246,7 +246,7 @@ def test_problems() -> None:
     args.url = "http://openqa.opensuse.org/tests/101"
     openqa.log.info = MagicMock()
     openqa.main(args)
-    openqa.log.info.assert_called_with("Job 101 (foo) is passed, skipping bisection")
+    openqa.log.info.assert_called_with("Job %d (%s) is passed, skipping bisection", 101, "foo")
     assert (
         call(
             "http://openqa.opensuse.org/tests/101/investigation_ajax",