Perform initial black pass that will also cover other linter issues

.github/workflows/lint.yml

@@ -17,8 +17,9 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install mypy types-setuptools
+          pip install mypy types-setuptools black
 
       - name: Run linters
         run: |
           mypy avocado_i2n
+          black --check --diff --color avocado_i2n

avocado_i2n/cartgraph/object.py

@@ -131,8 +131,9 @@ def regenerate_params(self, verbose: bool = False) -> None:
 
         :param verbose: whether to show generated parameter dictionaries
         """
-        generic_params = self.recipe.get_params(dict_index=self.dict_index,
-                                                show_dictionaries=verbose)
+        generic_params = self.recipe.get_params(
+            dict_index=self.dict_index, show_dictionaries=verbose
+        )
         self._params_cache = self.object_typed_params(generic_params)
         for key, value in list(self._params_cache.items()):
             if key.startswith("only_") or key.startswith("no_"):
@@ -148,6 +149,7 @@ class NetObject(TestObject):
     def component_form(self) -> str:
         # TODO: an unexpected order of joining in the Cartesian config requires us to override base property
         return self.params["name"]
+
     component_form = property(fget=component_form)
 
     def __init__(self, name: str, recipe: param.Reparsable) -> None:
@@ -182,10 +184,12 @@ def id(self) -> str:
         """Sufficiently unique ID to identify a test object."""
         assert len(self.composites) == 1, "Image objects need a unique composite"
         return self.long_suffix + "-" + self.composites[0].params["name"]
+
     id = property(fget=id)
 
     def component_form(self) -> str:
         return self.composites[0].component_form
+
     component_form = property(fget=component_form)
 
     def __init__(self, name: str, recipe: param.Reparsable) -> None:

avocado_i2n/cartgraph/worker.py

@@ -30,7 +30,8 @@
 from __future__ import annotations
 
 import logging as log
-logging = log.getLogger('avocado.job.' + __name__)
+
+logging = log.getLogger("avocado.job." + __name__)
 
 import aexpect
 from aexpect.exceptions import ShellTimeoutError
@@ -113,7 +114,7 @@ def overwrite_with_slot(self, slot: str) -> None:
             env_net = ""
             env_name = "c" + env_tuple[0] if env_tuple[0] else ""
             if env_name != "":
-                prefix = self.params['nets_ip_prefix']
+                prefix = self.params["nets_ip_prefix"]
                 ip = f"{prefix}.{env_name[1:]}"
             else:
                 ip = "localhost"
@@ -125,8 +126,10 @@ def overwrite_with_slot(self, slot: str) -> None:
             env_net = env_tuple[0]
             env_name = env_tuple[1]
             if not env_name.isdigit():
-                raise RuntimeError(f"Invalid remote host '{env_name}', "
-                                   f"only numbers (as forwarded ports) accepted")
+                raise RuntimeError(
+                    f"Invalid remote host '{env_name}', "
+                    f"only numbers (as forwarded ports) accepted"
+                )
             env_type = "remote"
             port = f"22{env_name}"
             ip = env_net
@@ -153,6 +156,7 @@ def start(self) -> bool:
             return True
         elif isolation_type == "lxc":
             import lxc
+
             cid = self.params["nets_host"]
             container = lxc.Container(cid)
             if not container.running:
@@ -180,6 +184,7 @@ def stop(self) -> bool:
             return True
         elif isolation_type == "lxc":
             import lxc
+
             cid = self.params["nets_host"]
             container = lxc.Container(cid)
             if container.running:
@@ -207,15 +212,21 @@ def get_session(self) -> RemoteSession:
         if session:
             # check for corrupted sessions
             try:
-                logging.debug("Remote session health check: " + session.cmd_output("date"))
+                logging.debug(
+                    "Remote session health check: " + session.cmd_output("date")
+                )
             except ShellTimeoutError as error:
                 logging.warning(f"Bad remote session health for {address}!")
                 session = None
         if not session:
-            session = remote.wait_for_login(self.params["nets_shell_client"],
-                                            self.params["nets_shell_host"], self.params["nets_shell_port"],
-                                            self.params["nets_username"], self.params["nets_password"],
-                                            self.params["nets_shell_prompt"])
+            session = remote.wait_for_login(
+                self.params["nets_shell_client"],
+                self.params["nets_shell_host"],
+                self.params["nets_shell_port"],
+                self.params["nets_username"],
+                self.params["nets_password"],
+                self.params["nets_shell_prompt"],
+            )
             cache[address] = session
 
         return session

avocado_i2n/cmd_parser.py

@@ -17,7 +17,8 @@
 import os
 import re
 import logging
-log = logging.getLogger('avocado.job.' + __name__)
+
+log = logging.getLogger("avocado.job." + __name__)
 
 from avocado.core.settings import settings
 from virttest import env_process
@@ -45,7 +46,7 @@ def params_from_cmd(config: Params) -> None:
         some standards for doing this first. Until then, the user won't directly
         interact with these keys anyway.
     """
