util/agents/network_interface: run ruff format

Signed-off-by: Bastian Krause <[email protected]>

Author: Bastian-Krause

labgrid/util/agents/network_interface.py

@@ -4,14 +4,16 @@
 from time import monotonic, sleep
 
 import gi
-gi.require_version('NM', '1.0')
+
+gi.require_version("NM", "1.0")
 from gi.repository import GLib, NM
 
 # ensure all wrapper objects for Settings types are created
 for _name in dir(NM):
-    if _name.startswith('Setting'):
+    if _name.startswith("Setting"):
         getattr(NM, _name)
 
+
 class Future:
     def __init__(self):
         self._event = threading.Event()
@@ -38,6 +40,7 @@ def run(self):
             GLib.MainLoop(None).run()
         except Exception:
             import traceback
+
             traceback.print_exc(file=sys.stderr)
             sys.exit(1)
 
@@ -53,13 +56,11 @@ def cb(data):
         done.wait()
         return result[0]
 
+
 def address_from_str(s):
-    assert '/' in s, "IP address must be in the form address/prefix"
-    (addr, prefix) = s.split('/', 1)
-    return NM.IPAddress.new(
-            socket.AF_INET6 if ':' in addr else socket.AF_INET,
-            addr,
-            int(prefix))
+    assert "/" in s, "IP address must be in the form address/prefix"
+    (addr, prefix) = s.split("/", 1)
+    return NM.IPAddress.new(socket.AF_INET6 if ":" in addr else socket.AF_INET, addr, int(prefix))
 
 
 def connection_from_dict(data):
@@ -71,13 +72,13 @@ def connection_from_dict(data):
             if not v:
                 continue
 
-            if k == 'mac-address':
-                v = ':'.join(f"{x:02X}" for x in v)
-            elif k == 'ssid':
+            if k == "mac-address":
+                v = ":".join(f"{x:02X}" for x in v)
+            elif k == "ssid":
                 if isinstance(v, str):
                     v = v.encode()
                 v = GLib.Bytes(v)
-            elif k == 'addresses':
+            elif k == "addresses":
                 # setting addresses via the property doesn't seem to work
                 for x in v:
                     setting.add_address(address_from_str(x))
@@ -92,13 +93,15 @@ def connection_from_dict(data):
         con.add_setting(setting)
     return con
 
+
 class NMDev:
     def __init__(self, interface):
         self._interface = interface
-        self._nm_dev = nm.get_device_by_iface(interface) # pylint: disable=possibly-used-before-assignment
+        self._nm_dev = nm.get_device_by_iface(interface)  # pylint: disable=possibly-used-before-assignment
 
     def _delete_connection(self, con):
         future = Future()
+
         def cb(con, res, error):
             assert error is None
             try:
@@ -114,7 +117,7 @@ def cb(con, res, error):
         )
 
     def get_settings(self):
-        lg_con = nm.get_connection_by_id(f"labgrid-{self._interface}") # pylint: disable=possibly-used-before-assignment
+        lg_con = nm.get_connection_by_id(f"labgrid-{self._interface}")  # pylint: disable=possibly-used-before-assignment
         if lg_con:
             return dict(lg_con.to_dbus(NM.ConnectionSerializationFlags.ALL))
 
@@ -125,26 +128,29 @@ def get_active_settings(self):
             return dict(con.to_dbus(NM.ConnectionSerializationFlags.ALL))
 
     def configure(self, data):
-        lg_con = nm.get_connection_by_id(f"labgrid-{self._interface}") # pylint: disable=possibly-used-before-assignment
+        lg_con = nm.get_connection_by_id(f"labgrid-{self._interface}")  # pylint: disable=possibly-used-before-assignment
         if lg_con:
             self._delete_connection(lg_con)
-        data['connection'].update({
-            'id': f"labgrid-{self._interface}",
-            'autoconnect': False,
-            'interface-name': self._interface,
-        })
+        data["connection"].update(
+            {
+                "id": f"labgrid-{self._interface}",
+                "autoconnect": False,
+                "interface-name": self._interface,
+            }
+        )
         con = connection_from_dict(data)
 
         future = Future()
+
         def cb(dev, res, error):
             assert error is None
             try:
-                res = nm.add_and_activate_connection_finish(res) # pylint: disable=possibly-used-before-assignment
+                res = nm.add_and_activate_connection_finish(res)  # pylint: disable=possibly-used-before-assignment
                 future.set(res)
             except Exception as e:
                 future.set(e)
 
