Add type hints to blueman.plugins

apps/blueman-mechanism.in

@@ -2,29 +2,17 @@
 import sys
 import os
 import logging
-import importlib
 
-from gi.repository import Gio, GLib
+from blueman.main.MechanismApplication import MechanismApplication
 
 # support running uninstalled
 _dirname = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
 if 'BLUEMAN_SOURCE' in os.environ:
     sys.path = [_dirname, os.path.join(_dirname, 'module', '.libs')] + sys.path
-    timeout = 9999
     os.environ["GSETTINGS_SCHEMA_DIR"] = os.path.join(_dirname, "data")
-else:
-    timeout = 30
 
-from blueman.Constants import POLKIT
 from blueman.Functions import set_proc_title, create_logger, create_parser
 
-import blueman.plugins.mechanism
-from blueman.plugins.MechanismPlugin import MechanismPlugin
-
-from blueman.main.DbusService import DbusService, DbusError
-
-loop = GLib.MainLoop()
-
 
 class StreamToLogger:
     """
@@ -76,95 +64,6 @@ logging.info("Starting blueman-mechanism")
 
 os.environ["PATH"] = "/usr/bin:/bin:/usr/sbin:/sbin"
 
-dhcp_pids = []
-
-
-class Timer:
-    def __init__(self):
-        self.time = 0
-        self.stopped = False
-        GLib.timeout_add(1000, self.tick)
-
-    def tick(self):
-        if not self.stopped:
-            self.time += 1
-            if self.time == timeout:
-                logging.info("Exiting")
-                loop.quit()
-
-        return True
-
-    def reset(self):
-        self.time = 0
-
-    def stop(self):
-        self.stopped = True
-
-    def resume(self):
-        self.stopped = False
-        self.reset()
-
-
-class ConfService(DbusService):
-    def __init__(self):
-        super().__init__("org.blueman.Mechanism", "org.blueman.Mechanism", "/org/blueman/mechanism", Gio.BusType.SYSTEM)
-        self.timer = Timer()
-        if args.stoptimer:
-            self.timer.stop()
-
-        if POLKIT:
-            try:
-                self.pk = Gio.DBusProxy.new_for_bus_sync(
-                    Gio.BusType.SYSTEM,
-                    Gio.DBusProxyFlags.NONE,
-                    None,
-                    'org.freedesktop.PolicyKit1',
-                    '/org/freedesktop/PolicyKit1/Authority',
-                    'org.freedesktop.PolicyKit1.Authority')
-            except Exception as e:
-                logging.exception(e)
-                self.pk = None
-        else:
-            self.pk = None
-
-        path = os.path.dirname(blueman.plugins.mechanism.__file__)
-        plugins = []
-        for root, dirs, files in os.walk(path):
-            for f in files:
-                if f.endswith(".py") and not (f.endswith(".pyc") or f.endswith("_.py")):
-                    plugins.append(f[0:-3])
-
-        for plugin in plugins:
-            try:
-                importlib.import_module(f"blueman.plugins.mechanism.{plugin}")
-            except ImportError:
-                logging.error(f"Skipping plugin {plugin}", exc_info=True)
-
-        classes = MechanismPlugin.__subclasses__()
-        for cls in classes:
-            logging.info(f"loading {cls.__name__}")
-            cls(self)
-
-        self.register()
-
-    def confirm_authorization(self, subject, action_id):
-        self.timer.reset()
-        if not POLKIT:
-            return
-        else:
-            if not self.pk:
-                raise DbusError("Blueman was built with PolicyKit-1 support, but it's not available on the system")
-
-        v_subject = GLib.Variant('s', subject)
-        res = self.pk.CheckAuthorization('((sa{sv})sa{ss}us)', ("system-bus-name", {"name": v_subject}),
-                                         action_id, {}, 1, "")
-
-        logging.debug(str(res))
-        (is_authorized, is_challenge, details) = res
-        if not is_authorized:
-            raise DbusError("Not authorized")
-
-
 set_proc_title()
-ConfService()
-loop.run()
+app = MechanismApplication(args.stoptimer)
+app.run()

blueman/Service.py

@@ -1,10 +1,11 @@
+from abc import ABC, abstractmethod
 from typing import Optional
 
 from blueman.Sdp import ServiceUUID
 from blueman.bluez.Device import Device
 
 
-class Service:
+class Service(ABC):
     __svclass_id__: int
     __description__ = None
     __icon__: str
@@ -41,3 +42,13 @@ def icon(self) -> str:
     @property
     def priority(self) -> int:
         return self.__priority__
+
+    @property
+    @abstractmethod
+    def connected(self) -> bool:
+        ...
+
+    @property
+    @abstractmethod
+    def available(self) -> bool:
+        ...

blueman/main/DhcpClient.py

@@ -54,6 +54,7 @@ def _on_timeout(self):
         if not self._client.poll():
             logging.warning("Timeout reached, terminating DHCP client")
             self._client.terminate()
+        return False
 
     def _check_client(self):
         netifs = get_local_interfaces()
@@ -63,6 +64,7 @@ def complete():
                 ip = netifs[self._interface][0]
                 logging.info(f"bound to {ip}")
                 self.emit("connected", ip)
+                return False
 
             GLib.timeout_add(1000, complete)
             DhcpClient.quering.remove(self._interface)

blueman/main/Manager.py

@@ -29,6 +29,8 @@ class Blueman(Gtk.Application):
     def __init__(self):
         super().__init__(application_id="org.blueman.Manager")
 
+    window: Optional[Gtk.ApplicationWindow]
+
     def do_startup(self):
         def doquit(_a, _param):
             self.quit()

blueman/main/MechanismApplication.py

@@ -0,0 +1,101 @@
+import importlib
+import logging
+import os
+
+import blueman
+from blueman.Constants import POLKIT
+from gi.repository import GLib, Gio
+
+from blueman.main.DbusService import DbusService, DbusError
+from blueman.plugins.MechanismPlugin import MechanismPlugin
+
+
+class Timer:
+    def __init__(self, loop: GLib.MainLoop):
+        self.time = 0
+        self.stopped = False
+        self._loop = loop
+        GLib.timeout_add(1000, self.tick)
+
+    def tick(self):
+        if not self.stopped:
+            self.time += 1
+            if self.time == (9999 if 'BLUEMAN_SOURCE' in os.environ else 30):
+                logging.info("Exiting")
+                self._loop.quit()
+
+        return True
+
+    def reset(self):
+        self.time = 0
+
+    def stop(self):
+        self.stopped = True
+
+    def resume(self):
+        self.stopped = False
+        self.reset()
+
+
+class MechanismApplication(DbusService):
+    def __init__(self, stoptimer: bool):
+        super().__init__("org.blueman.Mechanism", "org.blueman.Mechanism", "/org/blueman/mechanism", Gio.BusType.SYSTEM)
+        self._loop = GLib.MainLoop()
+        self.timer = Timer(self._loop)
+        if stoptimer:
+            self.timer.stop()
+
+        if POLKIT:
+            try:
+                self.pk = Gio.DBusProxy.new_for_bus_sync(
+                    Gio.BusType.SYSTEM,
+                    Gio.DBusProxyFlags.NONE,
+                    None,
+                    'org.freedesktop.PolicyKit1',
+                    '/org/freedesktop/PolicyKit1/Authority',
+                    'org.freedesktop.PolicyKit1.Authority')
+            except Exception as e:
+                logging.exception(e)
+                self.pk = None
+        else:
+            self.pk = None
+
+        path = os.path.dirname(blueman.plugins.mechanism.__file__)
+        plugins = []
+        for root, dirs, files in os.walk(path):
+            for f in files:
+                if f.endswith(".py") and not (f.endswith(".pyc") or f.endswith("_.py")):
+                    plugins.append(f[0:-3])
+
+        for plugin in plugins:
+            try:
+                importlib.import_module(f"blueman.plugins.mechanism.{plugin}")
+            except ImportError:
+                logging.error(f"Skipping plugin {plugin}", exc_info=True)
+
+        classes = MechanismPlugin.__subclasses__()
+        for cls in classes:
+            logging.info(f"loading {cls.__name__}")
+            cls(self)
+
+        self.register()
+
+    def run(self) -> None:
+        self._loop.run()
+
+    def confirm_authorization(self, subject, action_id):
+        self.timer.reset()
+        if not POLKIT:
+            return
+        else:
+            if not self.pk:
+                raise DbusError("Blueman was built with PolicyKit-1 support, but it's not available on the system")
+
+        v_subject = GLib.Variant('s', subject)
+        res = self.pk.CheckAuthorization('((sa{sv})sa{ss}us)', ("system-bus-name", {"name": v_subject}),
+                                         action_id, {}, 1, "")
+
+        logging.debug(str(res))
+        (is_authorized, is_challenge, details) = res
+        if not is_authorized:
+            raise DbusError("Not authorized")

blueman/main/NetConf.py

@@ -8,13 +8,26 @@
 from time import sleep
 import logging
 import signal
-from typing import List, Tuple
+from typing import List, Tuple, TYPE_CHECKING
 
 from blueman.Constants import DHCP_CONFIG_FILE
 from blueman.Functions import have
 from _blueman import create_bridge, destroy_bridge, BridgeException
 from subprocess import call, Popen, PIPE
 
+if TYPE_CHECKING:
+    from typing_extensions import Protocol
+
+    class DHCPHandler(Protocol):
+        def __init__(self, netconf: "NetConf"):
+            ...
+
+        def do_apply(self) -> None:
+            ...
+
+        def do_remove(self) -> None:
+            ...
+
 
 class NetworkSetupError(Exception):
     pass

blueman/main/PulseAudioUtils.py

@@ -375,6 +375,7 @@ def connect_pulseaudio(self):
             pa_context_set_subscribe_callback(self.pa_context,
                                               self.event_cb,
                                               None)
+        return False
 
     def _on_delete(self):
         logging.info("Destroying PulseAudioUtils instance")