jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-15 13:16:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use session dbus for network tests (#4191)

Browse Source 
* Use session dbus for network tests

* Don't use session dbus for coresys yet
This commit is contained in:
Mike Degatano 2023-03-17 09:26:41 -04:00 committed by GitHub
parent 8386b5cb3a
commit 0ddce4d9bc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
28 changed files with 2547 additions and 420 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
supervisor/dbus/const.py
View File

@ -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"

2
supervisor/dbus/network/__init__.py
View File

@ -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:

7
supervisor/dbus/network/interface.py
View File

@ -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

9
supervisor/dbus/network/wireless.py
View File

@ -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."""

43
tests/conftest.py
View File

@ -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

28
tests/dbus/network/setting/test_generate.py Normal file
View File

@ -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"]

176
tests/dbus/network/setting/test_init.py
View File

@ -1,87 +1,54 @@
"""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]
@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"]
@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."""
interface = Interface.from_dbus_interface(dbus_interface)
conn = get_connection_from_interface(
interface,
name=dbus_interface.settings.connection.id,
uuid=dbus_interface.settings.connection.uuid,
)
await dbus_interface.settings.update(conn)
assert len(connection_settings_service.Update.calls) == 1
settings = connection_settings_service.Update.calls[0][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(
@ -103,9 +70,7 @@ async def mock_call_dbus_get_settings_signature(
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 settings["ipv4"]["routes"] == Variant("aau", [[8038592, 24, 17435146, 0]])
assert "ipv6" in settings
assert settings["ipv6"]["method"] == Variant("s", "auto")
@ -121,9 +86,7 @@ async def mock_call_dbus_get_settings_signature(
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 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"]
@ -131,55 +94,32 @@ async def mock_call_dbus_get_settings_signature(
assert "vlan" not in settings
async def test_update(coresys: CoreSys):
"""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]
)
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,
)
with patch.object(
DBus,
"call_dbus",
new=mock_call_dbus_get_settings_signature,
):
await coresys.dbus.network.interfaces[TEST_INTERFACE].settings.update(conn)
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()]

34
tests/dbus/network/test_accesspoint.py
View File

@ -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

73
tests/dbus/network/test_connection.py
View File

@ -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

50
tests/dbus/network/test_dns.py
View File

@ -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"

140
tests/dbus/network/test_interface.py
View File

@ -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

99
tests/dbus/network/test_ip_configuration.py
View File

@ -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

115
tests/dbus/network/test_network_manager.py
View File

@ -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()

59
tests/dbus/network/test_setting.py
View File

@ -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

64
tests/dbus/network/test_settings.py
View File

@ -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()]

82
tests/dbus/network/test_wireless.py
View File

@ -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

2
tests/dbus_service_mocks/hostname.py
View File

@ -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":

128
tests/dbus_service_mocks/network_access_point.py Normal file
View File

@ -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

110
tests/dbus_service_mocks/network_active_connection.py Normal file
View File

@ -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]

144
tests/dbus_service_mocks/network_connection_settings.py Normal file
View File

@ -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 {}

371
tests/dbus_service_mocks/network_device.py Normal file
View File

@ -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."""

102
tests/dbus_service_mocks/network_device_wireless.py Normal file
View File

@ -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."""

49
tests/dbus_service_mocks/network_dns_manager.py Normal file
View File

@ -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),
}
]

108
tests/dbus_service_mocks/network_ip4config.py Normal file
View File

@ -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 []

369
tests/dbus_service_mocks/network_ip6config.py Normal file
View File

@ -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

328
tests/dbus_service_mocks/network_manager.py Normal file
View File

@ -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."""

94
tests/dbus_service_mocks/network_settings.py Normal file
View File

@ -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."""

110
tests/dbus_service_mocks/resolved.py
View File

@ -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)":