-    suite_path = settings.as_dict().get('i2n.common.suite_path', ".")
+    suite_path = settings.as_dict().get("i2n.common.suite_path", ".")
     sys.path.insert(1, os.path.join(suite_path, "utils"))
 
     # validate typed vm names and possible vm specific restrictions
@@ -67,8 +68,10 @@ def params_from_cmd(config: Params) -> None:
     for cmd_param in config["params"]:
         re_param = re.match(r"(\w+)=(.*)", cmd_param)
         if re_param is None:
-            raise ValueError(f"Found malformed parameter on the command line '{cmd_param}' - "
-                             f"must be of the form <key>=<val>")
+            raise ValueError(
+                f"Found malformed parameter on the command line '{cmd_param}' - "
+                f"must be of the form <key>=<val>"
+            )
         (key, value) = re_param.group(1, 2)
         if key == "only" or key == "no":
             # detect if this is the primary restriction to escape defaults
@@ -82,33 +85,47 @@ def params_from_cmd(config: Params) -> None:
             tests_str += "%s %s\n" % (key, value)
         elif key.startswith("only_") or key.startswith("no_"):
             if re.match("(only|no)_nets", key):
-                nets_str = "%s %s\n" % (key.replace("_nets", ""), value) if value else ""
+                nets_str = (
+                    "%s %s\n" % (key.replace("_nets", ""), value) if value else ""
+                )
                 # TODO: unify nets_str with vm_strs treatment across the Cartesian graph interface
-                param_dict["nets"] = " ".join(param.all_suffixes_by_restriction(nets_str))
+                param_dict["nets"] = " ".join(
+                    param.all_suffixes_by_restriction(nets_str)
+                )
             else:
                 for vm_name in available_vms:
                     if re.match(f"(only|no)_{vm_name}", key):
                         # escape defaults for this vm and use the command line
                         use_vms_default[vm_name] = False
                         # main vm restriction part
-                        vm_str = "%s %s\n" % (key.replace(f"_{vm_name}", ""), value) if value else ""
+                        vm_str = (
+                            "%s %s\n" % (key.replace(f"_{vm_name}", ""), value)
+                            if value
+                            else ""
+                        )
                         vm_strs[vm_name] += vm_str
                         break
                 else:
-                    raise ValueError(f"Invalid object restriction {key} (no such object)")
+                    raise ValueError(
+                        f"Invalid object restriction {key} (no such object)"
+                    )
         # NOTE: comma in a parameter sense implies the same as space in config file
         elif key == "vms":
             # NOTE: no restrictions of the required vms are allowed during tests since
             # these are specified by each test (allowed only for manual setup steps)
             with_selected_vms[:] = value.split(",")
             for vm_name in with_selected_vms:
                 if vm_name not in available_vms:
-                    raise ValueError("The vm '%s' is not among the supported vms: "
-                                     "%s" % (vm_name, ", ".join(available_vms)))
+                    raise ValueError(
+                        "The vm '%s' is not among the supported vms: "
+                        "%s" % (vm_name, ", ".join(available_vms))
+                    )
         elif key == "nets":
             if nets_str != "":