-        nm.add_and_activate_connection_async( # pylint: disable=possibly-used-before-assignment
+        nm.add_and_activate_connection_async(  # pylint: disable=possibly-used-before-assignment
             con,
             self._nm_dev,
             None,  # specific_object
@@ -172,7 +178,7 @@ def wait_state(self, expected, timeout):
             )
 
     def disable(self):
-        lg_con = nm.get_connection_by_id(f"labgrid-{self._interface}") # pylint: disable=possibly-used-before-assignment
+        lg_con = nm.get_connection_by_id(f"labgrid-{self._interface}")  # pylint: disable=possibly-used-before-assignment
         if lg_con:
             self._delete_connection(lg_con)
 
@@ -182,58 +188,59 @@ def _flags_to_str(self, flags):
     def _accesspoint_to_dict(self, ap):
         res = {}
 
-        res['flags'] = self._flags_to_str(ap.get_flags())
-        res['wpa-flags'] = self._flags_to_str(ap.get_wpa_flags())
-        res['rsn-flags'] = self._flags_to_str(ap.get_rsn_flags())
-        res['bssid'] = ap.get_bssid()
-        res['ssid'] = ap.get_ssid().get_data().decode(errors='surrogateescape')
-        res['frequency'] = ap.get_frequency()
-        res['mode'] = self._flags_to_str(ap.get_mode())
-        res['max-bitrate'] = ap.get_max_bitrate()
-        res['strength'] = ap.get_strength()
+        res["flags"] = self._flags_to_str(ap.get_flags())
+        res["wpa-flags"] = self._flags_to_str(ap.get_wpa_flags())
+        res["rsn-flags"] = self._flags_to_str(ap.get_rsn_flags())
+        res["bssid"] = ap.get_bssid()
+        res["ssid"] = ap.get_ssid().get_data().decode(errors="surrogateescape")
+        res["frequency"] = ap.get_frequency()
+        res["mode"] = self._flags_to_str(ap.get_mode())
+        res["max-bitrate"] = ap.get_max_bitrate()
+        res["strength"] = ap.get_strength()
 
         return res
 
     def get_state(self):
         state = {}
 
         device = {}
-        device['capabilities'] = self._flags_to_str(self._nm_dev.get_capabilities())
-        device['device-type'] = self._flags_to_str(self._nm_dev.get_device_type())
-        device['state'] = self._flags_to_str(self._nm_dev.get_state())
-        device['state-reason'] = self._flags_to_str(self._nm_dev.get_state_reason())
-        device['hw-address'] = self._nm_dev.get_hw_address()
-        device['driver'] = self._nm_dev.get_driver()
-        device['udi'] = self._nm_dev.get_udi()
-        device['mtu'] = self._nm_dev.get_mtu()
-        device['vendor'] = self._nm_dev.get_vendor()
-        device['product'] = self._nm_dev.get_product()
-        state['device'] = device
+        device["capabilities"] = self._flags_to_str(self._nm_dev.get_capabilities())
+        device["device-type"] = self._flags_to_str(self._nm_dev.get_device_type())
+        device["state"] = self._flags_to_str(self._nm_dev.get_state())
+        device["state-reason"] = self._flags_to_str(self._nm_dev.get_state_reason())
+        device["hw-address"] = self._nm_dev.get_hw_address()
+        device["driver"] = self._nm_dev.get_driver()
+        device["udi"] = self._nm_dev.get_udi()
+        device["mtu"] = self._nm_dev.get_mtu()
+        device["vendor"] = self._nm_dev.get_vendor()
+        device["product"] = self._nm_dev.get_product()
+        state["device"] = device
 
         for name, ip_cfg in (
-                ('ip4-config', self._nm_dev.get_ip4_config()),
-                ('ip6-config', self._nm_dev.get_ip6_config()),
-                ):
+            ("ip4-config", self._nm_dev.get_ip4_config()),
+            ("ip6-config", self._nm_dev.get_ip6_config()),
+        ):
             if ip_cfg is None:
                 continue
             cfg = {}
-            cfg['addresses'] = []
+            cfg["addresses"] = []
             for address in ip_cfg.get_addresses():
-                cfg['addresses'].append(f"{address.get_address()}/{address.get_prefix()}")
-            cfg['gateway'] = ip_cfg.get_gateway()
+                cfg["addresses"].append(f"{address.get_address()}/{address.get_prefix()}")
+            cfg["gateway"] = ip_cfg.get_gateway()
             state[name] = cfg
 
         try:
             ap = self._nm_dev.get_active_access_point()
         except AttributeError:
             ap = None
         if ap:
-            state['active-access-point'] = self._accesspoint_to_dict(ap)
+            state["active-access-point"] = self._accesspoint_to_dict(ap)
 
         return state
 
     def request_scan(self):
         future = Future()
+
         def cb(dev, res, error):
             assert error is None
             try:
@@ -251,7 +258,7 @@ def cb(dev, res, error):
         future.wait()  # we must wait, but don't need to return it here
 
     def get_access_points(self, scan):
-        if scan is None: # automatically scan if needed
+        if scan is None:  # automatically scan if needed
             age = NM.utils_get_timestamp_msec() - self._nm_dev.get_last_scan()
             scan = bool(age > 30_000)
 
@@ -275,80 +282,93 @@ def get_dhcpd_leases(self):
         with open(f"/var/lib/NetworkManager/dnsmasq-{self._interface}.leases") as f:
             for line in f:
                 line = line.strip().split()
-                if line[3] == '*':
+                if line[3] == "*":
                     line[3] = None
-                if line[4] == '*':
+                if line[4] == "*":
                     line[4] = None
-                leases.append({
-                    'expire': int(line[0]),
-                    'mac': line[1],
-                    'ip': line[2],
-                    'hostname': line[3],
-                    'id': line[4],
-                })
+                leases.append(
+                    {
+                        "expire": int(line[0]),
+                        "mac": line[1],
+                        "ip": line[2],
+                        "hostname": line[3],
+                        "id": line[4],
+                    }
+                )
         return leases
 
-if getattr(NM.Client, '__gtype__', None):
+
+if getattr(NM.Client, "__gtype__", None):
     # hide this from sphinx autodoc
     bl = BackgroundLoop()
     bl.start()
     nm = bl.block_on(NM.Client.new, None)
 
 _nmdevs = {}
 
+
 def _get_nmdev(interface):
     if interface not in _nmdevs:
         _nmdevs[interface] = NMDev(interface)
     return _nmdevs[interface]
 
+
 def handle_configure(interface, settings):
     nmdev = _get_nmdev(interface)
     return nmdev.configure(settings)
 
+
 def handle_wait_state(interface, expected, timeout=60):
     nmdev = _get_nmdev(interface)
     return nmdev.wait_state(expected, timeout)
 
+
 def handle_disable(interface):
     nmdev = _get_nmdev(interface)
     return nmdev.disable()
 
+
 def handle_get_active_settings(interface):
     nmdev = _get_nmdev(interface)
     return nmdev.get_active_settings()
 
+
 def handle_get_settings(interface):
     nmdev = _get_nmdev(interface)
     return nmdev.get_settings()
 
+
 def handle_get_state(interface):
     nmdev = _get_nmdev(interface)
     return nmdev.get_state()
 
+
 def handle_get_dhcpd_leases(interface):
     nmdev = _get_nmdev(interface)
     return nmdev.get_dhcpd_leases()
 
+
 def handle_request_scan(interface):
     nmdev = _get_nmdev(interface)
     return nmdev.request_scan()
 
+
 def handle_get_access_points(interface, scan=None):
     nmdev = _get_nmdev(interface)
     return nmdev.get_access_points(scan)
 
 
 methods = {
     # basic
-    'configure': handle_configure,
-    'wait_state': handle_wait_state,
-    'disable': handle_disable,
-    'get_active_settings': handle_get_active_settings,
-    'get_settings': handle_get_settings,
-    'get_state':  handle_get_state,
+    "configure": handle_configure,
+    "wait_state": handle_wait_state,
+    "disable": handle_disable,
+    "get_active_settings": handle_get_active_settings,
+    "get_settings": handle_get_settings,
+    "get_state": handle_get_state,
     # dhcpd
-    'get_dhcpd_leases': handle_get_dhcpd_leases,
+    "get_dhcpd_leases": handle_get_dhcpd_leases,
     # wireless
-    'request_scan': handle_request_scan,
-    'get_access_points': handle_get_access_points,
+    "request_scan": handle_request_scan,
+    "get_access_points": handle_get_access_points,
 }

pyproject.toml

@@ -224,6 +224,7 @@ include = [
   "labgrid/remote/**/*.py",
   "labgrid/resource/httpvideostream.py",
   "labgrid/resource/provider.py",
+  "labgrid/util/agents/network_interface.py",
   "labgrid/util/agents/usb_hid_relay.py",
   "labgrid/util/exceptions.py",
 ]