Linting compliance

Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>

Author: nuclearcat

config/docker/data/tast_parser.py

@@ -17,12 +17,12 @@
 # along with this library; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
-from functools import partial
-import subprocess
-import sys
-import os
 import json
+import os
 import pwd
+import subprocess
+import sys
+from functools import partial
 
 FAILED_RUN_FILE = "failed_run"
 STDERR_FILE = "stderr.log"
@@ -69,7 +69,9 @@ def report_lava(test_data):
     if "measurements" in test_data:
         lava_test_set_start(test_data["name"])
         for measurement in test_data["measurements"]:
-            report_lava_test_case(measurement["name"], test_data["result"], measurement)
+            report_lava_test_case(
+                measurement["name"], test_data["result"], measurement
+            )
         lava_test_set_stop(test_data["name"])
     else:
         report_lava_test_case(test_data["name"], test_data["result"])
@@ -178,7 +180,9 @@ def parse_test_results():
                 report_lava_critical("Tast tests run failed, no results")
                 sys.exit(1)
         for test_data in parse_results(results):
-            results_chart = os.path.join(test_data["outDir"], RESULTS_CHART_FILE)
+            results_chart = os.path.join(
+                test_data["outDir"], RESULTS_CHART_FILE
+            )
             if os.path.isfile(results_chart):
                 with open(results_chart, "r") as rc_file:
                     rc_data = json.load(rc_file)

kernelci/api/__init__.py

@@ -12,8 +12,8 @@
 import urllib
 from typing import Dict, Optional, Sequence
 
-from cloudevents.http import CloudEvent
 import requests
+from cloudevents.http import CloudEvent
 from requests.adapters import HTTPAdapter
 from urllib3.util import Retry
 
@@ -22,7 +22,9 @@
 HTTP_ERROR_BODY_SNIPPET = 512
 
 
-def _http_error_body_snippet(response, max_length: int = HTTP_ERROR_BODY_SNIPPET):
+def _http_error_body_snippet(
+    response, max_length: int = HTTP_ERROR_BODY_SNIPPET
+):
     """Return a truncated response body useful for HTTP error logging.
     Also try to extract a more specific error message from the body if it's a JSON
     with a "detail", "error" or "message" field.
@@ -150,7 +152,10 @@ def _get(self, path, params=None):
         session.mount("https://", adapter)
 
         resp = session.get(
-            url, params=params, headers=self.data.headers, timeout=self.data.timeout
+            url,
+            params=params,
+            headers=self.data.headers,
+            timeout=self.data.timeout,
         )
         try:
             resp.raise_for_status()
@@ -190,13 +195,19 @@ def _post(self, path, data=None, params=None, json_data=True):
             # requests.post, but in this case we have to explicitly
             # specify the Content-Type header
             if json_data:
-                headers = self.data.headers | {"Content-Type": "application/json"}
+                headers = self.data.headers | {
+                    "Content-Type": "application/json"
+                }
             else:
                 headers = self.data.headers | {
                     "Content-Type": "application/x-www-form-urlencoded"
                 }
             resp = session.post(
-                url, data, headers=headers, params=params, timeout=self.data.timeout
+                url,
+                data,
+                headers=headers,
+                params=params,
+                timeout=self.data.timeout,
             )
         else:
             # When passing a dict to requests.post, it will
@@ -325,7 +336,9 @@ def _delete(self, path):
         session.mount("http://", adapter)
         session.mount("https://", adapter)
 
-        resp = session.delete(url, headers=self.data.headers, timeout=self.data.timeout)
+        resp = session.delete(
+            url, headers=self.data.headers, timeout=self.data.timeout
+        )
         try:
             resp.raise_for_status()
         except requests.exceptions.HTTPError as err:

kernelci/api/helper.py

@@ -7,9 +7,10 @@
 
 """KernelCI API helpers"""
 
-from typing import Dict
 import json
 import os
+from typing import Dict
+
 import requests
 
 from . import API
@@ -36,7 +37,9 @@ def merge(primary: dict, secondary: dict):
     for key in primary:
         result[key] = primary[key]
         if key in secondary:
-            if isinstance(primary[key], dict) and isinstance(secondary[key], dict):
+            if isinstance(primary[key], dict) and isinstance(
+                secondary[key], dict
+            ):
                 result[key] = merge(primary[key], secondary[key])
             else:
                 result[key] = secondary[key]
@@ -62,7 +65,9 @@ def api(self):
         """API object"""
         return self._api
 
-    def subscribe_filters(self, filters=None, channel="node", promiscuous=False):
+    def subscribe_filters(
+        self, filters=None, channel="node", promiscuous=False
+    ):
         """Subscribe to a channel with some added filters"""
         sub_id = self.api.subscribe(channel, promiscuous)
         self._filters[sub_id] = filters
@@ -140,7 +145,9 @@ def receive_event_node(self, sub_id):
             # Crude (provisional) filtering of non-node events
             if not node:
                 continue
-            if all(self.pubsub_event_filter(sub_id, obj) for obj in [node, event]):
+            if all(
+                self.pubsub_event_filter(sub_id, obj) for obj in [node, event]
+            ):
                 return node, event.get("is_hierarchy")
 
     def _find_container(self, field, node):
@@ -188,7 +195,9 @@ def _is_allowed(self, rules, key, node):
         #   the node on the basis that its rules aren; t fulfilled
         if not base:
             if len(allow) > 0:
-                print(f"rules[{key}]: attribute '{key}' not found in node hierarchy")
+                print(
+                    f"rules[{key}]: attribute '{key}' not found in node hierarchy"
+                )
                 return False
             return True
 
@@ -217,7 +226,9 @@ def _is_allowed(self, rules, key, node):
                     found = True
 
             if not found:
-                print(f"rules[{key}]: {key.capitalize()} missing one of {allow}")
+                print(
+                    f"rules[{key}]: {key.capitalize()} missing one of {allow}"
+                )
                 return False
 
         else:
@@ -280,7 +291,10 @@ def _match_combos(self, node, combos):
         """
         match = None
         for combo in combos:
-            if node["tree"] == combo["tree"] and node["branch"] == combo["branch"]:
+            if (
+                node["tree"] == combo["tree"]
+                and node["branch"] == combo["branch"]
+            ):
                 match = combo
                 break
         return match
@@ -415,15 +429,17 @@ def should_create_node(self, rules, node):
                 rule_major = rules[key]["version"]
                 rule_minor = rules[key]["patchlevel"]
                 if key.startswith("min") and (
-                    (major < rule_major) or (major == rule_major and minor < rule_minor)
+                    (major < rule_major)
+                    or (major == rule_major and minor < rule_minor)
                 ):
                     print(
                         f"rules[{key}]: Version {major}.{minor} older than minimum version "
                         f"({rule_major}.{rule_minor})"
                     )
                     return False
                 if key.startswith("max") and (
-                    (major > rule_major) or (major == rule_major and minor > rule_minor)
+                    (major > rule_major)
+                    or (major == rule_major and minor > rule_minor)
                 ):
                     print(
                         f"rules[{key}]: Version {major}.{minor} newer than maximum version "
@@ -473,7 +489,9 @@ def _is_job_filtered(self, node):
             #   * jobfilter contains the job name suffixed with '+'
             #   * at least one element of the node's 'path' appears in jobilfter
             #     with a '+' suffix
-            for filt in (item.rstrip("+") for item in jobfilter if item.endswith("+")):
+            for filt in (
+                item.rstrip("+") for item in jobfilter if item.endswith("+")
+            ):
                 if filt in node["path"] or filt == node["name"]:
                     return False
 
@@ -483,7 +501,13 @@ def _is_job_filtered(self, node):
 
     # pylint: disable=too-many-arguments
     def create_job_node(
-        self, job_config, input_node, *, runtime=None, platform=None, retry_counter=0
+        self,
+        job_config,
+        input_node,
+        *,
+        runtime=None,
+        platform=None,
+        retry_counter=0,
     ):
         """Create a new job node based on input and configuration"""
         jobfilter = input_node.get("jobfilter")
@@ -525,10 +549,14 @@ def create_job_node(
         # Test-specific fields inherited from parent node (kbuild or
         # job) if available
         if job_config.kind == "job":
-            job_node["data"]["kernel_type"] = input_node["data"].get("kernel_type")
+            job_node["data"]["kernel_type"] = input_node["data"].get(
+                "kernel_type"
+            )
             job_node["data"]["arch"] = input_node["data"].get("arch")
             job_node["data"]["defconfig"] = input_node["data"].get("defconfig")
-            job_node["data"]["config_full"] = input_node["data"].get("config_full")
+            job_node["data"]["config_full"] = input_node["data"].get(
+                "config_full"
+            )
             job_node["data"]["compiler"] = input_node["data"].get("compiler")
         # This information is highly useful, as we might
         # extract from it the following, for example:
@@ -574,7 +602,9 @@ def create_job_node(
             job_node = self._fsanitize_node_fields(job_node, "commit_message")
 
             try:
-                job_node["data"] = platform.format_params(job_node["data"], extra_args)
+                job_node["data"] = platform.format_params(
+                    job_node["data"], extra_args
+                )
             except Exception as error:
                 print(f"Exception Error, node id: {input_node['id']}, {error}")
                 raise error