-                raise ValueError(f"Cannot specify explicit net suffixes {value} together with "
-                                 f"a nets restriction, currently also specified '{nets_str.rstrip()}'")
+                raise ValueError(
+                    f"Cannot specify explicit net suffixes {value} together with "
+                    f"a nets restriction, currently also specified '{nets_str.rstrip()}'"
+                )
             value = value.replace(",", " ")
             param_dict[key] = value
         else:
@@ -119,52 +136,70 @@ def params_from_cmd(config: Params) -> None:
     log.debug("Parsed param dict '%s'", param_dict)
 
     # get minimal configurations and parse defaults if no command line arguments
-    config["vms_params"], config["vm_strs"] = full_vm_params_and_strs(param_dict, vm_strs,
-                                                                      use_vms_default)
+    config["vms_params"], config["vm_strs"] = full_vm_params_and_strs(
+        param_dict, vm_strs, use_vms_default
+    )
     config["vms_params"]["vms"] = " ".join(with_selected_vms)
     config["available_vms"] = vm_strs.copy()
     for vm_name in available_vms:
         # the keys of vm strings must be equivalent to the selected vms
         if vm_name not in with_selected_vms:
             del config["vm_strs"][vm_name]
-    config["tests_params"], config["tests_str"] = full_tests_params_and_str(param_dict, tests_str,
-                                                                            use_tests_default)
+    config["tests_params"], config["tests_str"] = full_tests_params_and_str(
+        param_dict, tests_str, use_tests_default
+    )
     config["available_restrictions"] = available_restrictions
 
     # control against invoking only runnable tests and empty Cartesian products
     control_config = param.Reparsable()
-    control_config.parse_next_batch(base_file="sets.cfg",
-                                    ovrwrt_file=param.tests_ovrwrt_file(),
-                                    ovrwrt_str=config["tests_str"],
-                                    ovrwrt_dict=config["param_dict"])
+    control_config.parse_next_batch(
+        base_file="sets.cfg",
+        ovrwrt_file=param.tests_ovrwrt_file(),
+        ovrwrt_str=config["tests_str"],
+        ovrwrt_dict=config["param_dict"],
+    )
     control_parser = control_config.get_parser()
     if with_nontrivial_restrictions:
-        log.info("%s tests with nontrivial restriction %s",
-                 len(list(control_parser.get_dicts())), config["tests_str"])
+        log.info(
+            "%s tests with nontrivial restriction %s",
+            len(list(control_parser.get_dicts())),
+            config["tests_str"],
+        )
 
     # prefix for all tests of the current run making it possible to perform multiple runs in one command
     config["prefix"] = ""
 
     # log into files for each major level the way it was done for autotest
     config["job.run.store_logging_stream"] = ["avocado.core:DEBUG"]
     # dump parsed and traversed graph at each test loading and running step
-    graph.set_graph_logging_level(level=config["tests_params"].get_numeric("cartgraph_verbose_level", 20))
+    graph.set_graph_logging_level(
+        level=config["tests_params"].get_numeric("cartgraph_verbose_level", 20)
+    )
 
     # set default off and on state backends
     from .states import lvm, qcow2, lxc, btrfs, ramfile, pool, vmnet
-    ss.BACKENDS = {"qcow2": qcow2.QCOW2Backend, "qcow2ext": qcow2.QCOW2ExtBackend,
-                   "lvm": lvm.LVMBackend,
-                   "lxc": lxc.LXCBackend, "btrfs": btrfs.BtrfsBackend,
-                   "qcow2vt": qcow2.QCOW2VTBackend, "ramfile": ramfile.RamfileBackend,
-                   "vmnet": vmnet.VMNetBackend}
+
+    ss.BACKENDS = {
+        "qcow2": qcow2.QCOW2Backend,
+        "qcow2ext": qcow2.QCOW2ExtBackend,
+        "lvm": lvm.LVMBackend,
+        "lxc": lxc.LXCBackend,
+        "btrfs": btrfs.BtrfsBackend,
+        "qcow2vt": qcow2.QCOW2VTBackend,
+        "ramfile": ramfile.RamfileBackend,
+        "vmnet": vmnet.VMNetBackend,
+    }
     ramfile.RamfileBackend.image_state_backend = qcow2.QCOW2ExtBackend
 
     # attach environment processing hooks
     env_process_hooks()
 
 
