Merge pull request labgrid-project#1823 from jluebbe/coding-style

run ruff format on some files

Author: Emantor

contrib/coordinator-statsd.py

@@ -72,11 +72,7 @@ async def report_reservations(session, tags, gauges):
         for group_name, group in groups.items():
             inc_gauge(
                 gauges,
-                ".".join(
-                    ["reservations", group_name]
-                    + [group.get(t, "") for t in tags]
-                    + [reservation.state.name]
-                ),
+                ".".join(["reservations", group_name] + [group.get(t, "") for t in tags] + [reservation.state.name]),
             )
 
 
@@ -92,9 +88,7 @@ async def report_places(session, tags, gauges):
 
 
 def main():
-    parser = argparse.ArgumentParser(
-        description="Report Labgrid usage metrics to statsd"
-    )
+    parser = argparse.ArgumentParser(description="Report Labgrid usage metrics to statsd")
     parser.add_argument(
         "-x",
         "--coordinator",

contrib/sync-places.py

@@ -112,21 +112,21 @@ async def do_sync(session, args):
                         print(f"Adding match '{match}' for place {name}")
 
                     if not args.dry_run:
-                        request = labgrid_coordinator_pb2.AddPlaceMatchRequest(placename=name, pattern=match, rename=rename)
+                        request = labgrid_coordinator_pb2.AddPlaceMatchRequest(
+                            placename=name, pattern=match, rename=rename
+                        )
                         await session.stub.AddPlaceMatch(request)
                         await session.sync_with_coordinator()
                     changed = True
 
-            tags = { str(k): str(v) for k, v in config["places"][name].get("tags", {}).items() }
+            tags = {str(k): str(v) for k, v in config["places"][name].get("tags", {}).items()}
 
             if place_tags != tags:
                 print(
                     "Setting tags for place %s to %s"
                     % (
                         name,
-                        ", ".join(
-                            "%s=%s" % (key, value) for (key, value) in tags.items()
-                        ),
+                        ", ".join("%s=%s" % (key, value) for (key, value) in tags.items()),
                     )
                 )
 
@@ -146,10 +146,7 @@ async def do_dump(session, args):
         config = {"places": {}}
         for name, place in session.places.items():
             config["places"][name] = {
-                "matches": [
-                    {repr(m): m.rename} if m.rename else repr(m)
-                    for m in place.matches
-                    ],
+                "matches": [{repr(m): m.rename} if m.rename else repr(m) for m in place.matches],
                 "tags": {k: v for k, v in place.tags.items()},
             }
 
@@ -191,9 +188,7 @@ async def do_dump(session, args):
     subparsers = parser.add_subparsers()
     subparsers.required = True
 
-    sync_parser = subparsers.add_parser(
-        "sync", help="Synchronize coordinator places with file"
-    )
+    sync_parser = subparsers.add_parser("sync", help="Synchronize coordinator places with file")
     sync_parser.add_argument(
         "places",
         metavar="FILE",

dockerfiles/staging/client/simple-test/remote_shell_test.py

@@ -1,8 +1,8 @@
 def test_shell(target):
-    ssh_driver = target.get_driver('SSHDriver')
+    ssh_driver = target.get_driver("SSHDriver")
     target.activate(ssh_driver)
 
-    stdout, stderr, returncode = ssh_driver.run('uname -r')
+    stdout, stderr, returncode = ssh_driver.run("uname -r")
 
     assert stdout
-    print(f'Kernel {stdout}')
+    print(f"Kernel {stdout}")

labgrid/autoinstall/main.py

@@ -1,4 +1,5 @@
 """The autoinstall.main module runs an installation script automatically on multiple targets."""
+
 import ast
 import argparse
 import logging
@@ -20,44 +21,46 @@ def __init__(self, env, args, name):
         self.name = name
 
         self.context = {
-            'name': self.name,
-            'env': self.env,
-            'config': self.config,
+            "name": self.name,
+            "env": self.env,
+            "config": self.config,
         }
         self.context.update(target_factory.resources)
         self.context.update(target_factory.drivers)
 
     def _get_function(self, name, context):
-        snippet = self.config.data['autoinstall'].get(name)
+        snippet = self.config.data["autoinstall"].get(name)
         if not snippet:
             return None
 
         code = f"def {name}():\n{textwrap.indent(snippet, ' ')}"
         tree = ast.parse(code, filename=self.env.config_file)
         ast.increment_lineno(tree, snippet.start_mark.line)
-        co = compile(tree, filename=self.env.config_file, mode='exec')
+        co = compile(tree, filename=self.env.config_file, mode="exec")
 
         stage = {}
         exec(co, context, stage)  # pylint: disable=exec-used
         return stage[name]
 
     def _get_setup_function(self):
         context = self.context
-        context['log'] = self.log.getChild('setup')
-        setup = self._get_function('setup', self.context)
+        context["log"] = self.log.getChild("setup")
+        setup = self._get_function("setup", self.context)
         if setup is None:
+
             def setup():
                 pass
+
         return setup
 
     def _get_handler_function(self):
         context = self.context.copy()
-        context['log'] = self.log.getChild('handler')
-        handler = self._get_function('handler', context)
+        context["log"] = self.log.getChild("handler")
+        handler = self._get_function("handler", context)
         return handler
 
     def _get_initial_resource(self):
-        cls = self.config.data['autoinstall'].get('initial-resource')
+        cls = self.config.data["autoinstall"].get("initial-resource")
         if not cls:
             return None
 
@@ -69,7 +72,7 @@ def run(self):
 
         try:
             self.target = self.env.get_target(self.name)
-            self.context['target'] = self.target
+            self.context["target"] = self.target
             if self.target is None:
                 raise KeyError
         except Exception:  # pylint: disable=broad-except
@@ -88,8 +91,7 @@ def run(self):
     def run_once(self):
         try:
             if self.initial_resource:
-                self.log.info("waiting until %s is available",
-                              self.initial_resource.display_name)
+                self.log.info("waiting until %s is available", self.initial_resource.display_name)
                 while True:
                     self.target.update_resources()
                     if self.initial_resource.avail:
@@ -100,13 +102,11 @@ def run_once(self):
             self.target.update_resources()
             result = self.handler()
             if result is not None:
-                self.log.warning("unexpected return value from handler: %s",
-                                 repr(result))
+                self.log.warning("unexpected return value from handler: %s", repr(result))
             self.log.info("completed handler")
 
             if self.initial_resource:
-                self.log.info("waiting until %s is unavailable",
-                              self.initial_resource.display_name)
+                self.log.info("waiting until %s is unavailable", self.initial_resource.display_name)
                 while True:
                     self.target.update_resources()
                     if not self.initial_resource.avail:
@@ -115,11 +115,9 @@ def run_once(self):
 
         except NoResourceFoundError as e:
             if e.filter and len(e.filter) > 1:
-                self.log.warning("resources %s not found, restarting",
-                                 e.filter)
+                self.log.warning("resources %s not found, restarting", e.filter)
             elif e.filter:
-                self.log.warning("resource %s not found, restarting",
-                                 next(iter(e.filter)))
+                self.log.warning("resource %s not found, restarting", next(iter(e.filter)))
             else:
                 self.log.warning("resource not found, restarting")
         except Exception:  # pylint: disable=broad-except
@@ -141,24 +139,20 @@ def __init__(self, env, args):
         self.log = logging.getLogger("manager")
 
     def configure(self):
-        if not 'autoinstall' in self.env.config.data:
-            self.log.error("no 'autoinstall' section found in '%s'",
-                           self.env.config_file)
+        if not "autoinstall" in self.env.config.data:
+            self.log.error("no 'autoinstall' section found in '%s'", self.env.config_file)
             return False
 
-        if not 'handler' in self.env.config.data['autoinstall']:
-            self.log.error("no 'handler' definition found in '%s'",
-                           self.env.config_file)
+        if not "handler" in self.env.config.data["autoinstall"]:
+            self.log.error("no 'handler' definition found in '%s'", self.env.config_file)
             return False
 
         self.handlers = {}
-        for target_name in self.config.data.get('targets', {}).keys():
-            self.handlers[target_name] = Handler(self.env, self.args,
-                                                 target_name)
+        for target_name in self.config.data.get("targets", {}).keys():
+            self.handlers[target_name] = Handler(self.env, self.args, target_name)
 
         if not self.handlers:
-            self.log.error("no targets found in '%s'",
-                           self.env.config_file)
+            self.log.error("no targets found in '%s'", self.env.config_file)
             return False
 
         return True
@@ -175,28 +169,14 @@ def join(self):
         for handler in self.handlers.values():
             handler.join()
 
+
 def main():
     basicConfig(level=logging.INFO)
 
     parser = argparse.ArgumentParser()
-    parser.add_argument(
-        '-d',
-        '--debug',
-        action='store_true',
-        default=False,
-        help="enable debug mode"
-    )
-    parser.add_argument(
-        '--once',
-        action='store_true',
-        default=False,
-        help="handle each target only once"
-    )
-    parser.add_argument(
-        'config',
-        type=str,
-        help="config file"
-    )
+    parser.add_argument("-d", "--debug", action="store_true", default=False, help="enable debug mode")
+    parser.add_argument("--once", action="store_true", default=False, help="handle each target only once")
+    parser.add_argument("config", type=str, help="config file")
 
     args = parser.parse_args()
 

labgrid/binding.py

@@ -41,8 +41,7 @@ class BindingMixin:
 
     # these are controlled by the Target
     target = attr.ib()
-    name = attr.ib(
-        validator=attr.validators.optional(attr.validators.instance_of(str)))
+    name = attr.ib(validator=attr.validators.optional(attr.validators.instance_of(str)))
     state = attr.ib(default=BindingState.idle, init=False)
 
     def __attrs_post_init__(self):
@@ -108,9 +107,7 @@ def check_bound(cls, func):
         @wraps(func)
         def wrapper(self, *_args, **_kwargs):
             if self.state is BindingState.active:
-                raise StateError(
-                    f'{self} is active, but must be deactivated to call {func.__qualname__}'
-                )
+                raise StateError(f"{self} is active, but must be deactivated to call {func.__qualname__}")
             elif self.state is not BindingState.bound:
                 raise StateError(
                     f'{self} has not been bound, {func.__qualname__} cannot be called in state "{self.state.name}"'  # pylint: disable=line-too-long
@@ -123,6 +120,7 @@ class NamedBinding:
         """
         Marks a binding (or binding set) as requiring an explicit name.
         """
+
         def __init__(self, value):
             self.value = value
 

labgrid/consoleloggingreporter.py

@@ -11,6 +11,7 @@ class ConsoleLoggingReporter:
     Args:
         logpath (str): path to store the logfiles in
     """
+
     instance = None
 
     @classmethod
@@ -49,42 +50,33 @@ def get_logfile(self, event):
             log = self._logcache[source]
         except KeyError:
             if source.name:
-                name = f'console_{source.target.name}_{source.name}'
+                name = f"console_{source.target.name}_{source.name}"
             else:
-                name = f'console_{source.target.name}'
+                name = f"console_{source.target.name}"
             name = os.path.join(self.logpath, name)
             try:
-                log = self._logcache[source] = open(name, mode='ab',
-                                                    buffering=0)
+                log = self._logcache[source] = open(name, mode="ab", buffering=0)
             except OSError as e:
                 print(f"failed to open log file {name}: {e}", file=sys.stderr)
                 log = self._logcache[source] = None
             if not log:
                 return None
 
             if source.name:
-                log.write(
-                    f"Labgrid Console Logfile for {source.target.name} {source.name}\n"
-                    .encode("utf-8")
-                )
+                log.write(f"Labgrid Console Logfile for {source.target.name} {source.name}\n".encode("utf-8"))
             else:
-                log.write(
-                    f"Labgrid Console Logfile for {source.target.name}\n"
-                    .encode("utf-8")
-                )
-            log.write(
-                f"Logfile started at {datetime.now()}\n".encode("utf-8")
-            )
-            log.write("=== Log starts here ===\n".encode('utf-8'))
+                log.write(f"Labgrid Console Logfile for {source.target.name}\n".encode("utf-8"))
+            log.write(f"Logfile started at {datetime.now()}\n".encode("utf-8"))
+            log.write("=== Log starts here ===\n".encode("utf-8"))
 
         return log
 
     def notify(self, event):
         """This is the callback function for steps"""
         step = event.step
-        if step.tag == 'console':
-            if step.title == 'read':
-                if event.data.get('state') == 'stop':
+        if step.tag == "console":
+            if step.title == "read":
+                if event.data.get("state") == "stop":
                     if step.result and step.source:
                         log = self.get_logfile(event)
                         if not log:

labgrid/driver/commandmixin.py

@@ -15,7 +15,7 @@ def __attrs_post_init__(self):
         super().__attrs_post_init__()
 
     @Driver.check_active
-    @step(args=['cmd', 'pattern'])
+    @step(args=["cmd", "pattern"])
     def wait_for(self, cmd, pattern, timeout=30.0, sleepduration=1):
         """
         Wait until the pattern is detected in the output of cmd. Raises
@@ -34,7 +34,7 @@ def wait_for(self, cmd, pattern, timeout=30.0, sleepduration=1):
             raise ExecutionError("Wait timeout expired")
 
     @Driver.check_active
-    @step(args=['cmd', 'expected', 'tries', 'timeout', 'sleepduration'])
+    @step(args=["cmd", "expected", "tries", "timeout", "sleepduration"])
     def poll_until_success(self, cmd, *, expected=0, tries=None, timeout=30.0, sleepduration=1):
         """
         Poll a command until a specific exit code is detected.
@@ -63,8 +63,7 @@ def poll_until_success(self, cmd, *, expected=0, tries=None, timeout=30.0, sleep
                     break
         return False
 
-    def _run_check(self, cmd: str, *, timeout=30, codec: str = "utf-8",
-                   decodeerrors: str = "strict"):
+    def _run_check(self, cmd: str, *, timeout=30, codec: str = "utf-8", decodeerrors: str = "strict"):
         """
         Internal function which runs the specified command on the shell and
         returns the output if successful, raises ExecutionError otherwise.
@@ -75,14 +74,13 @@ def _run_check(self, cmd: str, *, timeout=30, codec: str = "utf-8",
         Returns:
             List[str]: stdout of the executed command
         """
-        stdout, stderr, exitcode = self._run(cmd, timeout=timeout, codec=codec,
-                                             decodeerrors=decodeerrors)
+        stdout, stderr, exitcode = self._run(cmd, timeout=timeout, codec=codec, decodeerrors=decodeerrors)
         if exitcode != 0:
             raise ExecutionError(cmd, stdout, stderr)
         return stdout
 
     @Driver.check_active
-    @step(args=['cmd'], result=True)
+    @step(args=["cmd"], result=True)
     def run_check(self, cmd: str, *, timeout=30, codec="utf-8", decodeerrors="strict"):
         """
         External run_check function, only available if the driver is active.