Save the dbus connection in the adapter instance

pyproject.toml

@@ -104,6 +104,7 @@ exclude_lines = [
     "if (?:typing\\.)?TYPE_CHECKING:",
     "@(abc\\.)?abstractmethod",
     "@(typing\\.)?overload",
+    "raise NotImplementedError"
 ]
 
 [tool.coverage.coverage_conditional_plugin.omit]

src/wakepy/core/dbus.py

@@ -279,6 +279,10 @@ class DBusAdapter:
     wakepy (Modes).
     """
 
+    def __init__(self) -> None:
+        # The values are DBusConnections. Type is defined by the used library.
+        self._connections: Dict[Optional[Union[str, BusType]], object] = dict()
+
     def process(self, call: DBusMethodCall) -> object:
         """Processes a :class:`~wakepy.core.DBusMethodCall`.
 
@@ -290,6 +294,29 @@ def process(self, call: DBusMethodCall) -> object:
           created within the :func:`~wakepy.DBusAdapter.process` call (this may
           of course be cached)."""
 
+    def _get_connection(
+        self, bus: Optional[Union[str, BusType]] = BusType.SESSION
+    ) -> object:
+        """Gets either a new connection or a cached one, if there is such.
+        Caching of connections is done on bus level."""
+
+        if bus in self._connections:
+            return self._connections[bus]
+
+        connection = self._create_new_connection(bus)
+
+        self._connections[bus] = connection
+        return connection
+
+    def _create_new_connection(
+        self, bus: Optional[Union[str, BusType]] = BusType.SESSION
+    ) -> object:
+        """Create a new Dbus connection for a bus using the library of choice.
+        For example, when creating DBusAdapter subclass for jeepney, could
+        return an instance of ``jeepney.io.blocking.DBusConnection``.
+        """
+        raise NotImplementedError("Implement in subclass")
+
 
 def get_dbus_adapter(
     dbus_adapter: Optional[Type[DBusAdapter] | DBusAdapterTypeSeq] = None,

src/wakepy/dbus_adapters/jeepney.py

@@ -1,18 +1,17 @@
 from __future__ import annotations
 
 import typing
+from typing import cast
 
 from jeepney import DBusAddress, new_method_call
-from jeepney.io.blocking import open_dbus_connection
+from jeepney.io.blocking import DBusConnection, open_dbus_connection
 from jeepney.wrappers import unwrap_msg
 
 from wakepy.core import DBusAdapter, DBusMethodCall
 from wakepy.core.dbus import BusType
 
 if typing.TYPE_CHECKING:
-    from typing import Dict, Optional, Union
-
-    from jeepney.io.blocking import DBusConnection
+    from typing import Optional, Union
 
     TypeOfBus = Optional[Union[BusType, str]]
 
@@ -29,9 +28,6 @@ class JeepneyDBusAdapter(DBusAdapter):
     # timeout for dbus calls, in seconds
     timeout = 2
 
-    def __init__(self) -> None:
-        self._connections: Dict[TypeOfBus, DBusConnection] = dict()  # type: ignore[no-any-unimported]
-
     def process(self, call: DBusMethodCall) -> object:
         addr = DBusAddress(
             object_path=call.method.path,
@@ -46,25 +42,11 @@ def process(self, call: DBusMethodCall) -> object:
             body=call.args,
         )
 
-        connection = self._get_connection(call.method.bus)
+        connection = cast(DBusConnection, self._get_connection(call.method.bus))  # type: ignore[no-any-unimported]
         reply = connection.send_and_get_reply(msg, timeout=self.timeout)
         resp = unwrap_msg(reply)
         return resp
 
-    def _get_connection(  # type: ignore[no-any-unimported]
-        self, bus: TypeOfBus = BusType.SESSION
-    ) -> DBusConnection:
-        """Gets either a new connection or a cached one, if there is such.
-        Caching of connections is done on bus level."""
-
-        if bus in self._connections:
-            return self._connections[bus]
-
-        connection = self._create_new_connection(bus)
-
-        self._connections[bus] = connection
-        return connection
-
     def _create_new_connection(  # type: ignore[no-any-unimported]
         self, bus: TypeOfBus = BusType.SESSION
     ) -> DBusConnection:

tests/integration/conftest.py

@@ -6,6 +6,7 @@
 import logging
 import subprocess
 import sys
+import warnings
 
 import pytest
 
@@ -32,7 +33,11 @@ def gc_collect_after_dbus_integration_tests():
     # this as garbage colletion is triggered also automatically. The garbage
     # collection must be triggered here manually as the warnings are
     # ResourceWarning is only filtered away in the dbus integration tests.
-    gc.collect()
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        gc.collect()
+
     logger.debug("called gc.collect")