-def full_vm_params_and_strs(param_dict: dict[str, str] | None, vm_strs: dict[str, str],
-                            use_vms_default: dict[str, bool]) -> tuple[Params, dict[str,str]]:
+def full_vm_params_and_strs(
+    param_dict: dict[str, str] | None,
+    vm_strs: dict[str, str],
+    use_vms_default: dict[str, bool],
+) -> tuple[Params, dict[str, str]]:
     """
     Add default vm parameters and strings for missing command line such.
 
@@ -177,9 +212,11 @@ def full_vm_params_and_strs(param_dict: dict[str, str] | None, vm_strs: dict[str
              restriction could be found for some vm
     """
     vms_config = param.Reparsable()
-    vms_config.parse_next_batch(base_file="guest-base.cfg",
-                                ovrwrt_file=param.vms_ovrwrt_file(),
-                                ovrwrt_dict=param_dict)
+    vms_config.parse_next_batch(
+        base_file="guest-base.cfg",
+        ovrwrt_file=param.vms_ovrwrt_file(),
+        ovrwrt_dict=param_dict,
+    )
     vms_params = vms_config.get_params()
     for vm_name in param.all_objects("vms"):
         if use_vms_default[vm_name]:
@@ -189,8 +226,9 @@ def full_vm_params_and_strs(param_dict: dict[str, str] | None, vm_strs: dict[str
     return vms_params, vm_strs
 
 
-def full_tests_params_and_str(param_dict: dict[str, str] | None,
-                              tests_str: str, use_tests_default: bool) -> tuple[Params, str]:
+def full_tests_params_and_str(
+    param_dict: dict[str, str] | None, tests_str: str, use_tests_default: bool
+) -> tuple[Params, str]:
     """
     Add default tests parameters and string for missing command line such.
 
@@ -202,16 +240,20 @@ def full_tests_params_and_str(param_dict: dict[str, str] | None,
              not valid (among the available ones)
     """
     tests_config = param.Reparsable()
-    tests_config.parse_next_batch(base_file="groups-base.cfg",
-                                  ovrwrt_file=param.tests_ovrwrt_file(),
-                                  ovrwrt_dict=param_dict)
+    tests_config.parse_next_batch(
+        base_file="groups-base.cfg",
+        ovrwrt_file=param.tests_ovrwrt_file(),
+        ovrwrt_dict=param_dict,
+    )
     tests_params = tests_config.get_params()
     if use_tests_default:
         default = tests_params.get("default_only", "all")
         available_restrictions = param.all_restrictions()
         if default not in available_restrictions:
-            raise ValueError("Invalid primary restriction 'only=%s'! It has to be one "
-                             "of %s" % (default, ", ".join(available_restrictions)))
+            raise ValueError(
+                "Invalid primary restriction 'only=%s'! It has to be one "
+                "of %s" % (default, ", ".join(available_restrictions))
+            )
         tests_str += "only %s\n" % default
     log.debug("Parsed tests string '%s'", tests_str)
     return tests_params, tests_str
@@ -222,18 +264,23 @@ def env_process_hooks() -> None:
     Add env processing hooks to handle on/off state get/set operations
     and vmnet networking setup and instance attachment to environment.
     """
+
     def on_state(fn: Callable[[Any], Any]) -> Any:
         def wrapper(test: VirtTest, params: Params, env: Env) -> Any:
             params["skip_types"] = "nets/vms/images nets"
             fn(params, env)
             del params["skip_types"]
+
         return wrapper
-    def off_state(fn: Callable[[Any], Any])-> Any:
+
+    def off_state(fn: Callable[[Any], Any]) -> Any:
         def wrapper(test: VirtTest, params: Params, env: Env) -> Any:
             params["skip_types"] = "nets/vms"
             fn(params, env)
             del params["skip_types"]
+
         return wrapper
+
     env_process.preprocess_vm_off_hook = off_state(ss.get_states)
     env_process.preprocess_vm_on_hook = on_state(ss.get_states)
     env_process.postprocess_vm_on_hook = on_state(ss.set_states)