From 0ddce4d9bc00ae71a3c45d56d9c5c7e13b7f82d8 Mon Sep 17 00:00:00 2001 From: Mike Degatano Date: Fri, 17 Mar 2023 09:26:41 -0400 Subject: [PATCH] Use session dbus for network tests (#4191) * Use session dbus for network tests * Don't use session dbus for coresys yet --- supervisor/dbus/const.py | 1 + supervisor/dbus/network/__init__.py | 2 +- supervisor/dbus/network/interface.py | 7 +- supervisor/dbus/network/wireless.py | 9 +- tests/conftest.py | 43 +- tests/dbus/network/setting/test_generate.py | 28 ++ tests/dbus/network/setting/test_init.py | 246 +++++------- tests/dbus/network/test_accesspoint.py | 34 +- tests/dbus/network/test_connection.py | 73 +++- tests/dbus/network/test_dns.py | 50 ++- tests/dbus/network/test_interface.py | 140 +++++-- tests/dbus/network/test_ip_configuration.py | 99 +++-- tests/dbus/network/test_network_manager.py | 115 +++--- tests/dbus/network/test_setting.py | 59 --- tests/dbus/network/test_settings.py | 64 ++- tests/dbus/network/test_wireless.py | 82 ++-- tests/dbus_service_mocks/hostname.py | 2 +- .../network_access_point.py | 128 ++++++ .../network_active_connection.py | 110 ++++++ .../network_connection_settings.py | 144 +++++++ tests/dbus_service_mocks/network_device.py | 371 ++++++++++++++++++ .../network_device_wireless.py | 102 +++++ .../dbus_service_mocks/network_dns_manager.py | 49 +++ tests/dbus_service_mocks/network_ip4config.py | 108 +++++ tests/dbus_service_mocks/network_ip6config.py | 369 +++++++++++++++++ tests/dbus_service_mocks/network_manager.py | 328 ++++++++++++++++ tests/dbus_service_mocks/network_settings.py | 94 +++++ tests/dbus_service_mocks/resolved.py | 110 +++++- 28 files changed, 2547 insertions(+), 420 deletions(-) create mode 100644 tests/dbus/network/setting/test_generate.py delete mode 100644 tests/dbus/network/test_setting.py create mode 100644 tests/dbus_service_mocks/network_access_point.py create mode 100644 tests/dbus_service_mocks/network_active_connection.py create mode 100644 tests/dbus_service_mocks/network_connection_settings.py create mode 100644 tests/dbus_service_mocks/network_device.py create mode 100644 tests/dbus_service_mocks/network_device_wireless.py create mode 100644 tests/dbus_service_mocks/network_dns_manager.py create mode 100644 tests/dbus_service_mocks/network_ip4config.py create mode 100644 tests/dbus_service_mocks/network_ip6config.py create mode 100644 tests/dbus_service_mocks/network_manager.py create mode 100644 tests/dbus_service_mocks/network_settings.py diff --git a/supervisor/dbus/const.py b/supervisor/dbus/const.py index a75aef2d9..093008762 100644 --- a/supervisor/dbus/const.py +++ b/supervisor/dbus/const.py @@ -64,6 +64,7 @@ DBUS_ATTR_ACTIVE_ACCESSPOINT = "ActiveAccessPoint" DBUS_ATTR_ACTIVE_CONNECTION = "ActiveConnection" DBUS_ATTR_ACTIVE_CONNECTIONS = "ActiveConnections" DBUS_ATTR_ADDRESS_DATA = "AddressData" +DBUS_ATTR_BITRATE = "Bitrate" DBUS_ATTR_BOARD = "Board" DBUS_ATTR_BOOT_SLOT = "BootSlot" DBUS_ATTR_CACHE_STATISTICS = "CacheStatistics" diff --git a/supervisor/dbus/network/__init__.py b/supervisor/dbus/network/__init__.py index 610ad9ae6..cb9fa1cad 100644 --- a/supervisor/dbus/network/__init__.py +++ b/supervisor/dbus/network/__init__.py @@ -184,7 +184,7 @@ class NetworkManager(DBusInterfaceProxy): interface = curr_devices[device] await interface.update() else: - interface = NetworkInterface(self.dbus, device) + interface = NetworkInterface(device) # Connect to interface try: diff --git a/supervisor/dbus/network/interface.py b/supervisor/dbus/network/interface.py index 373cd16a2..b4b0423e1 100644 --- a/supervisor/dbus/network/interface.py +++ b/supervisor/dbus/network/interface.py @@ -4,7 +4,6 @@ from typing import Any from dbus_fast.aio.message_bus import MessageBus -from ...utils.dbus import DBus from ..const import ( DBUS_ATTR_ACTIVE_CONNECTION, DBUS_ATTR_DEVICE_INTERFACE, @@ -33,7 +32,7 @@ class NetworkInterface(DBusInterfaceProxy): properties_interface: str = DBUS_IFACE_DEVICE sync_properties: bool = False - def __init__(self, nm_dbus: DBus, object_path: str) -> None: + def __init__(self, object_path: str) -> None: """Initialize NetworkConnection object.""" super().__init__() @@ -43,7 +42,6 @@ class NetworkInterface(DBusInterfaceProxy): self._connection: NetworkConnection | None = None self._wireless: NetworkWireless | None = None - self._nm_dbus: DBus = nm_dbus @property @dbus_property @@ -114,7 +112,10 @@ class NetworkInterface(DBusInterfaceProxy): await super().update(changed) # Abort if device is not managed + # Shutdown and disconnect managed objects if it became unmanaged if not self.managed: + self.connection = None + self.wireless = None return # If active connection exists diff --git a/supervisor/dbus/network/wireless.py b/supervisor/dbus/network/wireless.py index 99f19aba8..8d3141417 100644 --- a/supervisor/dbus/network/wireless.py +++ b/supervisor/dbus/network/wireless.py @@ -5,11 +5,12 @@ from typing import Any from ..const import ( DBUS_ATTR_ACTIVE_ACCESSPOINT, + DBUS_ATTR_BITRATE, DBUS_IFACE_DEVICE_WIRELESS, DBUS_NAME_NM, DBUS_OBJECT_BASE, ) -from ..interface import DBusInterfaceProxy +from ..interface import DBusInterfaceProxy, dbus_property from ..utils import dbus_connected from .accesspoint import NetworkWirelessAP @@ -33,6 +34,12 @@ class NetworkWireless(DBusInterfaceProxy): self._active: NetworkWirelessAP | None = None + @property + @dbus_property + def bitrate(self) -> int: + """Bitrate currently used by wireless device in Kb/s.""" + return self.properties[DBUS_ATTR_BITRATE] + @property def active(self) -> NetworkWirelessAP | None: """Return details about active connection.""" diff --git a/tests/conftest.py b/tests/conftest.py index 565c04413..32da58118 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -327,15 +327,41 @@ async def dbus_is_connected(): yield is_connected +@pytest.fixture(name="network_manager_services") +async def fixture_network_manager_services( + dbus_session_bus: MessageBus, +) -> dict[str, DBusServiceMock]: + """Mock all services network manager connects to.""" + yield await mock_dbus_services( + { + "network_access_point": [ + "/org/freedesktop/NetworkManager/AccessPoint/43099", + "/org/freedesktop/NetworkManager/AccessPoint/43100", + ], + "network_active_connection": None, + "network_connection_settings": None, + "network_device_wireless": None, + "network_device": [ + "/org/freedesktop/NetworkManager/Devices/1", + "/org/freedesktop/NetworkManager/Devices/3", + ], + "network_dns_manager": None, + "network_ip4config": None, + "network_ip6config": None, + "network_manager": None, + "network_settings": None, + }, + dbus_session_bus, + ) + + @pytest.fixture -async def network_manager(dbus, dbus_bus: MessageBus) -> NetworkManager: +async def network_manager( + network_manager_services: dict[str, DBusServiceMock], dbus_session_bus: MessageBus +) -> NetworkManager: """Mock NetworkManager.""" nm_obj = NetworkManager() - nm_obj.dbus = dbus - - # Init - await nm_obj.connect(dbus_bus) - + await nm_obj.connect(dbus_session_bus) yield nm_obj @@ -386,7 +412,7 @@ async def udisks2(dbus: DBus, dbus_bus: MessageBus) -> UDisks2: @pytest.fixture async def coresys( - event_loop, docker, network_manager, dbus_bus, aiohttp_client, run_dir + event_loop, docker, dbus, dbus_bus, aiohttp_client, run_dir ) -> CoreSys: """Create a CoreSys Mock.""" with patch("supervisor.bootstrap.initialize_system"), patch( @@ -411,6 +437,9 @@ async def coresys( # Mock host communication coresys_obj._dbus._bus = dbus_bus + network_manager = NetworkManager() + network_manager.dbus = dbus + await network_manager.connect(dbus_bus) coresys_obj._dbus._network = network_manager # Mock docker diff --git a/tests/dbus/network/setting/test_generate.py b/tests/dbus/network/setting/test_generate.py new file mode 100644 index 000000000..8078aa4d5 --- /dev/null +++ b/tests/dbus/network/setting/test_generate.py @@ -0,0 +1,28 @@ +"""Test settings generation from interface.""" + +import pytest + +from supervisor.dbus.network import NetworkManager +from supervisor.dbus.network.setting.generate import get_connection_from_interface +from supervisor.host.network import Interface + +from tests.const import TEST_INTERFACE + + +@pytest.mark.asyncio +async def test_get_connection_from_interface(network_manager: NetworkManager): + """Test network interface.""" + dbus_interface = network_manager.interfaces[TEST_INTERFACE] + interface = Interface.from_dbus_interface(dbus_interface) + connection_payload = get_connection_from_interface(interface) + + assert "connection" in connection_payload + + assert connection_payload["connection"]["interface-name"].value == TEST_INTERFACE + assert connection_payload["connection"]["type"].value == "802-3-ethernet" + + assert connection_payload["ipv4"]["method"].value == "auto" + assert "address-data" not in connection_payload["ipv4"] + + assert connection_payload["ipv6"]["method"].value == "auto" + assert "address-data" not in connection_payload["ipv6"] diff --git a/tests/dbus/network/setting/test_init.py b/tests/dbus/network/setting/test_init.py index c4532f799..856ab4cfb 100644 --- a/tests/dbus/network/setting/test_init.py +++ b/tests/dbus/network/setting/test_init.py @@ -1,185 +1,125 @@ """Test Network Manager Connection object.""" -import asyncio -from typing import Any -from unittest.mock import patch -from dbus_fast.aio.proxy_object import ProxyInterface +from dbus_fast.aio.message_bus import MessageBus from dbus_fast.signature import Variant +import pytest -from supervisor.coresys import CoreSys +from supervisor.dbus.network.interface import NetworkInterface +from supervisor.dbus.network.setting import NetworkSetting from supervisor.dbus.network.setting.generate import get_connection_from_interface from supervisor.host.const import InterfaceMethod from supervisor.host.network import Interface -from supervisor.utils.dbus import DBus -from tests.common import fire_watched_signal -from tests.const import TEST_INTERFACE - -SETTINGS_WITH_SIGNATURE = { - "connection": { - "id": Variant("s", "Wired connection 1"), - "interface-name": Variant("s", "eth0"), - "permissions": Variant("as", []), - "timestamp": Variant("t", 1598125548), - "type": Variant("s", "802-3-ethernet"), - "uuid": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"), - }, - "ipv4": { - "address-data": Variant( - "aa{sv}", - [ - { - "address": Variant("s", "192.168.2.148"), - "prefix": Variant("u", 24), - } - ], - ), - "addresses": Variant("aau", [[2483202240, 24, 16951488]]), - "dns": Variant("au", [16951488]), - "dns-search": Variant("as", []), - "gateway": Variant("s", "192.168.2.1"), - "method": Variant("s", "auto"), - "route-data": Variant( - "aa{sv}", - [ - { - "dest": Variant("s", "192.168.122.0"), - "prefix": Variant("u", 24), - "next-hop": Variant("s", "10.10.10.1"), - } - ], - ), - "routes": Variant("aau", [[8038592, 24, 17435146, 0]]), - }, - "ipv6": { - "address-data": Variant("aa{sv}", []), - "addresses": Variant("a(ayuay)", []), - "dns": Variant("au", []), - "dns-search": Variant("as", []), - "method": Variant("s", "auto"), - "route-data": Variant("aa{sv}", []), - "routes": Variant("aau", []), - "addr-gen-mode": Variant("i", 0), - }, - "proxy": {}, - "802-3-ethernet": { - "auto-negotiate": Variant("b", False), - "mac-address-blacklist": Variant("as", []), - "s390-options": Variant("a{ss}", {}), - }, - "802-11-wireless": {"ssid": Variant("ay", bytes([78, 69, 84, 84]))}, -} +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.network_connection_settings import ( + ConnectionSettings as ConnectionSettingsService, +) -async def mock_call_dbus_get_settings_signature( - _: 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 unpack_variants: - return SETTINGS_WITH_SIGNATURE - else: - assert method == "call_update" - settings = args[0] - - assert "connection" in settings - assert settings["connection"]["id"] == Variant("s", "Supervisor eth0") - assert settings["connection"]["interface-name"] == Variant("s", "eth0") - assert settings["connection"]["uuid"] == Variant( - "s", "0c23631e-2118-355c-bbb0-8943229cb0d6" - ) - assert settings["connection"]["autoconnect"] == Variant("b", True) - - assert "ipv4" in settings - assert settings["ipv4"]["method"] == Variant("s", "auto") - assert "gateway" not in settings["ipv4"] - assert "dns" not in settings["ipv4"] - assert "address-data" not in settings["ipv4"] - assert "addresses" not in settings["ipv4"] - assert len(settings["ipv4"]["route-data"].value) == 1 - assert settings["ipv4"]["route-data"].value[0]["dest"] == Variant( - "s", "192.168.122.0" - ) - assert settings["ipv4"]["route-data"].value[0]["prefix"] == Variant("u", 24) - assert settings["ipv4"]["route-data"].value[0]["next-hop"] == Variant( - "s", "10.10.10.1" - ) - assert settings["ipv4"]["routes"] == Variant( - "aau", [[8038592, 24, 17435146, 0]] - ) - - assert "ipv6" in settings - assert settings["ipv6"]["method"] == Variant("s", "auto") - assert "gateway" not in settings["ipv6"] - assert "dns" not in settings["ipv6"] - assert "address-data" not in settings["ipv6"] - assert "addresses" not in settings["ipv6"] - assert settings["ipv6"]["addr-gen-mode"] == Variant("i", 0) - - assert "proxy" in settings - - assert "802-3-ethernet" in settings - assert settings["802-3-ethernet"]["auto-negotiate"] == Variant("b", False) - - assert "802-11-wireless" in settings - assert settings["802-11-wireless"]["ssid"] == Variant( - "ay", bytes([78, 69, 84, 84]) - ) - assert "mode" not in settings["802-11-wireless"] - assert "powersave" not in settings["802-11-wireless"] - - assert "802-11-wireless-security" not in settings - assert "vlan" not in settings +@pytest.fixture(name="connection_settings_service", autouse=True) +async def fixture_connection_settings_service( + network_manager_services: dict[str, DBusServiceMock] +) -> ConnectionSettingsService: + """Mock Connection Settings service.""" + yield network_manager_services["network_connection_settings"] -async def test_update(coresys: CoreSys): +@pytest.fixture(name="dbus_interface") +async def fixture_dbus_interface(dbus_session_bus: MessageBus) -> NetworkInterface: + """Get connected dbus interface.""" + dbus_interface = NetworkInterface("/org/freedesktop/NetworkManager/Devices/1") + await dbus_interface.connect(dbus_session_bus) + yield dbus_interface + + +async def test_update( + dbus_interface: NetworkInterface, + connection_settings_service: ConnectionSettingsService, +): """Test network manager update.""" - await coresys.dbus.network.interfaces[TEST_INTERFACE].connect(coresys.dbus.bus) - interface = Interface.from_dbus_interface( - coresys.dbus.network.interfaces[TEST_INTERFACE] - ) + interface = Interface.from_dbus_interface(dbus_interface) conn = get_connection_from_interface( interface, - name=coresys.dbus.network.interfaces[TEST_INTERFACE].settings.connection.id, - uuid=coresys.dbus.network.interfaces[TEST_INTERFACE].settings.connection.uuid, + name=dbus_interface.settings.connection.id, + uuid=dbus_interface.settings.connection.uuid, ) - with patch.object( - DBus, - "call_dbus", - new=mock_call_dbus_get_settings_signature, - ): - await coresys.dbus.network.interfaces[TEST_INTERFACE].settings.update(conn) + await dbus_interface.settings.update(conn) + + assert len(connection_settings_service.Update.calls) == 1 + settings = connection_settings_service.Update.calls[0][0] + + assert settings["connection"]["id"] == Variant("s", "Supervisor eth0") + assert settings["connection"]["interface-name"] == Variant("s", "eth0") + assert settings["connection"]["uuid"] == Variant( + "s", "0c23631e-2118-355c-bbb0-8943229cb0d6" + ) + assert settings["connection"]["autoconnect"] == Variant("b", True) + + assert "ipv4" in settings + assert settings["ipv4"]["method"] == Variant("s", "auto") + assert "gateway" not in settings["ipv4"] + assert "dns" not in settings["ipv4"] + assert "address-data" not in settings["ipv4"] + assert "addresses" not in settings["ipv4"] + assert len(settings["ipv4"]["route-data"].value) == 1 + assert settings["ipv4"]["route-data"].value[0]["dest"] == Variant( + "s", "192.168.122.0" + ) + assert settings["ipv4"]["route-data"].value[0]["prefix"] == Variant("u", 24) + assert settings["ipv4"]["route-data"].value[0]["next-hop"] == Variant( + "s", "10.10.10.1" + ) + assert settings["ipv4"]["routes"] == Variant("aau", [[8038592, 24, 17435146, 0]]) + + assert "ipv6" in settings + assert settings["ipv6"]["method"] == Variant("s", "auto") + assert "gateway" not in settings["ipv6"] + assert "dns" not in settings["ipv6"] + assert "address-data" not in settings["ipv6"] + assert "addresses" not in settings["ipv6"] + assert settings["ipv6"]["addr-gen-mode"] == Variant("i", 0) + + assert "proxy" in settings + + assert "802-3-ethernet" in settings + assert settings["802-3-ethernet"]["auto-negotiate"] == Variant("b", False) + + assert "802-11-wireless" in settings + assert settings["802-11-wireless"]["ssid"] == Variant("ay", b"NETT") + assert "mode" not in settings["802-11-wireless"] + assert "powersave" not in settings["802-11-wireless"] + + assert "802-11-wireless-security" not in settings + assert "vlan" not in settings -async def test_ipv6_disabled_is_link_local(coresys: CoreSys): +async def test_ipv6_disabled_is_link_local(dbus_interface: NetworkInterface): """Test disabled equals link local for ipv6.""" - await coresys.dbus.network.interfaces[TEST_INTERFACE].connect(coresys.dbus.bus) - interface = Interface.from_dbus_interface( - coresys.dbus.network.interfaces[TEST_INTERFACE] - ) + interface = Interface.from_dbus_interface(dbus_interface) interface.ipv4.method = InterfaceMethod.DISABLED interface.ipv6.method = InterfaceMethod.DISABLED conn = get_connection_from_interface( interface, - name=coresys.dbus.network.interfaces[TEST_INTERFACE].settings.connection.id, - uuid=coresys.dbus.network.interfaces[TEST_INTERFACE].settings.connection.uuid, + name=dbus_interface.settings.connection.id, + uuid=dbus_interface.settings.connection.uuid, ) assert conn["ipv4"]["method"] == Variant("s", "disabled") assert conn["ipv6"]["method"] == Variant("s", "link-local") -async def test_watching_updated_signal(coresys: CoreSys, dbus: list[str]): +async def test_watching_updated_signal( + connection_settings_service: ConnectionSettingsService, dbus_session_bus: MessageBus +): """Test get settings called on update signal.""" - await coresys.dbus.network.interfaces[TEST_INTERFACE].connect(coresys.dbus.bus) - dbus.clear() + connection_settings_service.GetSettings.calls.clear() + settings = NetworkSetting("/org/freedesktop/NetworkManager/Settings/1") + await settings.connect(dbus_session_bus) - fire_watched_signal( - coresys.dbus.network.interfaces[TEST_INTERFACE].settings, - "org.freedesktop.NetworkManager.Settings.Connection.Updated", - [], - ) - await asyncio.sleep(0) - assert dbus == [ - "/org/freedesktop/NetworkManager/Settings/1-org.freedesktop.NetworkManager.Settings.Connection.GetSettings" - ] + connection_settings_service.GetSettings.calls == [tuple()] + + connection_settings_service.Updated() + await connection_settings_service.ping() + await connection_settings_service.ping() + assert connection_settings_service.GetSettings.calls == [tuple(), tuple()] diff --git a/tests/dbus/network/test_accesspoint.py b/tests/dbus/network/test_accesspoint.py index 8688aa662..bc855ebe0 100644 --- a/tests/dbus/network/test_accesspoint.py +++ b/tests/dbus/network/test_accesspoint.py @@ -1,17 +1,47 @@ """Test NetworkWireless AP object.""" + from dbus_fast.aio.message_bus import MessageBus +import pytest from supervisor.dbus.network.accesspoint import NetworkWirelessAP +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.network_access_point import ( + AccessPoint as AccessPointService, +) -async def test_accesspoint(dbus: list[str], dbus_bus: MessageBus): + +@pytest.fixture(name="access_point_service", autouse=True) +async def fixture_access_point_service( + dbus_session_bus: MessageBus, +) -> AccessPointService: + """Mock Access Point service.""" + yield ( + await mock_dbus_services( + { + "network_access_point": "/org/freedesktop/NetworkManager/AccessPoint/43099" + }, + dbus_session_bus, + ) + )["network_access_point"] + + +async def test_accesspoint( + access_point_service: AccessPointService, dbus_session_bus: MessageBus +): """Test accesspoint.""" wireless_ap = NetworkWirelessAP("/org/freedesktop/NetworkManager/AccessPoint/43099") assert wireless_ap.mac is None assert wireless_ap.mode is None - await wireless_ap.connect(dbus_bus) + await wireless_ap.connect(dbus_session_bus) assert wireless_ap.mac == "E4:57:40:A9:D7:DE" assert wireless_ap.mode == 2 + assert wireless_ap.strength == 47 + + # We don't listen for property changes on access points, too noisy + access_point_service.emit_properties_changed({"Strength": 74}) + await access_point_service.ping() + assert wireless_ap.strength == 47 diff --git a/tests/dbus/network/test_connection.py b/tests/dbus/network/test_connection.py index 49dcfccba..a4718c60d 100644 --- a/tests/dbus/network/test_connection.py +++ b/tests/dbus/network/test_connection.py @@ -1,47 +1,98 @@ """Test connection object.""" -import asyncio +from dbus_fast.aio.message_bus import MessageBus +import pytest +from supervisor.dbus.const import ConnectionStateFlags from supervisor.dbus.network import NetworkManager +from supervisor.dbus.network.connection import NetworkConnection -from tests.common import fire_property_change_signal from tests.const import TEST_INTERFACE +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.network_active_connection import ( + ActiveConnection as ActiveConnectionService, +) -async def test_old_ipv4_disconnect(network_manager: NetworkManager): +@pytest.fixture(name="active_connection_service", autouse=True) +async def fixture_active_connection_service( + network_manager_services: dict[str, DBusServiceMock] +) -> ActiveConnectionService: + """Mock Active Connection service.""" + yield network_manager_services["network_active_connection"] + + +async def test_active_connection( + active_connection_service: ActiveConnectionService, dbus_session_bus: MessageBus +): + """Test active connection.""" + active_connection = NetworkConnection( + "/org/freedesktop/NetworkManager/ActiveConnection/1" + ) + + assert active_connection.id is None + + await active_connection.connect(dbus_session_bus) + + assert active_connection.id == "Wired connection 1" + assert active_connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" + assert active_connection.state_flags == { + ConnectionStateFlags.LIFETIME_BOUND_TO_PROFILE_VISIBILITY, + ConnectionStateFlags.IP6_READY, + ConnectionStateFlags.IP4_READY, + ConnectionStateFlags.LAYER2_READY, + } + + active_connection_service.emit_properties_changed({"Id": "Wired connection 2"}) + await active_connection_service.ping() + assert active_connection.id == "Wired connection 2" + + active_connection_service.emit_properties_changed({}, ["Id"]) + await active_connection_service.ping() + await active_connection_service.ping() + assert active_connection.id == "Wired connection 1" + + +async def test_old_ipv4_disconnect( + network_manager: NetworkManager, active_connection_service: ActiveConnectionService +): """Test old ipv4 disconnects on ipv4 change.""" connection = network_manager.interfaces[TEST_INTERFACE].connection ipv4 = connection.ipv4 assert ipv4.is_connected is True - fire_property_change_signal(connection, {"Ip4Config": "/"}) - await asyncio.sleep(0) + active_connection_service.emit_properties_changed({"Ip4Config": "/"}) + await active_connection_service.ping() assert connection.ipv4 is None assert ipv4.is_connected is False -async def test_old_ipv6_disconnect(network_manager: NetworkManager): +async def test_old_ipv6_disconnect( + network_manager: NetworkManager, active_connection_service: ActiveConnectionService +): """Test old ipv6 disconnects on ipv6 change.""" connection = network_manager.interfaces[TEST_INTERFACE].connection ipv6 = connection.ipv6 assert ipv6.is_connected is True - fire_property_change_signal(connection, {"Ip6Config": "/"}) - await asyncio.sleep(0) + active_connection_service.emit_properties_changed({"Ip6Config": "/"}) + await active_connection_service.ping() assert connection.ipv6 is None assert ipv6.is_connected is False -async def test_old_settings_disconnect(network_manager: NetworkManager): +async def test_old_settings_disconnect( + network_manager: NetworkManager, active_connection_service: ActiveConnectionService +): """Test old settings disconnects on settings change.""" connection = network_manager.interfaces[TEST_INTERFACE].connection settings = connection.settings assert settings.is_connected is True - fire_property_change_signal(connection, {"Connection": "/"}) - await asyncio.sleep(0) + active_connection_service.emit_properties_changed({"Connection": "/"}) + await active_connection_service.ping() assert connection.settings is None assert settings.is_connected is False diff --git a/tests/dbus/network/test_dns.py b/tests/dbus/network/test_dns.py index 40bcb8fad..8fe3cd4d7 100644 --- a/tests/dbus/network/test_dns.py +++ b/tests/dbus/network/test_dns.py @@ -1,27 +1,51 @@ """Test DNS Manager object.""" -import asyncio + from ipaddress import IPv4Address -from supervisor.dbus.network import NetworkManager +from dbus_fast.aio.message_bus import MessageBus +import pytest + from supervisor.dbus.network.configuration import DNSConfiguration +from supervisor.dbus.network.dns import NetworkManagerDNS -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.network_dns_manager import DnsManager as DnsManagerService -async def test_dns(network_manager: NetworkManager): +@pytest.fixture(name="dns_manager_service", autouse=True) +async def fixture_dns_manager_service( + dbus_session_bus: MessageBus, +) -> DnsManagerService: + """Mock DnsManager dbus service.""" + yield (await mock_dbus_services({"network_dns_manager": None}, dbus_session_bus))[ + "network_dns_manager" + ] + + +async def test_dns( + dns_manager_service: DnsManagerService, dbus_session_bus: MessageBus +): """Test dns manager.""" - assert network_manager.dns.mode == "default" - assert network_manager.dns.rc_manager == "file" - assert network_manager.dns.configuration == [ + dns_manager = NetworkManagerDNS() + + assert dns_manager.mode is None + assert dns_manager.rc_manager is None + + await dns_manager.connect(dbus_session_bus) + + assert dns_manager.mode == "default" + assert dns_manager.rc_manager == "file" + assert dns_manager.configuration == [ DNSConfiguration( [IPv4Address("192.168.30.1")], ["syshack.ch"], "eth0", 100, False ) ] - fire_property_change_signal(network_manager.dns, {"Mode": "test"}) - await asyncio.sleep(0) - assert network_manager.dns.mode == "test" + dns_manager_service.emit_properties_changed({"Mode": "test"}) + await dns_manager_service.ping() + assert dns_manager.mode == "test" - fire_property_change_signal(network_manager.dns, {}, ["Mode"]) - await asyncio.sleep(0) - assert network_manager.dns.mode == "default" + dns_manager_service.emit_properties_changed({}, ["Mode"]) + await dns_manager_service.ping() + await dns_manager_service.ping() + assert dns_manager.mode == "default" diff --git a/tests/dbus/network/test_interface.py b/tests/dbus/network/test_interface.py index 18c96a8df..141260497 100644 --- a/tests/dbus/network/test_interface.py +++ b/tests/dbus/network/test_interface.py @@ -1,34 +1,73 @@ """Test NetwrokInterface.""" -import asyncio + from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface +from dbus_fast.aio.message_bus import MessageBus import pytest from supervisor.dbus.const import DeviceType, InterfaceMethod from supervisor.dbus.network import NetworkManager from supervisor.dbus.network.interface import NetworkInterface -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.network_device import Device as DeviceService -@pytest.mark.asyncio -async def test_network_interface_ethernet(network_manager: NetworkManager): +@pytest.fixture(name="device_eth0_service") +async def fixture_device_eth0_service( + network_manager_services: dict[str, DBusServiceMock] +) -> DeviceService: + """Mock Device eth0 service.""" + yield network_manager_services["network_device"][0] + + +@pytest.fixture(name="device_wlan0_service") +async def fixture_device_wlan0_service( + network_manager_services: dict[str, DBusServiceMock] +) -> DeviceService: + """Mock Device wlan0 service.""" + yield network_manager_services["network_device"][1] + + +@pytest.fixture(name="device_unmanaged_service") +async def fixture_device_unmanaged_service( + dbus_session_bus: MessageBus, +) -> DeviceService: + """Mock Device unmanaged service.""" + yield ( + await mock_dbus_services( + {"network_device": "/org/freedesktop/NetworkManager/Devices/35"}, + dbus_session_bus, + ) + )["network_device"] + + +async def test_network_interface_ethernet( + device_eth0_service: DeviceService, dbus_session_bus: MessageBus +): """Test network interface.""" - interface = network_manager.interfaces[TEST_INTERFACE] + interface = NetworkInterface("/org/freedesktop/NetworkManager/Devices/1") + + assert interface.sync_properties is False + assert interface.name is None + assert interface.type is None + + await interface.connect(dbus_session_bus) + assert interface.sync_properties is True assert interface.name == TEST_INTERFACE assert interface.type == DeviceType.ETHERNET + assert interface.managed is True + assert interface.wireless is None assert interface.connection.state == 2 assert interface.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" assert interface.connection.ipv4.address == [IPv4Interface("192.168.2.148/24")] assert interface.connection.ipv6.address == [ IPv6Interface("2a03:169:3df5:0:6be9:2588:b26a:a679/64"), - IPv6Interface("fd14:949b:c9cc:0:522b:8108:8ff8:cca3/64"), IPv6Interface("2a03:169:3df5::2f1/128"), - IPv6Interface("fd14:949b:c9cc::2f1/128"), - IPv6Interface("fe80::ffe3:319e:c630:9f51/64"), ] assert interface.connection.ipv4.gateway == IPv4Address("192.168.2.1") @@ -44,61 +83,100 @@ async def test_network_interface_ethernet(network_manager: NetworkManager): assert interface.settings.ipv6.method == InterfaceMethod.AUTO assert interface.settings.connection.id == "Wired connection 1" - fire_property_change_signal(interface.connection, {"State": 4}) - await asyncio.sleep(0) - assert interface.connection.state == 4 + device_eth0_service.emit_properties_changed({"Managed": False}) + await device_eth0_service.ping() + assert interface.managed is False - fire_property_change_signal(interface.connection, {}, ["State"]) - await asyncio.sleep(0) - assert interface.connection.state == 2 + device_eth0_service.emit_properties_changed({}, ["Managed"]) + await device_eth0_service.ping() + await device_eth0_service.ping() + assert interface.managed is True -@pytest.mark.asyncio -async def test_network_interface_wlan(network_manager: NetworkManager): - """Test network interface.""" - interface = network_manager.interfaces[TEST_INTERFACE_WLAN] +async def test_network_interface_wlan( + device_wlan0_service: DeviceService, dbus_session_bus: MessageBus +): + """Test wlan network interface.""" + interface = NetworkInterface("/org/freedesktop/NetworkManager/Devices/3") + + assert interface.wireless is None + + await interface.connect(dbus_session_bus) + assert interface.sync_properties is True assert interface.name == TEST_INTERFACE_WLAN assert interface.type == DeviceType.WIRELESS + assert interface.wireless is not None + assert interface.wireless.bitrate == 0 -async def test_old_connection_disconnect(network_manager: NetworkManager): +async def test_old_connection_disconnect( + network_manager: NetworkManager, device_eth0_service: DeviceService +): """Test old connection disconnects on connection change.""" interface = network_manager.interfaces[TEST_INTERFACE] connection = interface.connection assert connection.is_connected is True - fire_property_change_signal(interface, {"ActiveConnection": "/"}) - await asyncio.sleep(0) + device_eth0_service.emit_properties_changed({"ActiveConnection": "/"}) + await device_eth0_service.ping() assert interface.connection is None assert connection.is_connected is False -async def test_old_wireless_disconnect(network_manager: NetworkManager): +async def test_old_wireless_disconnect( + network_manager: NetworkManager, device_wlan0_service: DeviceService +): """Test old wireless disconnects on type change.""" interface = network_manager.interfaces[TEST_INTERFACE_WLAN] wireless = interface.wireless assert wireless.is_connected is True - fire_property_change_signal(interface, {"DeviceType": DeviceType.ETHERNET}) - await asyncio.sleep(0) + device_wlan0_service.emit_properties_changed({"DeviceType": DeviceType.ETHERNET}) + await device_wlan0_service.ping() assert interface.wireless is None assert wireless.is_connected is False -async def test_unmanaged_interface(network_manager: NetworkManager): +async def test_unmanaged_interface( + device_unmanaged_service: DeviceService, dbus_session_bus: MessageBus +): """Test unmanaged interfaces don't sync properties.""" - interface = NetworkInterface( - network_manager.dbus, "/org/freedesktop/NetworkManager/Devices/35" - ) - await interface.connect(network_manager.dbus.bus) + interface = NetworkInterface("/org/freedesktop/NetworkManager/Devices/35") + await interface.connect(dbus_session_bus) assert interface.managed is False assert interface.connection is None assert interface.driver == "veth" assert interface.sync_properties is False - with pytest.raises(AssertionError): - fire_property_change_signal(interface, {"Driver": "test"}) + device_unmanaged_service.emit_properties_changed({"Driver": "test"}) + await device_unmanaged_service.ping() + assert interface.driver == "veth" + + +async def test_interface_becomes_unmanaged( + network_manager: NetworkManager, + device_eth0_service: DeviceService, + device_wlan0_service: DeviceService, +): + """Test managed objects disconnect when interface becomes unmanaged.""" + eth0 = network_manager.interfaces[TEST_INTERFACE] + connection = eth0.connection + wlan0 = network_manager.interfaces[TEST_INTERFACE_WLAN] + wireless = wlan0.wireless + + assert connection.is_connected is True + assert wireless.is_connected is True + + device_eth0_service.emit_properties_changed({"Managed": False}) + await device_eth0_service.ping() + device_wlan0_service.emit_properties_changed({"Managed": False}) + await device_wlan0_service.ping() + + assert wlan0.wireless is None + assert wireless.is_connected is False + assert eth0.connection is None + assert connection.is_connected is False diff --git a/tests/dbus/network/test_ip_configuration.py b/tests/dbus/network/test_ip_configuration.py index 6daa0b488..e7d7747d8 100644 --- a/tests/dbus/network/test_ip_configuration.py +++ b/tests/dbus/network/test_ip_configuration.py @@ -1,50 +1,91 @@ """Test Network Manager IP configuration object.""" -import asyncio from ipaddress import IPv4Address, IPv6Address -from supervisor.dbus.network import NetworkManager +from dbus_fast.aio.message_bus import MessageBus +import pytest -from tests.common import fire_property_change_signal -from tests.const import TEST_INTERFACE +from supervisor.dbus.network.ip_configuration import IpConfiguration + +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.network_ip4config import IP4Config as IP4ConfigService +from tests.dbus_service_mocks.network_ip6config import IP6Config as IP6ConfigService -async def test_ipv4_configuration(network_manager: NetworkManager): +@pytest.fixture(name="ip4config_service") +async def fixture_ip4config_service(dbus_session_bus: MessageBus) -> IP4ConfigService: + """Mock IP4Config service.""" + yield (await mock_dbus_services({"network_ip4config": None}, dbus_session_bus))[ + "network_ip4config" + ] + + +@pytest.fixture(name="ip6config_service") +async def fixture_ip6config_service(dbus_session_bus: MessageBus) -> IP4ConfigService: + """Mock IP6Config service.""" + yield (await mock_dbus_services({"network_ip6config": None}, dbus_session_bus))[ + "network_ip6config" + ] + + +async def test_ipv4_configuration( + ip4config_service: IP4ConfigService, dbus_session_bus: MessageBus +): """Test ipv4 configuration object.""" - ipv4 = network_manager.interfaces[TEST_INTERFACE].connection.ipv4 - assert ipv4.gateway == IPv4Address("192.168.2.1") - assert ipv4.nameservers == [IPv4Address("192.168.2.2")] + ip4 = IpConfiguration("/org/freedesktop/NetworkManager/IP4Config/1") - fire_property_change_signal(ipv4, {"Gateway": "192.168.100.1"}) - await asyncio.sleep(0) - assert ipv4.gateway == IPv4Address("192.168.100.1") + assert ip4.gateway is None + assert ip4.nameservers is None - fire_property_change_signal(ipv4, {}, ["Gateway"]) - await asyncio.sleep(0) - assert ipv4.gateway == IPv4Address("192.168.2.1") + await ip4.connect(dbus_session_bus) + + assert ip4.gateway == IPv4Address("192.168.2.1") + assert ip4.nameservers == [IPv4Address("192.168.2.2")] + + ip4config_service.emit_properties_changed({"Gateway": "192.168.100.1"}) + await ip4config_service.ping() + assert ip4.gateway == IPv4Address("192.168.100.1") + + ip4config_service.emit_properties_changed({}, ["Gateway"]) + await ip4config_service.ping() + await ip4config_service.ping() + assert ip4.gateway == IPv4Address("192.168.2.1") -async def test_ipv6_configuration(network_manager: NetworkManager): +async def test_ipv6_configuration( + ip6config_service: IP6ConfigService, dbus_session_bus: MessageBus +): """Test ipv4 configuration object.""" - ipv6 = network_manager.interfaces[TEST_INTERFACE].connection.ipv6 - assert ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69") - assert ipv6.nameservers == [ + ip6 = IpConfiguration("/org/freedesktop/NetworkManager/IP6Config/1", ip4=False) + + assert ip6.gateway is None + assert ip6.nameservers is None + + await ip6.connect(dbus_session_bus) + + assert ip6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69") + assert ip6.nameservers == [ IPv6Address("2001:1620:2777:1::10"), IPv6Address("2001:1620:2777:2::20"), ] - fire_property_change_signal(ipv6, {"Gateway": "2001:1620:2777:1::10"}) - await asyncio.sleep(0) - assert ipv6.gateway == IPv6Address("2001:1620:2777:1::10") + ip6config_service.emit_properties_changed({"Gateway": "2001:1620:2777:1::10"}) + await ip6config_service.ping() + assert ip6.gateway == IPv6Address("2001:1620:2777:1::10") - fire_property_change_signal(ipv6, {}, ["Gateway"]) - await asyncio.sleep(0) - assert ipv6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69") + ip6config_service.emit_properties_changed({}, ["Gateway"]) + await ip6config_service.ping() + await ip6config_service.ping() + assert ip6.gateway == IPv6Address("fe80::da58:d7ff:fe00:9c69") -async def test_gateway_empty_string(network_manager: NetworkManager): +async def test_gateway_empty_string( + ip4config_service: IP4ConfigService, dbus_session_bus: MessageBus +): """Test empty string in gateway returns None.""" - ipv4 = network_manager.interfaces[TEST_INTERFACE].connection.ipv4 - fire_property_change_signal(ipv4, {"Gateway": ""}) - await asyncio.sleep(0) - assert ipv4.gateway is None + ip4 = IpConfiguration("/org/freedesktop/NetworkManager/IP4Config/1", ip4=True) + await ip4.connect(dbus_session_bus) + + ip4config_service.emit_properties_changed({"Gateway": ""}) + await ip4config_service.ping() + assert ip4.gateway is None diff --git a/tests/dbus/network/test_network_manager.py b/tests/dbus/network/test_network_manager.py index ddae41cf8..0197ff75c 100644 --- a/tests/dbus/network/test_network_manager.py +++ b/tests/dbus/network/test_network_manager.py @@ -1,7 +1,7 @@ """Test NetworkInterface.""" -import asyncio + import logging -from unittest.mock import AsyncMock, Mock, PropertyMock, patch +from unittest.mock import Mock, PropertyMock, patch from dbus_fast.aio.message_bus import MessageBus import pytest @@ -12,59 +12,75 @@ from supervisor.dbus.network.interface import NetworkInterface from supervisor.exceptions import DBusFatalError, DBusParseError, HostNotSupportedError from supervisor.utils.dbus import DBus -from .setting.test_init import SETTINGS_WITH_SIGNATURE - -from tests.common import fire_property_change_signal from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN - -# pylint: disable=protected-access +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.network_connection_settings import SETTINGS_FIXTURE +from tests.dbus_service_mocks.network_manager import ( + NetworkManager as NetworkManagerService, +) -@pytest.mark.asyncio -async def test_network_manager(network_manager: NetworkManager): +@pytest.fixture(name="network_manager_service", autouse=True) +async def fixture_network_manager_service( + network_manager_services: dict[str, DBusServiceMock], +) -> NetworkManagerService: + """Mock NetworkManager dbus service.""" + yield network_manager_services["network_manager"] + + +async def test_network_manager( + network_manager_service: NetworkManagerService, dbus_session_bus: MessageBus +): """Test network manager update.""" + network_manager = NetworkManager() + + assert network_manager.connectivity_enabled is None + + await network_manager.connect(dbus_session_bus) + assert TEST_INTERFACE in network_manager.interfaces assert network_manager.connectivity_enabled is True - fire_property_change_signal(network_manager, {"ConnectivityCheckEnabled": False}) - await asyncio.sleep(0) + network_manager_service.emit_properties_changed({"ConnectivityCheckEnabled": False}) + await network_manager_service.ping() assert network_manager.connectivity_enabled is False - fire_property_change_signal(network_manager, {"ConnectivityCheckEnabled": True}) - await asyncio.sleep(0) + network_manager_service.emit_properties_changed({}, ["ConnectivityCheckEnabled"]) + await network_manager_service.ping() + await network_manager_service.ping() assert network_manager.connectivity_enabled is True -@pytest.mark.asyncio -async def test_network_manager_version(network_manager: NetworkManager): +async def test_network_manager_version( + network_manager_service: NetworkManagerService, network_manager: NetworkManager +): """Test if version validate work.""" await network_manager._validate_version() assert network_manager.version == "1.22.10" - network_manager.dbus.get_properties = AsyncMock(return_value={"Version": "1.13.9"}) + network_manager_service.version = "1.13.9" with pytest.raises(HostNotSupportedError): await network_manager._validate_version() assert network_manager.version == "1.13.9" -async def test_check_connectivity(network_manager: NetworkManager, dbus: list[str]): +async def test_check_connectivity( + network_manager_service: NetworkManagerService, network_manager: NetworkManager +): """Test connectivity check.""" - dbus.clear() + network_manager_service.CheckConnectivity.calls.clear() + assert await network_manager.check_connectivity() == 4 - assert dbus == [ - "/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.Connectivity" - ] + assert network_manager_service.CheckConnectivity.calls == [] - dbus.clear() assert await network_manager.check_connectivity(force=True) == 4 - assert dbus == [ - "/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.CheckConnectivity" - ] + assert network_manager_service.CheckConnectivity.calls == [tuple()] -async def test_activate_connection(network_manager: NetworkManager, dbus: list[str]): +async def test_activate_connection( + network_manager_service: NetworkManagerService, network_manager: NetworkManager +): """Test activate connection.""" - dbus.clear() connection = await network_manager.activate_connection( "/org/freedesktop/NetworkManager/Settings/1", "/org/freedesktop/NetworkManager/Devices/1", @@ -73,19 +89,23 @@ async def test_activate_connection(network_manager: NetworkManager, dbus: list[s assert ( connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1" ) - assert dbus == [ - "/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.ActivateConnection", - "/org/freedesktop/NetworkManager/Settings/1-org.freedesktop.NetworkManager.Settings.Connection.GetSettings", + assert network_manager_service.ActivateConnection.calls == [ + ( + "/org/freedesktop/NetworkManager/Settings/1", + "/org/freedesktop/NetworkManager/Devices/1", + "/", + ) ] async def test_add_and_activate_connection( - network_manager: NetworkManager, dbus: list[str] + network_manager_service: NetworkManagerService, network_manager: NetworkManager ): """Test add and activate connection.""" - dbus.clear() + network_manager_service.AddAndActivateConnection.calls.clear() + settings, connection = await network_manager.add_and_activate_connection( - SETTINGS_WITH_SIGNATURE, "/org/freedesktop/NetworkManager/Devices/1" + SETTINGS_FIXTURE, "/org/freedesktop/NetworkManager/Devices/1" ) assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" assert settings.ipv4.method == "auto" @@ -93,21 +113,20 @@ async def test_add_and_activate_connection( assert ( connection.settings.object_path == "/org/freedesktop/NetworkManager/Settings/1" ) - assert dbus == [ - "/org/freedesktop/NetworkManager-org.freedesktop.NetworkManager.AddAndActivateConnection", - "/org/freedesktop/NetworkManager/Settings/1-org.freedesktop.NetworkManager.Settings.Connection.GetSettings", + assert network_manager_service.AddAndActivateConnection.calls == [ + (SETTINGS_FIXTURE, "/org/freedesktop/NetworkManager/Devices/1", "/") ] -async def test_removed_devices_disconnect(network_manager: NetworkManager): +async def test_removed_devices_disconnect( + network_manager_service: NetworkManagerService, network_manager: NetworkManager +): """Test removed devices are disconnected.""" wlan = network_manager.interfaces[TEST_INTERFACE_WLAN] assert wlan.is_connected is True - fire_property_change_signal( - network_manager, {"Devices": ["/org/freedesktop/NetworkManager/Devices/1"]} - ) - await asyncio.sleep(0) + network_manager_service.emit_properties_changed({"Devices": []}) + await network_manager_service.ping() assert TEST_INTERFACE_WLAN not in network_manager.interfaces assert wlan.is_connected is False @@ -162,15 +181,15 @@ async def test_handling_bad_devices( async def test_ignore_veth_only_changes( - network_manager: NetworkManager, dbus_bus: MessageBus + network_manager_service: NetworkManagerService, network_manager: NetworkManager ): """Changes to list of devices is ignored unless it changes managed devices.""" assert network_manager.properties["Devices"] == [ "/org/freedesktop/NetworkManager/Devices/1", "/org/freedesktop/NetworkManager/Devices/3", ] - with patch.object(NetworkInterface, "update") as update: - await network_manager.update( + with patch.object(NetworkInterface, "connect") as connect: + network_manager_service.emit_properties_changed( { "Devices": [ "/org/freedesktop/NetworkManager/Devices/1", @@ -179,9 +198,11 @@ async def test_ignore_veth_only_changes( ] } ) - update.assert_not_called() + await network_manager_service.ping() + connect.assert_not_called() - await network_manager.update( + network_manager_service.emit_properties_changed( {"Devices": ["/org/freedesktop/NetworkManager/Devices/35"]} ) - update.assert_called_once() + await network_manager_service.ping() + connect.assert_called_once() diff --git a/tests/dbus/network/test_setting.py b/tests/dbus/network/test_setting.py deleted file mode 100644 index e979ece09..000000000 --- a/tests/dbus/network/test_setting.py +++ /dev/null @@ -1,59 +0,0 @@ -"""Test NetwrokInterface.""" -import asyncio - -import pytest - -from supervisor.dbus.const import DeviceType -from supervisor.dbus.network import NetworkManager -from supervisor.dbus.network.setting.generate import get_connection_from_interface -from supervisor.host.network import Interface - -from tests.common import fire_property_change_signal -from tests.const import TEST_INTERFACE, TEST_INTERFACE_WLAN - - -@pytest.mark.asyncio -async def test_get_connection_from_interface(network_manager: NetworkManager): - """Test network interface.""" - dbus_interface = network_manager.interfaces[TEST_INTERFACE] - interface = Interface.from_dbus_interface(dbus_interface) - connection_payload = get_connection_from_interface(interface) - - assert "connection" in connection_payload - - assert connection_payload["connection"]["interface-name"].value == TEST_INTERFACE - assert connection_payload["connection"]["type"].value == "802-3-ethernet" - - assert connection_payload["ipv4"]["method"].value == "auto" - assert "address-data" not in connection_payload["ipv4"] - - assert connection_payload["ipv6"]["method"].value == "auto" - assert "address-data" not in connection_payload["ipv6"] - - -async def test_network_interface(network_manager: NetworkManager): - """Test network interface.""" - interface = network_manager.interfaces[TEST_INTERFACE] - assert interface.name == TEST_INTERFACE - assert interface.type == DeviceType.ETHERNET - assert interface.managed is True - - fire_property_change_signal( - network_manager.interfaces[TEST_INTERFACE], {"Managed": False} - ) - await asyncio.sleep(0) - assert network_manager.interfaces[TEST_INTERFACE].managed is False - - fire_property_change_signal( - network_manager.interfaces[TEST_INTERFACE], {}, ["Managed"] - ) - await asyncio.sleep(0) - assert network_manager.interfaces[TEST_INTERFACE].managed is True - - -@pytest.mark.asyncio -async def test_network_interface_wlan(network_manager: NetworkManager): - """Test wireless network interface.""" - interface = network_manager.interfaces[TEST_INTERFACE_WLAN] - assert interface.name == TEST_INTERFACE_WLAN - assert interface.type == DeviceType.WIRELESS diff --git a/tests/dbus/network/test_settings.py b/tests/dbus/network/test_settings.py index 3d6c7f487..2cec20837 100644 --- a/tests/dbus/network/test_settings.py +++ b/tests/dbus/network/test_settings.py @@ -1,25 +1,55 @@ """Test Network Manager Connection Settings Profile Manager.""" -from supervisor.dbus.network import NetworkManager -from tests.dbus.network.setting.test_init import SETTINGS_WITH_SIGNATURE +from dbus_fast.aio.message_bus import MessageBus +import pytest + +from supervisor.dbus.network.settings import NetworkManagerSettings +from supervisor.exceptions import DBusNotConnectedError + +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.network_connection_settings import SETTINGS_FIXTURE +from tests.dbus_service_mocks.network_settings import Settings as SettingsService -async def test_add_connection(network_manager: NetworkManager, dbus: list[str]): +@pytest.fixture(name="settings_service", autouse=True) +async def fixture_settings_service(dbus_session_bus: MessageBus) -> SettingsService: + """Mock Settings service.""" + yield ( + await mock_dbus_services( + {"network_settings": None, "network_connection_settings": None}, + dbus_session_bus, + ) + )["network_settings"] + + +async def test_add_connection( + settings_service: SettingsService, dbus_session_bus: MessageBus +): """Test adding settings connection.""" - dbus.clear() - settings = await network_manager.settings.add_connection(SETTINGS_WITH_SIGNATURE) - assert settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" - assert settings.ipv4.method == "auto" - assert dbus == [ - "/org/freedesktop/NetworkManager/Settings-org.freedesktop.NetworkManager.Settings.AddConnection", - "/org/freedesktop/NetworkManager/Settings/1-org.freedesktop.NetworkManager.Settings.Connection.GetSettings", - ] + settings = NetworkManagerSettings() + + with pytest.raises(DBusNotConnectedError): + await settings.add_connection(SETTINGS_FIXTURE) + + await settings.connect(dbus_session_bus) + + connection_settings = await settings.add_connection(SETTINGS_FIXTURE) + assert connection_settings.connection.uuid == "0c23631e-2118-355c-bbb0-8943229cb0d6" + assert connection_settings.ipv4.method == "auto" + + assert settings_service.AddConnection.calls == [(SETTINGS_FIXTURE,)] -async def test_reload_connections(network_manager: NetworkManager, dbus: list[str]): +async def test_reload_connections( + settings_service: SettingsService, dbus_session_bus: MessageBus +): """Test reload connections.""" - dbus.clear() - assert await network_manager.settings.reload_connections() is True - assert dbus == [ - "/org/freedesktop/NetworkManager/Settings-org.freedesktop.NetworkManager.Settings.ReloadConnections" - ] + settings = NetworkManagerSettings() + + with pytest.raises(DBusNotConnectedError): + await settings.reload_connections() + + await settings.connect(dbus_session_bus) + + assert await settings.reload_connections() is True + assert settings_service.ReloadConnections.calls == [tuple()] diff --git a/tests/dbus/network/test_wireless.py b/tests/dbus/network/test_wireless.py index 4a995e53d..07c47ddce 100644 --- a/tests/dbus/network/test_wireless.py +++ b/tests/dbus/network/test_wireless.py @@ -1,50 +1,69 @@ """Test Network Manager Wireless object.""" -import asyncio + +from dbus_fast.aio.message_bus import MessageBus +import pytest from supervisor.dbus.network import NetworkManager +from supervisor.dbus.network.wireless import NetworkWireless -from tests.common import fire_property_change_signal from tests.const import TEST_INTERFACE_WLAN +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.network_device_wireless import ( + DeviceWireless as DeviceWirelessService, +) -async def test_wireless(network_manager: NetworkManager): +@pytest.fixture(name="device_wireless_service", autouse=True) +async def fixture_device_wireless_service( + network_manager_services: dict[str, DBusServiceMock] +) -> DeviceWirelessService: + """Mock Device Wireless service.""" + yield network_manager_services["network_device_wireless"] + + +async def test_wireless( + device_wireless_service: DeviceWirelessService, dbus_session_bus: MessageBus +): """Test wireless properties.""" - assert network_manager.interfaces[TEST_INTERFACE_WLAN].wireless.active is None + wireless = NetworkWireless("/org/freedesktop/NetworkManager/Devices/3") - fire_property_change_signal( - network_manager.interfaces[TEST_INTERFACE_WLAN].wireless, - {"ActiveAccessPoint": "/org/freedesktop/NetworkManager/AccessPoint/43099"}, + assert wireless.bitrate is None + + await wireless.connect(dbus_session_bus) + + assert wireless.bitrate == 0 + assert wireless.active is None + + device_wireless_service.emit_properties_changed( + {"ActiveAccessPoint": "/org/freedesktop/NetworkManager/AccessPoint/43099"} ) - await asyncio.sleep(0) + await device_wireless_service.ping() + assert wireless.active is not None assert ( - network_manager.interfaces[TEST_INTERFACE_WLAN].wireless.active.mac - == "E4:57:40:A9:D7:DE" + wireless.active.object_path + == "/org/freedesktop/NetworkManager/AccessPoint/43099" ) - fire_property_change_signal( - network_manager.interfaces[TEST_INTERFACE_WLAN].wireless, - {}, - ["ActiveAccessPoint"], - ) - await asyncio.sleep(0) - assert network_manager.interfaces[TEST_INTERFACE_WLAN].wireless.active is None + device_wireless_service.emit_properties_changed({}, ["ActiveAccessPoint"]) + await device_wireless_service.ping() + await device_wireless_service.ping() + assert wireless.active is None -async def test_request_scan(network_manager: NetworkManager, dbus: list[str]): +async def test_request_scan( + network_manager: NetworkManager, device_wireless_service: DeviceWirelessService +): """Test request scan.""" - dbus.clear() + device_wireless_service.RequestScan.calls.clear() assert ( await network_manager.interfaces[TEST_INTERFACE_WLAN].wireless.request_scan() is None ) - assert dbus == [ - "/org/freedesktop/NetworkManager/Devices/3-org.freedesktop.NetworkManager.Device.Wireless.RequestScan" - ] + assert device_wireless_service.RequestScan.calls == [({},)] -async def test_get_all_access_points(network_manager: NetworkManager, dbus: list[str]): +async def test_get_all_access_points(network_manager: NetworkManager): """Test get all access points.""" - dbus.clear() accesspoints = await network_manager.interfaces[ TEST_INTERFACE_WLAN ].wireless.get_all_accesspoints() @@ -53,25 +72,18 @@ async def test_get_all_access_points(network_manager: NetworkManager, dbus: list assert accesspoints[0].mode == 2 assert accesspoints[1].mac == "18:4B:0D:23:A1:9C" assert accesspoints[1].mode == 2 - assert dbus == [ - "/org/freedesktop/NetworkManager/Devices/3-org.freedesktop.NetworkManager.Device.Wireless.GetAllAccessPoints" - ] async def test_old_active_ap_disconnects(network_manager: NetworkManager): """Test old access point disconnects on active ap change.""" wireless = network_manager.interfaces[TEST_INTERFACE_WLAN].wireless - fire_property_change_signal( - wireless, - {"ActiveAccessPoint": "/org/freedesktop/NetworkManager/AccessPoint/43099"}, - ) - await asyncio.sleep(0) + await wireless.update( + {"ActiveAccessPoint": "/org/freedesktop/NetworkManager/AccessPoint/43099"} + ) active = wireless.active assert active.is_connected is True - fire_property_change_signal(wireless, {"ActiveAccessPoint": "/"}) - await asyncio.sleep(0) - + await wireless.update({"ActiveAccessPoint": "/"}) assert wireless.active is None assert active.is_connected is False diff --git a/tests/dbus_service_mocks/hostname.py b/tests/dbus_service_mocks/hostname.py index 292dcdda8..e86882e63 100644 --- a/tests/dbus_service_mocks/hostname.py +++ b/tests/dbus_service_mocks/hostname.py @@ -129,7 +129,7 @@ class Hostname(DBusServiceMock): @dbus_method() def GetProductUUID(self, interactive: "b") -> "ay": """Get product UUID.""" - return bytearray("d153e353-2a32-4763-b930-b27fbc980da5", encoding="utf-8") + return b"d153e353-2a32-4763-b930-b27fbc980da5" @dbus_method() def Describe(self) -> "s": diff --git a/tests/dbus_service_mocks/network_access_point.py b/tests/dbus_service_mocks/network_access_point.py new file mode 100644 index 000000000..954dd66c1 --- /dev/null +++ b/tests/dbus_service_mocks/network_access_point.py @@ -0,0 +1,128 @@ +"""Mock of Network Manager Access Point service.""" + +from ctypes import c_byte, c_int32, c_uint32 +from dataclasses import dataclass + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock + +BUS_NAME = "org.freedesktop.NetworkManager" +DEFAULT_OBJECT_PATH = "/org/freedesktop/NetworkManager/AccessPoint/43099" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return AccessPoint(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class AccessPointFixture: + """Access Point fixture.""" + + Flags: c_uint32 + WpaFlags: c_uint32 + RsnFlags: c_uint32 + Ssid: bytes + Frequency: c_uint32 + HwAddress: str + Mode: c_uint32 + MaxBitrate: c_uint32 + Strength: c_byte + LastSeen: c_int32 + + +FIXTURES: dict[str, AccessPointFixture] = { + "/org/freedesktop/NetworkManager/AccessPoint/43099": AccessPointFixture( + Flags=3, + WpaFlags=0, + RsnFlags=392, + Ssid=b"UPC4814466", + Frequency=2462, + HwAddress="E4:57:40:A9:D7:DE", + Mode=2, + MaxBitrate=195000, + Strength=47, + LastSeen=1398776, + ), + "/org/freedesktop/NetworkManager/AccessPoint/43100": AccessPointFixture( + Flags=1, + WpaFlags=0, + RsnFlags=392, + Ssid=b"VQ@35(55720", + Frequency=5660, + HwAddress="18:4B:0D:23:A1:9C", + Mode=2, + MaxBitrate=540000, + Strength=63, + LastSeen=1398839, + ), +} + + +class AccessPoint(DBusServiceMock): + """Access Point mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/AccessPoint/1 + """ + + interface = "org.freedesktop.NetworkManager.AccessPoint" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: AccessPointFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def Flags(self) -> "u": + """Get Flags.""" + return self.fixture.Flags + + @dbus_property(access=PropertyAccess.READ) + def WpaFlags(self) -> "u": + """Get WpaFlags.""" + return self.fixture.WpaFlags + + @dbus_property(access=PropertyAccess.READ) + def RsnFlags(self) -> "u": + """Get RsnFlags.""" + return self.fixture.RsnFlags + + @dbus_property(access=PropertyAccess.READ) + def Ssid(self) -> "ay": + """Get Ssid.""" + return self.fixture.Ssid + + @dbus_property(access=PropertyAccess.READ) + def Frequency(self) -> "u": + """Get Frequency.""" + return self.fixture.Frequency + + @dbus_property(access=PropertyAccess.READ) + def HwAddress(self) -> "s": + """Get HwAddress.""" + return self.fixture.HwAddress + + @dbus_property(access=PropertyAccess.READ) + def Mode(self) -> "u": + """Get Mode.""" + return self.fixture.Mode + + @dbus_property(access=PropertyAccess.READ) + def MaxBitrate(self) -> "u": + """Get MaxBitrate.""" + return self.fixture.MaxBitrate + + @dbus_property(access=PropertyAccess.READ) + def Strength(self) -> "y": + """Get Strength.""" + return self.fixture.Strength + + @dbus_property(access=PropertyAccess.READ) + def LastSeen(self) -> "i": + """Get LastSeen.""" + return self.fixture.LastSeen diff --git a/tests/dbus_service_mocks/network_active_connection.py b/tests/dbus_service_mocks/network_active_connection.py new file mode 100644 index 000000000..67fee4a4c --- /dev/null +++ b/tests/dbus_service_mocks/network_active_connection.py @@ -0,0 +1,110 @@ +"""Mock of Network Manager Active Connection service.""" + +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return ActiveConnection() + + +# pylint: disable=invalid-name + + +class ActiveConnection(DBusServiceMock): + """Active Connection mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/ActiveConnection/1 + """ + + interface = "org.freedesktop.NetworkManager.Connection.Active" + object_path = "/org/freedesktop/NetworkManager/ActiveConnection/1" + + @dbus_property(access=PropertyAccess.READ) + def Connection(self) -> "o": + """Get Connection.""" + return "/org/freedesktop/NetworkManager/Settings/1" + + @dbus_property(access=PropertyAccess.READ) + def SpecificObject(self) -> "o": + """Get SpecificObject.""" + return "/" + + @dbus_property(access=PropertyAccess.READ) + def Id(self) -> "s": + """Get Id.""" + return "Wired connection 1" + + @dbus_property(access=PropertyAccess.READ) + def Uuid(self) -> "s": + """Get Uuid.""" + return "0c23631e-2118-355c-bbb0-8943229cb0d6" + + @dbus_property(access=PropertyAccess.READ) + def Type(self) -> "s": + """Get Type.""" + return "802-3-ethernet" + + @dbus_property(access=PropertyAccess.READ) + def Devices(self) -> "ao": + """Get Devices.""" + return ["/org/freedesktop/NetworkManager/Devices/1"] + + @dbus_property(access=PropertyAccess.READ) + def State(self) -> "u": + """Get State.""" + return 2 + + @dbus_property(access=PropertyAccess.READ) + def StateFlags(self) -> "u": + """Get StateFlags.""" + return 92 + + @dbus_property(access=PropertyAccess.READ) + def Default(self) -> "b": + """Get Default.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def Ip4Config(self) -> "o": + """Get Ip4Config.""" + return "/org/freedesktop/NetworkManager/IP4Config/1" + + @dbus_property(access=PropertyAccess.READ) + def Dhcp4Config(self) -> "o": + """Get Dhcp4Config.""" + return "/org/freedesktop/NetworkManager/DHCP4Config/1" + + @dbus_property(access=PropertyAccess.READ) + def Default6(self) -> "b": + """Get Default6.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def Ip6Config(self) -> "o": + """Get Ip6Config.""" + return "/org/freedesktop/NetworkManager/IP6Config/1" + + @dbus_property(access=PropertyAccess.READ) + def Dhcp6Config(self) -> "o": + """Get Dhcp6Config.""" + return "/" + + @dbus_property(access=PropertyAccess.READ) + def Vpn(self) -> "b": + """Get Vpn.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def Master(self) -> "o": + """Get Master.""" + return "/" + + @signal() + def StateChanged(self) -> "uu": + """Signal StateChanged.""" + return [0, 0] diff --git a/tests/dbus_service_mocks/network_connection_settings.py b/tests/dbus_service_mocks/network_connection_settings.py new file mode 100644 index 000000000..c30f6d985 --- /dev/null +++ b/tests/dbus_service_mocks/network_connection_settings.py @@ -0,0 +1,144 @@ +"""Mock of Network Manager Connection Settings service.""" + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.NetworkManager" +SETTINGS_FIXTURE = { + "connection": { + "id": Variant("s", "Wired connection 1"), + "interface-name": Variant("s", "eth0"), + "llmnr": Variant("i", 2), + "mdns": Variant("i", 2), + "permissions": Variant("as", []), + "timestamp": Variant("t", 1598125548), + "type": Variant("s", "802-3-ethernet"), + "uuid": Variant("s", "0c23631e-2118-355c-bbb0-8943229cb0d6"), + }, + "ipv4": { + "address-data": Variant( + "aa{sv}", + [ + { + "address": Variant("s", "192.168.2.148"), + "prefix": Variant("u", 24), + } + ], + ), + "addresses": Variant("aau", [[2483202240, 24, 16951488]]), + "dns": Variant("au", [16951488]), + "dns-search": Variant("as", []), + "gateway": Variant("s", "192.168.2.1"), + "method": Variant("s", "auto"), + "route-data": Variant( + "aa{sv}", + [ + { + "dest": Variant("s", "192.168.122.0"), + "prefix": Variant("u", 24), + "next-hop": Variant("s", "10.10.10.1"), + } + ], + ), + "routes": Variant("aau", [[8038592, 24, 17435146, 0]]), + }, + "ipv6": { + "address-data": Variant("aa{sv}", []), + "addresses": Variant("a(ayuay)", []), + "dns-search": Variant("as", []), + "method": Variant("s", "auto"), + "route-data": Variant("aa{sv}", []), + "routes": Variant("a(ayuayu)", []), + "addr-gen-mode": Variant("i", 0), + }, + "proxy": {}, + "802-3-ethernet": { + "assigned-mac-address": Variant("s", "preserve"), + "auto-negotiate": Variant("b", False), + "mac-address-blacklist": Variant("as", []), + "s390-options": Variant("a{ss}", {}), + }, + "802-11-wireless": {"ssid": Variant("ay", b"NETT")}, +} + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return ConnectionSettings() + + +# pylint: disable=invalid-name + + +class ConnectionSettings(DBusServiceMock): + """Connection Settings mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/Settings/1 + """ + + interface = "org.freedesktop.NetworkManager.Settings.Connection" + object_path = "/org/freedesktop/NetworkManager/Settings/1" + + @dbus_property(access=PropertyAccess.READ) + def Unsaved(self) -> "b": + """Get Unsaved.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def Flags(self) -> "u": + """Get Flags.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def Filename(self) -> "s": + """Get Unsaved.""" + return "/etc/NetworkManager/system-connections/Supervisor eth0.nmconnection" + + @signal() + def Updated(self) -> None: + """Signal Updated.""" + + @signal() + def Removed(self) -> None: + """Signal Removed.""" + + @dbus_method() + def Update(self, properties: "a{sa{sv}}") -> None: + """Do Update method.""" + self.Updated() + + @dbus_method() + def UpdateUnsaved(self, properties: "a{sa{sv}}") -> None: + """Do UpdateUnsaved method.""" + + @dbus_method() + def Delete(self) -> None: + """Do Delete method.""" + self.Removed() + + @dbus_method() + def GetSettings(self) -> "a{sa{sv}}": + """Do GetSettings method.""" + return SETTINGS_FIXTURE + + @dbus_method() + def GetSecrets(self, setting_name: "s") -> "a{sa{sv}}": + """Do GetSecrets method.""" + return self.GetSettings() + + @dbus_method() + def ClearSecrets(self) -> None: + """Do ClearSecrets method.""" + + @dbus_method() + def Save(self) -> None: + """Do Save method.""" + self.Updated() + + @dbus_method() + def Update2(self, settings: "a{sa{sv}}", flags: "u", args: "a{sv}") -> "a{sv}": + """Do Update2 method.""" + self.Updated() + return {} diff --git a/tests/dbus_service_mocks/network_device.py b/tests/dbus_service_mocks/network_device.py new file mode 100644 index 000000000..01bcc024d --- /dev/null +++ b/tests/dbus_service_mocks/network_device.py @@ -0,0 +1,371 @@ +"""Mock of Network Manager Device service.""" + +from ctypes import c_uint32 +from dataclasses import dataclass + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.NetworkManager" +DEFAULT_OBJECT_PATH = "/org/freedesktop/NetworkManager/Devices/1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Device(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class DeviceFixture: + """Device fixture.""" + + Udi: str + Path: str + Interface: str + IpInterface: str + Driver: str + DriverVersion: str + FirmwareVersion: str + Capabilities: c_uint32 + Ip4Address: c_uint32 + State: c_uint32 + StateReason: list[c_uint32] + ActiveConnection: str + Ip4Config: str + Dhcp4Config: str + Ip6Config: str + Dhcp6Config: str + Managed: bool + Autoconnect: bool + FirmwareMissing: bool + NmPluginMissing: bool + DeviceType: c_uint32 + AvailableConnections: list[str] + PhysicalPortId: str + Mtu: c_uint32 + Metered: c_uint32 + LldpNeighbors: list[dict[str, Variant]] + Real: bool + Ip4Connectivity: c_uint32 + Ip6Connectivity: c_uint32 + InterfaceFlags: c_uint32 + HwAddress: str + Ports: list[str] + + +FIXTURES: dict[str, DeviceFixture] = { + "/org/freedesktop/NetworkManager/Devices/1": DeviceFixture( + Udi="/sys/devices/pci0000:00/0000:00:1f.6/net/eth0", + Path="platform-ff3f0000.ethernet", + Interface="eth0", + IpInterface="eth0", + Driver="e1000e", + DriverVersion="3.2.6-k", + FirmwareVersion="0.7-4", + Capabilities=3, + Ip4Address=2499979456, + State=100, + StateReason=[100, 0], + ActiveConnection="/org/freedesktop/NetworkManager/ActiveConnection/1", + Ip4Config="/org/freedesktop/NetworkManager/IP4Config/1", + Dhcp4Config="/org/freedesktop/NetworkManager/DHCP4Config/1", + Ip6Config="/org/freedesktop/NetworkManager/IP6Config/1", + Dhcp6Config="/", + Managed=True, + Autoconnect=True, + FirmwareMissing=False, + NmPluginMissing=False, + DeviceType=1, + AvailableConnections=["/org/freedesktop/NetworkManager/Settings/1"], + PhysicalPortId="", + Mtu=1500, + Metered=4, + LldpNeighbors=[], + Real=True, + Ip4Connectivity=4, + Ip6Connectivity=3, + InterfaceFlags=65539, + HwAddress="AA:BB:CC:DD:EE:FF", + Ports=[], + ), + "/org/freedesktop/NetworkManager/Devices/3": DeviceFixture( + Udi="/sys/devices/platform/soc/fe300000.mmcnr/mmc_host/mmc1/mmc1:0001/mmc1:0001:1/net/wlan0", + Path="platform.fe300000.mmcnr", + Interface="wlan0", + IpInterface="", + Driver="brcmfmac", + DriverVersion="7.45.154", + FirmwareVersion="01-4fbe0b04", + Capabilities=1, + Ip4Address=0, + State=30, + StateReason=[30, 42], + ActiveConnection="/", + Ip4Config="/", + Dhcp4Config="/", + Ip6Config="/", + Dhcp6Config="/", + Managed=True, + Autoconnect=True, + FirmwareMissing=False, + NmPluginMissing=False, + DeviceType=2, + AvailableConnections=[], + PhysicalPortId="", + Mtu=1500, + Metered=0, + LldpNeighbors=[], + Real=True, + Ip4Connectivity=1, + Ip6Connectivity=1, + InterfaceFlags=65539, + HwAddress="FF:EE:DD:CC:BB:AA", + Ports=[], + ), + "/org/freedesktop/NetworkManager/Devices/35": DeviceFixture( + Udi="/sys/devices/virtual/net/veth87bd238'", + Path="", + Interface="veth87bd238", + IpInterface="veth87bd238", + Driver="veth", + DriverVersion="1.0", + FirmwareVersion="", + Capabilities=7, + Ip4Address=0, + State=10, + StateReason=[10, 0], + ActiveConnection="/", + Ip4Config="/", + Dhcp4Config="/", + Ip6Config="/", + Dhcp6Config="/", + Managed=False, + Autoconnect=True, + FirmwareMissing=False, + NmPluginMissing=False, + DeviceType=20, + AvailableConnections=[], + PhysicalPortId="", + Mtu=1500, + Metered=0, + LldpNeighbors=[], + Real=True, + Ip4Connectivity=0, + Ip6Connectivity=0, + InterfaceFlags=65539, + HwAddress="9A:4B:E3:9A:F8:D3", + Ports=[], + ), +} + + +class Device(DBusServiceMock): + """Device mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/Devices/1 + """ + + interface = "org.freedesktop.NetworkManager.Device" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: DeviceFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def Udi(self) -> "s": + """Get Udi.""" + return self.fixture.Udi + + @dbus_property(access=PropertyAccess.READ) + def Path(self) -> "s": + """Get Path.""" + return self.fixture.Path + + @dbus_property(access=PropertyAccess.READ) + def Interface(self) -> "s": + """Get Interface.""" + return self.fixture.Interface + + @dbus_property(access=PropertyAccess.READ) + def IpInterface(self) -> "s": + """Get IpInterface.""" + return self.fixture.IpInterface + + @dbus_property(access=PropertyAccess.READ) + def Driver(self) -> "s": + """Get Driver.""" + return self.fixture.Driver + + @dbus_property(access=PropertyAccess.READ) + def DriverVersion(self) -> "s": + """Get DriverVersion.""" + return self.fixture.DriverVersion + + @dbus_property(access=PropertyAccess.READ) + def FirmwareVersion(self) -> "s": + """Get FirmwareVersion.""" + return self.fixture.FirmwareVersion + + @dbus_property(access=PropertyAccess.READ) + def Capabilities(self) -> "u": + """Get Capabilities.""" + return self.fixture.Capabilities + + @dbus_property(access=PropertyAccess.READ) + def Ip4Address(self) -> "u": + """Get Ip4Address.""" + return self.fixture.Ip4Address + + @dbus_property(access=PropertyAccess.READ) + def State(self) -> "u": + """Get State.""" + return self.fixture.State + + @dbus_property(access=PropertyAccess.READ) + def StateReason(self) -> "(uu)": + """Get StateReason.""" + return self.fixture.StateReason + + @dbus_property(access=PropertyAccess.READ) + def ActiveConnection(self) -> "o": + """Get ActiveConnection.""" + return self.fixture.ActiveConnection + + @dbus_property(access=PropertyAccess.READ) + def Ip4Config(self) -> "o": + """Get Ip4Config.""" + return self.fixture.Ip4Config + + @dbus_property(access=PropertyAccess.READ) + def Dhcp4Config(self) -> "o": + """Get Dhcp4Config.""" + return self.fixture.Dhcp4Config + + @dbus_property(access=PropertyAccess.READ) + def Ip6Config(self) -> "o": + """Get Ip6Config.""" + return self.fixture.Ip6Config + + @dbus_property(access=PropertyAccess.READ) + def Dhcp6Config(self) -> "o": + """Get Dhcp6Config.""" + return self.fixture.Dhcp6Config + + @dbus_property() + def Managed(self) -> "b": + """Get Managed.""" + return self.fixture.Managed + + @Managed.setter + def Managed(self, value: "b"): + """Set Managed.""" + self.emit_properties_changed({"Managed": value}) + + @dbus_property() + def Autoconnect(self) -> "b": + """Get Autoconnect.""" + return self.fixture.Autoconnect + + @Autoconnect.setter + def Autoconnect(self, value: "b"): + """Set Autoconnect.""" + self.emit_properties_changed({"Autoconnect": value}) + + @dbus_property(access=PropertyAccess.READ) + def FirmwareMissing(self) -> "b": + """Get FirmwareMissing.""" + return self.fixture.FirmwareMissing + + @dbus_property(access=PropertyAccess.READ) + def NmPluginMissing(self) -> "b": + """Get NmPluginMissing.""" + return self.fixture.NmPluginMissing + + @dbus_property(access=PropertyAccess.READ) + def DeviceType(self) -> "u": + """Get DeviceType.""" + return self.fixture.DeviceType + + @dbus_property(access=PropertyAccess.READ) + def AvailableConnections(self) -> "ao": + """Get AvailableConnections.""" + return self.fixture.AvailableConnections + + @dbus_property(access=PropertyAccess.READ) + def PhysicalPortId(self) -> "s": + """Get PhysicalPortId.""" + return self.fixture.PhysicalPortId + + @dbus_property(access=PropertyAccess.READ) + def Mtu(self) -> "u": + """Get Mtu.""" + return self.fixture.Mtu + + @dbus_property(access=PropertyAccess.READ) + def Metered(self) -> "u": + """Get Metered.""" + return self.fixture.Metered + + @dbus_property(access=PropertyAccess.READ) + def LldpNeighbors(self) -> "aa{sv}": + """Get LldpNeighbors.""" + return self.fixture.LldpNeighbors + + @dbus_property(access=PropertyAccess.READ) + def Real(self) -> "b": + """Get Real.""" + return self.fixture.Real + + @dbus_property(access=PropertyAccess.READ) + def Ip4Connectivity(self) -> "u": + """Get Ip4Connectivity.""" + return self.fixture.Ip4Connectivity + + @dbus_property(access=PropertyAccess.READ) + def Ip6Connectivity(self) -> "u": + """Get Ip6Connectivity.""" + return self.fixture.Ip6Connectivity + + @dbus_property(access=PropertyAccess.READ) + def InterfaceFlags(self) -> "u": + """Get InterfaceFlags.""" + return self.fixture.InterfaceFlags + + @dbus_property(access=PropertyAccess.READ) + def HwAddress(self) -> "s": + """Get HwAddress.""" + return self.fixture.HwAddress + + @dbus_property(access=PropertyAccess.READ) + def Ports(self) -> "ao": + """Get Ports.""" + return self.fixture.Ports + + @signal() + def StateChanged(self) -> "uuu": + """Signal StateChanged.""" + return [120, 100, 1] + + @dbus_method() + def Reapply(self, connection: "a{sa{sv}}", version_id: "t", flags: "u") -> None: + """Do Reapply method.""" + + @dbus_method() + def GetAppliedConnection(self, flags: "u") -> "a{sa{sv}}t": + """Do GetAppliedConnection method.""" + return [{}, 0] + + @dbus_method() + def Disconnect(self) -> None: + """Do Disconnect method.""" + + @dbus_method() + def Delete(self) -> None: + """Do Delete method.""" diff --git a/tests/dbus_service_mocks/network_device_wireless.py b/tests/dbus_service_mocks/network_device_wireless.py new file mode 100644 index 000000000..e89774fd8 --- /dev/null +++ b/tests/dbus_service_mocks/network_device_wireless.py @@ -0,0 +1,102 @@ +"""Mock of Network Manager Device Wireless service.""" + +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return DeviceWireless() + + +# pylint: disable=invalid-name + + +class DeviceWireless(DBusServiceMock): + """Device Wireless mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/Devices/1 + """ + + interface = "org.freedesktop.NetworkManager.Device.Wireless" + object_path = "/org/freedesktop/NetworkManager/Devices/3" + + @dbus_property(access=PropertyAccess.READ) + def HwAddress(self) -> "s": + """Get HwAddress.""" + return "EA:3C:50:4C:B8:82" + + @dbus_property(access=PropertyAccess.READ) + def PermHwAddress(self) -> "s": + """Get PermHwAddress.""" + return "DC:A6:32:02:BA:21" + + @dbus_property(access=PropertyAccess.READ) + def Mode(self) -> "u": + """Get Mode.""" + return 2 + + @dbus_property(access=PropertyAccess.READ) + def Bitrate(self) -> "u": + """Get Bitrate.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def AccessPoints(self) -> "ao": + """Get AccessPoints.""" + return [ + "/org/freedesktop/NetworkManager/AccessPoint/41533", + "/org/freedesktop/NetworkManager/AccessPoint/41534", + "/org/freedesktop/NetworkManager/AccessPoint/41535", + "/org/freedesktop/NetworkManager/AccessPoint/41536", + "/org/freedesktop/NetworkManager/AccessPoint/41537", + "/org/freedesktop/NetworkManager/AccessPoint/41538", + "/org/freedesktop/NetworkManager/AccessPoint/41539", + "/org/freedesktop/NetworkManager/AccessPoint/41540", + "/org/freedesktop/NetworkManager/AccessPoint/41541", + ] + + @dbus_property(access=PropertyAccess.READ) + def ActiveAccessPoint(self) -> "o": + """Get ActiveAccessPoint.""" + return "/" + + @dbus_property(access=PropertyAccess.READ) + def WirelessCapabilities(self) -> "u": + """Get WirelessCapabilities.""" + return 2047 + + @dbus_property(access=PropertyAccess.READ) + def LastScan(self) -> "x": + """Get LastScan.""" + return 1343924585 + + @signal() + def AccessPointAdded(self) -> "o": + """Signal AccessPointAdded.""" + return "/org/freedesktop/NetworkManager/AccessPoint/43100" + + @signal() + def AccessPointRemoved(self) -> "o": + """Signal AccessPointRemoved.""" + return "/org/freedesktop/NetworkManager/AccessPoint/43100" + + @dbus_method() + def GetAccessPoints(self) -> "ao": + """Do GetAccessPoints method.""" + return self.GetAllAccessPoints() + + @dbus_method() + def GetAllAccessPoints(self) -> "ao": + """Do GetAllAccessPoints method.""" + return [ + "/org/freedesktop/NetworkManager/AccessPoint/43099", + "/org/freedesktop/NetworkManager/AccessPoint/43100", + ] + + @dbus_method() + def RequestScan(self, options: "a{sv}") -> None: + """Do RequestScan method.""" diff --git a/tests/dbus_service_mocks/network_dns_manager.py b/tests/dbus_service_mocks/network_dns_manager.py new file mode 100644 index 000000000..d639d5aeb --- /dev/null +++ b/tests/dbus_service_mocks/network_dns_manager.py @@ -0,0 +1,49 @@ +"""Mock of Network Manager DNS Manager service.""" + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return DnsManager() + + +# pylint: disable=invalid-name + + +class DnsManager(DBusServiceMock): + """DNS Manager mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/DnsManager + """ + + interface = "org.freedesktop.NetworkManager.DnsManager" + object_path = "/org/freedesktop/NetworkManager/DnsManager" + + @dbus_property(access=PropertyAccess.READ) + def Mode(self) -> "s": + """Get Mode.""" + return "default" + + @dbus_property(access=PropertyAccess.READ) + def RcManager(self) -> "s": + """Get RcManager.""" + return "file" + + @dbus_property(access=PropertyAccess.READ) + def Configuration(self) -> "aa{sv}": + """Get Configuration.""" + return [ + { + "nameservers": Variant("as", ["192.168.30.1"]), + "domains": Variant("as", ["syshack.ch"]), + "interface": Variant("s", "eth0"), + "priority": Variant("i", 100), + "vpn": Variant("b", False), + } + ] diff --git a/tests/dbus_service_mocks/network_ip4config.py b/tests/dbus_service_mocks/network_ip4config.py new file mode 100644 index 000000000..79350dd4a --- /dev/null +++ b/tests/dbus_service_mocks/network_ip4config.py @@ -0,0 +1,108 @@ +"""Mock of Network Manager IP4Config service.""" + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return IP4Config() + + +# pylint: disable=invalid-name + + +class IP4Config(DBusServiceMock): + """IP4Config mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/IP4Config/1 + """ + + interface = "org.freedesktop.NetworkManager.IP4Config" + object_path = "/org/freedesktop/NetworkManager/IP4Config/1" + + @dbus_property(access=PropertyAccess.READ) + def Addresses(self) -> "aau": + """Get Addresses.""" + return [[2499979456, 24, 16951488]] + + @dbus_property(access=PropertyAccess.READ) + def AddressData(self) -> "aa{sv}": + """Get AddressData.""" + return [{"address": Variant("s", "192.168.2.148"), "prefix": Variant("u", 24)}] + + @dbus_property(access=PropertyAccess.READ) + def Gateway(self) -> "s": + """Get Gateway.""" + return "192.168.2.1" + + @dbus_property(access=PropertyAccess.READ) + def Routes(self) -> "aau": + """Get Routes.""" + return [[174272, 24, 0, 100], [65193, 16, 0, 1000]] + + @dbus_property(access=PropertyAccess.READ) + def RouteData(self) -> "aa{sv}": + """Get RouteData.""" + return [ + { + "dest": Variant("s", "192.168.2.0"), + "prefix": Variant("u", 24), + "metric": Variant("u", 100), + }, + { + "dest": Variant("s", "169.254.0.0"), + "prefix": Variant("u", 16), + "metric": Variant("u", 1000), + }, + { + "dest": Variant("s", "0.0.0.0"), + "prefix": Variant("u", 0), + "next-hop": Variant("s", "192.168.2.1"), + "metric": Variant("u", 100), + }, + ] + + @dbus_property(access=PropertyAccess.READ) + def NameserverData(self) -> "aa{sv}": + """Get NameserverData.""" + return [{"address": Variant("s", "192.168.2.2")}] + + @dbus_property(access=PropertyAccess.READ) + def Nameservers(self) -> "au": + """Get Nameservers.""" + return [16951488] + + @dbus_property(access=PropertyAccess.READ) + def Domains(self) -> "as": + """Get Domains.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def Searches(self) -> "as": + """Get Searches.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def DnsOptions(self) -> "as": + """Get DnsOptions.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def DnsPriority(self) -> "i": + """Get DnsPriority.""" + return 100 + + @dbus_property(access=PropertyAccess.READ) + def WinsServerData(self) -> "as": + """Get WinsServerData.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def WinsServers(self) -> "au": + """Get WinsServers.""" + return [] diff --git a/tests/dbus_service_mocks/network_ip6config.py b/tests/dbus_service_mocks/network_ip6config.py new file mode 100644 index 000000000..dc2a0ecc0 --- /dev/null +++ b/tests/dbus_service_mocks/network_ip6config.py @@ -0,0 +1,369 @@ +"""Mock of Network Manager IP6Config service.""" + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return IP6Config() + + +# pylint: disable=invalid-name + + +class IP6Config(DBusServiceMock): + """IP6Config mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/IP6Config/1 + """ + + interface = "org.freedesktop.NetworkManager.IP6Config" + object_path = "/org/freedesktop/NetworkManager/IP6Config/1" + + @dbus_property(access=PropertyAccess.READ) + def Addresses(self) -> "a(ayuay)": + """Get Addresses.""" + return [ + [ + bytes( + [ + 0x2A, + 0x3, + 0x1, + 0x69, + 0x3D, + 0xF5, + 0x0, + 0x0, + 0x6B, + 0xE9, + 0x25, + 0x88, + 0xB2, + 0x6A, + 0xA6, + 0x79, + ] + ), + 64, + bytes( + [ + 0xFE, + 0x80, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0xDA, + 0x58, + 0xD7, + 0xFF, + 0xFE, + 0x0, + 0x9C, + 0x69, + ] + ), + ], + [ + bytes( + [ + 0x2A, + 0x3, + 0x1, + 0x69, + 0x3D, + 0xF5, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x2, + 0xF1, + ] + ), + 128, + bytes( + [ + 0xFE, + 0x80, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0xDA, + 0x58, + 0xD7, + 0xFF, + 0xFE, + 0x0, + 0x9C, + 0x69, + ] + ), + ], + ] + + @dbus_property(access=PropertyAccess.READ) + def AddressData(self) -> "aa{sv}": + """Get AddressData.""" + return [ + { + "address": Variant("s", "2a03:169:3df5:0:6be9:2588:b26a:a679"), + "prefix": Variant("u", 64), + }, + { + "address": Variant("s", "2a03:169:3df5::2f1"), + "prefix": Variant("u", 128), + }, + ] + + @dbus_property(access=PropertyAccess.READ) + def Gateway(self) -> "s": + """Get Gateway.""" + return "fe80::da58:d7ff:fe00:9c69" + + @dbus_property(access=PropertyAccess.READ) + def Routes(self) -> "a(ayuayu)": + """Get Routes.""" + return [ + [ + bytes( + [ + 0xFD, + 0x14, + 0x94, + 0x9B, + 0xC9, + 0xCC, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + ] + ), + 48, + bytes( + [ + 0xFE, + 0x80, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0xDA, + 0x58, + 0xD7, + 0xFF, + 0xFE, + 0x0, + 0x9C, + 0x69, + ] + ), + 100, + ], + [ + bytes( + [ + 0x2A, + 0x3, + 0x1, + 0x69, + 0x3D, + 0xF5, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x2, + 0xF1, + ] + ), + 128, + bytes( + [ + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + ] + ), + 100, + ], + [ + bytes( + [ + 0xFE, + 0x80, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + ] + ), + 64, + bytes( + [ + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + ] + ), + 100, + ], + ] + + @dbus_property(access=PropertyAccess.READ) + def RouteData(self) -> "aa{sv}": + """Get RouteData.""" + return [ + { + "dest": Variant("s", "fd14:949b:c9cc::"), + "prefix": Variant("u", 48), + "next-hop": Variant("s", "fe80::da58:d7ff:fe00:9c69"), + "metric": Variant("u", 100), + }, + { + "dest": Variant("s", "2a03:169:3df5::2f1"), + "prefix": Variant("u", 128), + "metric": Variant("u", 100), + }, + { + "dest": Variant("s", "fe80::"), + "prefix": Variant("u", 64), + "metric": Variant("u", 100), + }, + { + "dest": Variant("s", "ff00::"), + "prefix": Variant("u", 8), + "metric": Variant("u", 256), + "table": Variant("u", 255), + }, + ] + + @dbus_property(access=PropertyAccess.READ) + def Nameservers(self) -> "aay": + """Get Nameservers.""" + return [ + bytes( + [ + 0x20, + 0x1, + 0x16, + 0x20, + 0x27, + 0x77, + 0x0, + 0x1, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x10, + ] + ), + bytes( + [ + 0x20, + 0x1, + 0x16, + 0x20, + 0x27, + 0x77, + 0x0, + 0x2, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x20, + ] + ), + ] + + @dbus_property(access=PropertyAccess.READ) + def Domains(self) -> "as": + """Get Domains.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def Searches(self) -> "as": + """Get Searches.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def DnsOptions(self) -> "as": + """Get DnsOptions.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def DnsPriority(self) -> "i": + """Get DnsPriority.""" + return 100 diff --git a/tests/dbus_service_mocks/network_manager.py b/tests/dbus_service_mocks/network_manager.py new file mode 100644 index 000000000..23e385ebd --- /dev/null +++ b/tests/dbus_service_mocks/network_manager.py @@ -0,0 +1,328 @@ +"""Mock of Network Manager service.""" + +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return NetworkManager() + + +# pylint: disable=invalid-name + + +class NetworkManager(DBusServiceMock): + """Network Manager mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager + """ + + interface = "org.freedesktop.NetworkManager" + object_path = "/org/freedesktop/NetworkManager" + version = "1.22.10" + + @dbus_property(access=PropertyAccess.READ) + def Devices(self) -> "ao": + """Get Devices.""" + return [ + "/org/freedesktop/NetworkManager/Devices/1", + "/org/freedesktop/NetworkManager/Devices/3", + ] + + @dbus_property(access=PropertyAccess.READ) + def AllDevices(self) -> "ao": + """Get AllDevices.""" + return [ + "/org/freedesktop/NetworkManager/Devices/1", + "/org/freedesktop/NetworkManager/Devices/2", + "/org/freedesktop/NetworkManager/Devices/3", + ] + + @dbus_property(access=PropertyAccess.READ) + def Checkpoints(self) -> "ao": + """Get Checkpoints.""" + return [] + + @dbus_property(access=PropertyAccess.READ) + def NetworkingEnabled(self) -> "b": + """Get NetworkingEnabled.""" + return True + + @dbus_property() + def WirelessEnabled(self) -> "b": + """Get WirelessEnabled.""" + return True + + @WirelessEnabled.setter + def WirelessEnabled(self, value: "b"): + """Set WirelessEnabled.""" + self.emit_properties_changed({"WirelessEnabled": value}) + + @dbus_property(access=PropertyAccess.READ) + def WirelessHardwareEnabled(self) -> "b": + """Get WirelessHardwareEnabled.""" + return True + + @dbus_property() + def WwanEnabled(self) -> "b": + """Get WwanEnabled.""" + return True + + @WwanEnabled.setter + def WwanEnabled(self, value: "b"): + """Set WwanEnabled.""" + self.emit_properties_changed({"WwanEnabled": value}) + + @dbus_property(access=PropertyAccess.READ) + def WwanHardwareEnabled(self) -> "b": + """Get WwanHardwareEnabled.""" + return True + + @dbus_property() + def WimaxEnabled(self) -> "b": + """Get WimaxEnabled.""" + return False + + @WimaxEnabled.setter + def WimaxEnabled(self, value: "b"): + """Set WimaxEnabled.""" + self.emit_properties_changed({"WimaxEnabled": value}) + + @dbus_property(access=PropertyAccess.READ) + def WimaxHardwareEnabled(self) -> "b": + """Get WimaxHardwareEnabled.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def ActiveConnections(self) -> "ao": + """Get ActiveConnections.""" + return ["/org/freedesktop/NetworkManager/ActiveConnection/1"] + + @dbus_property(access=PropertyAccess.READ) + def PrimaryConnection(self) -> "o": + """Get PrimaryConnection.""" + return "/org/freedesktop/NetworkManager/ActiveConnection/1" + + @dbus_property(access=PropertyAccess.READ) + def PrimaryConnectionType(self) -> "s": + """Get PrimaryConnectionType.""" + return "802-3-ethernet" + + @dbus_property(access=PropertyAccess.READ) + def Metered(self) -> "u": + """Get Metered.""" + return 4 + + @dbus_property(access=PropertyAccess.READ) + def ActivatingConnection(self) -> "o": + """Get ActivatingConnection.""" + return "/" + + @dbus_property(access=PropertyAccess.READ) + def Startup(self) -> "b": + """Get Startup.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def Version(self) -> "s": + """Get Version.""" + return self.version + + @dbus_property(access=PropertyAccess.READ) + def Capabilities(self) -> "au": + """Get Capabilities.""" + return [1] + + @dbus_property(access=PropertyAccess.READ) + def State(self) -> "u": + """Get State.""" + return 70 + + @dbus_property(access=PropertyAccess.READ) + def Connectivity(self) -> "u": + """Get Connectivity.""" + return 4 + + @dbus_property(access=PropertyAccess.READ) + def ConnectivityCheckAvailable(self) -> "b": + """Get ConnectivityCheckAvailable.""" + return True + + @dbus_property() + def ConnectivityCheckEnabled(self) -> "b": + """Get ConnectivityCheckEnabled.""" + return True + + @ConnectivityCheckEnabled.setter + def ConnectivityCheckEnabled(self, value: "b"): + """Set ConnectivityCheckEnabled.""" + self.emit_properties_changed({"ConnectivityCheckEnabled": value}) + + @dbus_property(access=PropertyAccess.READ) + def ConnectivityCheckUri(self) -> "s": + """Get ConnectivityCheckUri.""" + return "http://connectivity-check.ubuntu.com/" + + @dbus_property() + def GlobalDnsConfiguration(self) -> "a{sv}": + """Get GlobalDnsConfiguration.""" + return {} + + @GlobalDnsConfiguration.setter + def GlobalDnsConfiguration(self, value: "a{sv}"): + """Set GlobalDnsConfiguration.""" + self.emit_properties_changed({"GlobalDnsConfiguration": value}) + + @signal() + def CheckPermissions(self) -> None: + """Signal CheckPermissions.""" + + # These signals all seem redundant. Their respective properties fire PropertiesChanged signals + @signal() + def StateChanged(self) -> "u": + """Signal StateChanged.""" + return 70 + + @signal() + def DeviceAdded(self) -> "o": + """Signal DeviceAdded.""" + return "/org/freedesktop/NetworkManager/Devices/2" + + @signal() + def DeviceRemoved(self) -> "o": + """Signal DeviceRemoved.""" + return "/org/freedesktop/NetworkManager/Devices/2" + + @dbus_method() + def Reload(self, flags: "u") -> None: + """Do Reload method.""" + + @dbus_method() + def GetDevices(self) -> "ao": + """Do GetDevices method.""" + return self.Devices + + @dbus_method() + def GetAllDevices(self) -> "ao": + """Do GetAllDevices method.""" + return self.AllDevices + + @dbus_method() + def GetDeviceByIpIface(self, iface: "s") -> "o": + """Do GetDeviceByIpIface method.""" + return "/org/freedesktop/NetworkManager/Devices/1" + + @dbus_method() + def ActivateConnection( + self, connection: "o", device: "o", specific_object: "o" + ) -> "o": + """Do ActivateConnection method.""" + return "/org/freedesktop/NetworkManager/ActiveConnection/1" + + @dbus_method() + def AddAndActivateConnection( + self, connection: "a{sa{sv}}", device: "o", speciic_object: "o" + ) -> "oo": + """Do AddAndActivateConnection method.""" + return [ + "/org/freedesktop/NetworkManager/Settings/1", + "/org/freedesktop/NetworkManager/ActiveConnection/1", + ] + + @dbus_method() + def AddAndActivateConnection2( + self, + connection: "a{sa{sv}}", + device: "o", + speciic_object: "o", + options: "a{sv}", + ) -> "ooa{sv}": + """Do AddAndActivateConnection2 method.""" + return [ + "/org/freedesktop/NetworkManager/Settings/1", + "/org/freedesktop/NetworkManager/ActiveConnection/1", + {}, + ] + + @dbus_method() + def DeactivateConnection(self, active_connection: "o") -> None: + """Do DeactivateConnection method.""" + + @dbus_method() + def Sleep(self, sleep: "b") -> None: + """Do Sleep method.""" + + @dbus_method() + def Enable(self, enable: "b") -> None: + """Do Enable method.""" + + @dbus_method() + def GetPermissions(self) -> "a{ss}": + """Do GetPermissions method.""" + return { + "org.freedesktop.NetworkManager.checkpoint-rollback": "yes", + "org.freedesktop.NetworkManager.enable-disable-connectivity-check": "yes", + "org.freedesktop.NetworkManager.enable-disable-network": "yes", + "org.freedesktop.NetworkManager.enable-disable-statistics": "yes", + "org.freedesktop.NetworkManager.enable-disable-wifi": "yes", + "org.freedesktop.NetworkManager.enable-disable-wimax": "yes", + "org.freedesktop.NetworkManager.enable-disable-wwan": "yes", + "org.freedesktop.NetworkManager.network-control": "yes", + "org.freedesktop.NetworkManager.reload": "yes", + "org.freedesktop.NetworkManager.settings.modify.global-dns": "yes", + "org.freedesktop.NetworkManager.settings.modify.hostname": "yes", + "org.freedesktop.NetworkManager.settings.modify.own": "yes", + "org.freedesktop.NetworkManager.settings.modify.system": "yes", + "org.freedesktop.NetworkManager.sleep-wake": "yes", + "org.freedesktop.NetworkManager.wifi.scan": "yes", + "org.freedesktop.NetworkManager.wifi.share.open": "yes", + "org.freedesktop.NetworkManager.wifi.share.protected": "yes", + } + + @dbus_method() + def SetLogging(self, level: "s", domains: "s") -> None: + """Do SetLogging method.""" + + @dbus_method() + def GetLogging(self) -> "ss": + """Do GetLogging method.""" + return [ + "INFO", + "PLATFORM,RFKILL,ETHER,WIFI,BT,MB,DHCP4,DHCP6,PPP,IP4,IP6,AUTOIP4,DNS,VPN," + "SHARING,SUPPLICANT,AGENTS,SETTINGS,SUSPEND,CORE,DEVICE,OLPC,INFINIBAND," + "FIREWALL,ADSL,BOND,VLAN,BRIDGE,TEAM,CONCHECK,DCB,DISPATCH,AUDIT,SYSTEMD,PROXY", + ] + + @dbus_method() + def CheckConnectivity(self) -> "u": + """Do CheckConnectivity method.""" + return 4 + + @dbus_method() + def state(self) -> "u": + """Do state method.""" + return self.State + + @dbus_method() + def CheckpointCreate(self, devices: "ao", rollback_timeout: "u", flags: "u") -> "o": + """Do CheckpointCreate method.""" + return "/org/freedesktop/NetworkManager/Checkpoint/1" + + @dbus_method() + def CheckpointDestroy(self, checkpoint: "o") -> None: + """Do CheckpointDestroy method.""" + + @dbus_method() + def CheckpointRollback(self, checkpoint: "o") -> "a{su}": + """Do CheckpointRollback method.""" + return {} + + @dbus_method() + def CheckpointAdjustRollbackTimeout( + self, checkpoint: "o", add_timeout: "u" + ) -> None: + """Do CheckpointAdjustRollbackTimeout method.""" diff --git a/tests/dbus_service_mocks/network_settings.py b/tests/dbus_service_mocks/network_settings.py new file mode 100644 index 000000000..c7f2e2c7a --- /dev/null +++ b/tests/dbus_service_mocks/network_settings.py @@ -0,0 +1,94 @@ +"""Mock of Network Manager Settings service.""" + +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.NetworkManager" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Settings() + + +# pylint: disable=invalid-name + + +class Settings(DBusServiceMock): + """Settings mock. + + gdbus introspect --system --dest org.freedesktop.NetworkManager --object-path /org/freedesktop/NetworkManager/Settings + """ + + interface = "org.freedesktop.NetworkManager.Settings" + object_path = "/org/freedesktop/NetworkManager/Settings" + + @dbus_property(access=PropertyAccess.READ) + def Connections(self) -> "ao": + """Get Connections.""" + return ["/org/freedesktop/NetworkManager/Settings/1"] + + @dbus_property(access=PropertyAccess.READ) + def Hostname(self) -> "s": + """Get Hostname.""" + return "homeassistant" + + @dbus_property(access=PropertyAccess.READ) + def CanModify(self) -> "b": + """Get CanModify.""" + return True + + @signal() + def NewConnection(self) -> "o": + """Signal NewConnection.""" + return "/org/freedesktop/NetworkManager/Settings/1" + + @signal() + def ConnectionRemoved(self) -> "o": + """Signal ConnectionRemoved.""" + return "/org/freedesktop/NetworkManager/Settings/1" + + @dbus_method() + def ListConnections(self) -> "ao": + """Do ListConnections method.""" + return self.Connections() + + @dbus_method() + def GetConnectionByUuid(self, uuid: "s") -> "o": + """Do GetConnectionByUuid method.""" + return "/org/freedesktop/NetworkManager/Settings/1" + + @dbus_method() + def AddConnection(self, connection: "a{sa{sv}}") -> "o": + """Do AddConnection method.""" + self.NewConnection() + return "/org/freedesktop/NetworkManager/Settings/1" + + @dbus_method() + def AddConnectionUnsaved(self, connection: "a{sa{sv}}") -> "o": + """Do AddConnectionUnsaved method.""" + return "/org/freedesktop/NetworkManager/Settings/1" + + @dbus_method() + def AddConnection2( + self, settings: "a{sa{sv}}", flags: "u", args: "a{sv}" + ) -> "oa{sv}": + """Do AddConnection2 method.""" + self.NewConnection() + return ["/org/freedesktop/NetworkManager/Settings/1", {}] + + @dbus_method() + def LoadConnections(self, filenames: "as") -> "bas": + """Do LoadConnections method.""" + self.NewConnection() + return [True, []] + + @dbus_method() + def ReloadConnections(self) -> "b": + """Do ReloadConnections method.""" + return True + + @dbus_method() + def SaveHostname(self, hostname: "s") -> None: + """Do SaveHostname method.""" diff --git a/tests/dbus_service_mocks/resolved.py b/tests/dbus_service_mocks/resolved.py index 313d8e5dd..91633f6dc 100644 --- a/tests/dbus_service_mocks/resolved.py +++ b/tests/dbus_service_mocks/resolved.py @@ -48,35 +48,125 @@ class Resolved(DBusServiceMock): def DNS(self) -> "a(iiay)": """Get DNS.""" return [ - [0, 2, bytearray([127, 0, 0, 1])], - [0, 10, bytearray([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1])], + [0, 2, bytes([127, 0, 0, 1])], + [ + 0, + 10, + bytes( + [ + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x1, + ] + ), + ], ] @dbus_property(access=PropertyAccess.READ) def DNSEx(self) -> "a(iiayqs)": """Get DNSEx.""" return [ - [0, 2, bytearray([127, 0, 0, 1]), 0, ""], - [0, 10, bytearray([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), 0, ""], + [0, 2, bytes([127, 0, 0, 1]), 0, ""], + [ + 0, + 10, + bytes( + [ + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x1, + ] + ), + 0, + "", + ], ] @dbus_property(access=PropertyAccess.READ) def FallbackDNS(self) -> "a(iiay)": """Get FallbackDNS.""" return [ - [0, 2, bytearray([1, 1, 1, 1])], - [0, 10, bytearray([38, 6, 71, 0, 71, 0, 0, 0, 0, 0, 0, 0, 0, 0, 17, 17])], + [0, 2, bytes([1, 1, 1, 1])], + [ + 0, + 10, + bytes( + [ + 0x26, + 0x6, + 0x47, + 0x0, + 0x47, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x11, + 0x11, + ] + ), + ], ] @dbus_property(access=PropertyAccess.READ) def FallbackDNSEx(self) -> "a(iiayqs)": """Get FallbackDNSEx.""" return [ - [0, 2, bytearray([1, 1, 1, 1]), 0, "cloudflare-dns.com"], + [0, 2, bytes([1, 1, 1, 1]), 0, "cloudflare-dns.com"], [ 0, 10, - bytearray([38, 6, 71, 0, 71, 0, 0, 0, 0, 0, 0, 0, 0, 0, 17, 17]), + bytes( + [ + 0x26, + 0x6, + 0x47, + 0x0, + 0x47, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x0, + 0x11, + 0x11, + ] + ), 0, "cloudflare-dns.com", ], @@ -85,12 +175,12 @@ class Resolved(DBusServiceMock): @dbus_property(access=PropertyAccess.READ) def CurrentDNSServer(self) -> "(iiay)": """Get CurrentDNSServer.""" - return [0, 2, bytearray([127, 0, 0, 1])] + return [0, 2, bytes([127, 0, 0, 1])] @dbus_property(access=PropertyAccess.READ) def CurrentDNSServerEx(self) -> "(iiayqs)": """Get CurrentDNSServerEx.""" - return [0, 2, bytearray([127, 0, 0, 1]), 0, ""] + return [0, 2, bytes([127, 0, 0, 1]), 0, ""] @dbus_property(access=PropertyAccess.READ) def Domains(self) -> "a(isb)":