mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-12-24 20:35:55 +00:00
Use dbus-fast unpack_variants option (#3885)
* Use dbus-fast unpack_variants option * More readable log on signals
This commit is contained in:
@@ -171,7 +171,7 @@ def dbus(dbus_bus: MessageBus) -> DBus:
|
||||
proxy_interface: ProxyInterface,
|
||||
method: str,
|
||||
*args,
|
||||
remove_signature: bool = True,
|
||||
unpack_variants: bool = True,
|
||||
):
|
||||
if (
|
||||
proxy_interface.introspection.name == DBUS_INTERFACE_PROPERTIES
|
||||
|
||||
@@ -72,10 +72,10 @@ SETTINGS_WITH_SIGNATURE = {
|
||||
|
||||
|
||||
async def mock_call_dbus_get_settings_signature(
|
||||
_: ProxyInterface, method: str, *args, remove_signature: bool = True
|
||||
_: ProxyInterface, method: str, *args, unpack_variants: bool = True
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Call dbus method mock for get settings that keeps signature."""
|
||||
if method == "call_get_settings" and not remove_signature:
|
||||
if method == "call_get_settings" and not unpack_variants:
|
||||
return SETTINGS_WITH_SIGNATURE
|
||||
else:
|
||||
assert method == "call_update"
|
||||
|
||||
@@ -80,7 +80,7 @@ async def test_signal_listener_disconnect(
|
||||
device = dev
|
||||
|
||||
proxy.obj.dbus.on_device_added(callback)
|
||||
proxy.on_device_added.assert_called_once_with(callback)
|
||||
proxy.on_device_added.assert_called_once_with(callback, unpack_variants=True)
|
||||
|
||||
fire_watched_signal(
|
||||
proxy.obj, "org.freedesktop.NetworkManager.DeviceAdded", ["/test/obj/1"]
|
||||
@@ -89,4 +89,4 @@ async def test_signal_listener_disconnect(
|
||||
assert device == "/test/obj/1"
|
||||
|
||||
proxy.obj.disconnect()
|
||||
proxy.off_device_added.assert_called_once_with(callback)
|
||||
proxy.off_device_added.assert_called_once_with(callback, unpack_variants=True)
|
||||
|
||||
@@ -140,7 +140,7 @@ async def test_scan_wifi_with_failures(coresys: CoreSys, caplog):
|
||||
proxy_interface: ProxyInterface,
|
||||
method: str,
|
||||
*args,
|
||||
remove_signature: bool = True,
|
||||
unpack_variants: bool = True,
|
||||
):
|
||||
if method == "call_get_all_access_points":
|
||||
return [
|
||||
@@ -150,7 +150,7 @@ async def test_scan_wifi_with_failures(coresys: CoreSys, caplog):
|
||||
]
|
||||
|
||||
return await call_dbus(
|
||||
proxy_interface, method, *args, remove_signature=remove_signature
|
||||
proxy_interface, method, *args, unpack_variants=unpack_variants
|
||||
)
|
||||
|
||||
with patch("supervisor.host.network.asyncio.sleep"), patch(
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
"""Check dbus-next implementation."""
|
||||
from dbus_fast.signature import Variant
|
||||
|
||||
from supervisor.utils.dbus import DBus
|
||||
|
||||
|
||||
def test_remove_dbus_signature():
|
||||
"""Check D-Bus signature clean-up."""
|
||||
test = DBus.remove_dbus_signature(Variant("s", "Value"))
|
||||
assert isinstance(test, str)
|
||||
assert test == "Value"
|
||||
|
||||
test_dict = DBus.remove_dbus_signature({"Key": Variant("s", "Value")})
|
||||
assert isinstance(test_dict["Key"], str)
|
||||
assert test_dict["Key"] == "Value"
|
||||
|
||||
test_dict = DBus.remove_dbus_signature([Variant("s", "Value")])
|
||||
assert isinstance(test_dict[0], str)
|
||||
assert test_dict[0] == "Value"
|
||||
Reference in New Issue
Block a user