Skip to content

Commit acca54c

Browse files
authored
Merge pull request #24 from bernhardkaindl/settings-helper-to_settings_dict
Thank you
2 parents eafd46e + b9eeb2d commit acca54c

File tree

7 files changed

+229
-19
lines changed

7 files changed

+229
-19
lines changed

examples/async/update-connection-async.py

Lines changed: 34 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,21 @@
33
#
44
# Update a property of a connection profile, looked up by connection id
55
#
6+
# This version uses connection_manager.connection_profile().to_settings_dict()
7+
# to retrieve the connection profile from NetworkManager as a settings dict.
8+
#
9+
# It then updates it dynamically using the given arguments:
10+
# The default is to set ipv4.dns-search to ["domain1.com", "domain2.com"].
11+
#
12+
# The dynamically updated dict is then used to update connection profile of NM.
13+
#
614
# The IPv4 settings of connections profiles are documented here:
715
# https://networkmanager.dev/docs/api/latest/settings-ipv4.html
816
#
917
import asyncio
1018
import sdbus
1119
from functools import partial
20+
from sdbus_async.networkmanager import ConnectionProfile
1221
from sdbus_async.networkmanager import NetworkManagerSettings
1322
from sdbus_async.networkmanager import NetworkConnectionSettings
1423
from pprint import pprint
@@ -17,24 +26,34 @@
1726

1827
async def update_connection_async(args: Dict[str, Any]) -> None:
1928
"""Update the settings for [key][entry] of the 1st matching connection"""
29+
30+
# Get the connection path of the connection(s) with the recieved id
2031
fn = NetworkManagerSettings().get_connections_by_id(args["connection_id"])
2132
connection_paths = await fn
22-
settings_domain, setting = args["connection_setting"]
23-
if connection_paths:
24-
connection_settings = NetworkConnectionSettings(connection_paths[0])
25-
properties = await connection_settings.get_settings()
26-
# For compatibility with old tools, NM adds and prefers them, delete:
27-
properties["ipv4"].pop("addresses") # -> Use ["ipv4"]["address-data"]
28-
properties["ipv4"].pop("routes") # ----> Use ["ipv4"]["route-data"]
33+
if not connection_paths:
34+
print(f"No connection {id}, create with add-wifi-psk-connection-async")
35+
return
36+
37+
# Get the profile settings of the first connecttion with given id
38+
connection_manager = NetworkConnectionSettings(connection_paths[0])
39+
existing_connection_profile = await connection_manager.connection_profile()
40+
settings = existing_connection_profile.to_settings_dict()
41+
42+
# Update the given setting's property using the given value
43+
setting, property = args["connection_setting"]
44+
settings[setting][property] = args["value"]
45+
46+
# Get a new ConnectionProfile with the change incorporated
47+
new_connection_profile = ConnectionProfile.from_settings_dict(settings)
48+
49+
# Update the new ConnectionProfile in NetworkManager's configuration
50+
await connection_manager.update(new_connection_profile.to_dbus())
2951

30-
# Update the setting's value in the given configuration group:
31-
properties[settings_domain][setting] = args["value"]
32-
await connection_settings.update(properties)
52+
print(f'Updated {new_connection_profile.connection.uuid}.{setting}:')
53+
partial(pprint, sort_dicts=False)(settings)
3354

34-
print(f'Updated {properties["connection"]["uuid"]}.{settings_domain}:')
35-
partial(pprint, sort_dicts=False)(properties[settings_domain])
36-
else:
37-
print(f"No connection matching {id}")
55+
# Restore the previous connection profile:
56+
await connection_manager.update(existing_connection_profile.to_dbus())
3857

3958

4059
if __name__ == "__main__":
@@ -43,7 +62,6 @@ async def update_connection_async(args: Dict[str, Any]) -> None:
4362
# Set MyConnectionExample.ipv4.dns-search to "domain1.com,domain2.com":
4463
"connection_id": "MyConnectionExample",
4564
"connection_setting": ("ipv4", "dns-search"),
46-
# "as" is the so-called DBus signature, it means "array of strings":
47-
"value": ("as", ["domain1.com", "domain2.com"]),
65+
"value": ["domain1.com", "domain2.com"],
4866
}
4967
asyncio.run(update_connection_async(args))

sdbus_async/networkmanager/__init__.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -250,6 +250,7 @@
250250
NetworkManagerSetting,
251251
NetworkManagerSettingsDomain,
252252
NetworkManagerConnectionProperties,
253+
SettingsDict,
253254
)
254255

255256
DEVICE_TYPE_TO_CLASS = {
@@ -479,4 +480,5 @@
479480
'NetworkManagerSetting',
480481
'NetworkManagerSettingsDomain',
481482
'NetworkManagerConnectionProperties',
483+
'SettingsDict',
482484
)

sdbus_async/networkmanager/objects.py

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
NetworkManagerVPNConnectionInterfaceAsync,
6565
NetworkManagerWifiP2PPeerInterfaceAsync)
6666
from .settings.profile import ConnectionProfile
67+
from .types import NetworkManagerConnectionProperties
6768

6869
NETWORK_MANAGER_SERVICE_NAME = 'org.freedesktop.NetworkManager'
6970

@@ -173,6 +174,19 @@ async def get_connections_by_id(self, connection_id: str) -> List[str]:
173174
connection_paths_with_matching_id.append(connection_path)
174175
return connection_paths_with_matching_id
175176

177+
async def get_settings_by_uuid(
178+
self, connection_uuid: str
179+
) -> NetworkManagerConnectionProperties:
180+
connection = await self.get_connection_by_uuid(connection_uuid)
181+
connection_manager = NetworkConnectionSettings(connection)
182+
connection_settings = await connection_manager.get_settings()
183+
return connection_settings
184+
185+
async def delete_connection_by_uuid(self, connection_uuid: str) -> None:
186+
conn_dbus_path = await self.get_connection_by_uuid(connection_uuid)
187+
connection_settings_manager = NetworkConnectionSettings(conn_dbus_path)
188+
await connection_settings_manager.delete()
189+
176190

177191
class NetworkConnectionSettings(
178192
NetworkManagerSettingsConnectionInterfaceAsync):

sdbus_async/networkmanager/settings/base.py

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,11 @@
1010

1111
class NetworkManagerSettingsMixin:
1212
def to_dbus(self) -> NetworkManagerSettingsDomain:
13-
"""TODO: Add proper docstring"""
13+
"""Return a dbus dictionary for NetworkManager to add/update profiles
14+
15+
The key names provided are exactly as documented in these tables:
16+
https://networkmanager.dev/docs/api/latest/nm-settings-dbus.html
17+
"""
1418
new_dict: NetworkManagerSettingsDomain = {}
1519

1620
for x in fields(self):
@@ -27,6 +31,36 @@ def to_dbus(self) -> NetworkManagerSettingsDomain:
2731

2832
return new_dict
2933

34+
def to_settings_dict(self, defaults: bool = False) -> Dict[str, Any]:
35+
"""Return a simple dictionary using the same key names like the dbus
36+
dict from to_dbus(), but without the dbus signatures returned by it.
37+
38+
The key names provided are exactly as documented in these tables:
39+
https://networkmanager.dev/docs/api/latest/nm-settings-dbus.html
40+
41+
Contrary to dataclasses.asdict(), it provides the orignal dbus keys,
42+
e.g. with numerical prefixes like "802-11-", dashes, and "id"/"type".
43+
44+
In addition, it can be selected if defaults shall be omitted in output,
45+
like NetworkConnectionSettings.get_settings() omits default values:
46+
47+
Because of this, all NetworkManager clients which read profiles have
48+
to have hard-coded knowledge of these defaults. By this omission, they
49+
are part of the stable API: They can be relied upon to never change.
50+
Omitting the defaults makes the typical output really small for review.
51+
"""
52+
new_dict = {}
53+
for x in fields(self):
54+
value = getattr(self, x.name)
55+
if value in [None, {}, []]:
56+
continue
57+
if not defaults and value == x.default:
58+
continue
59+
if x.metadata['dbus_type'] == 'aa{sv}':
60+
value = [x.to_settings_dict(defaults) for x in value]
61+
new_dict[x.metadata['dbus_name']] = value
62+
return new_dict
63+
3064
@classmethod
3165
def _unpack_variant(cls, key: str, signature: str, value: Any) -> Any:
3266
if signature == 'aa{sv}':

sdbus_async/networkmanager/settings/profile.py

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@
5151
from .wireless import WirelessSettings
5252
from .wireless_security import WirelessSecuritySettings
5353
from .wpan import WpanSettings
54-
from ..types import NetworkManagerConnectionProperties
54+
from ..types import NetworkManagerConnectionProperties, SettingsDict
5555

5656

5757
@dataclass
@@ -344,6 +344,27 @@ def to_dbus(self) -> NetworkManagerConnectionProperties:
344344

345345
return new_dict
346346

347+
def to_settings_dict(self, defaults: bool = False) -> SettingsDict:
348+
"""Return a simple dictionary using the same key names like the dbus
349+
dict from to_dbus(), but without the dbus signatures returned by it.
350+
351+
Contrary to dataclasses.asdict(), it provides the orignal dbus keys,
352+
e.g. with numerical prefixes like "802-11-", dashes, and "id"/"type".
353+
354+
The key names provided are exactly as documented in these tables:
355+
https://networkmanager.dev/docs/api/latest/nm-settings-dbus.html
356+
357+
param defaults: Whether properies with default values are returned.
358+
"""
359+
new_dict = {}
360+
for x in fields(self):
361+
settings_class = getattr(self, x.name)
362+
if settings_class:
363+
settingsdomain_dict = settings_class.to_settings_dict(defaults)
364+
if settingsdomain_dict != {}:
365+
new_dict[x.metadata['dbus_name']] = settingsdomain_dict
366+
return new_dict
367+
347368
@property
348369
def dbus_name_to_settings_class(self) -> Dict[str, str]:
349370
return {f.metadata['dbus_name']: f.name
@@ -368,7 +389,7 @@ def from_dbus(cls, dbus_dict: NetworkManagerConnectionProperties
368389

369390
@classmethod
370391
def from_settings_dict(
371-
cls, settings_dict: Dict[str, Dict[str, Any]]
392+
cls, settings_dict: SettingsDict
372393
) -> ConnectionProfile:
373394
"""Return a ConnectionProfile created from a simple settings dict
374395
A simple settings dict uses the same keys as from_dbus() and to_dbus()

sdbus_async/networkmanager/types.py

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,3 +27,6 @@
2727

2828
# All settings and properites of a connection, e.g. returned by get_settings()
2929
NetworkManagerConnectionProperties = Dict[str, NetworkManagerSettingsDomain]
30+
31+
# All settings and properites of a connection, but without dbus signatures
32+
SettingsDict = Dict[str, Dict[str, Any]]

tests/async/test_get_settings.py

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
#!/usr/bin/env python
2+
# SPDX-License-Identifier: LGPL-2.1-or-later
3+
import asyncio
4+
import unittest
5+
import pytest
6+
import sdbus
7+
from sdbus_async.networkmanager import (
8+
ConnectionProfile,
9+
NetworkManagerSettings as SettingsManager,
10+
NetworkManagerConnectionProperties as ConnectionProfileDict,
11+
)
12+
13+
14+
def profile_with_autoconnect_set_to(autoconnect: bool) -> ConnectionProfile:
15+
dummy_profile: ConnectionProfileDict = {
16+
"connection": {
17+
"id": ("s", 'DummyConnection'),
18+
"uuid": ("s", '16ea7af1-0e35-4036-831e-ced975f48510'),
19+
"type": ("s", "dummy"),
20+
"autoconnect": ("b", autoconnect),
21+
"interface-name": ("s", "dummy-notused0"),
22+
},
23+
}
24+
profile_from_dbus_dict = ConnectionProfile.from_dbus(dummy_profile)
25+
# Assert that ConnectionProfile.from_dbus parsed autoconnect correctly:
26+
assert profile_from_dbus_dict.connection.autoconnect is autoconnect
27+
return profile_from_dbus_dict
28+
29+
30+
def set_sdbus_default_bus() -> None:
31+
sdbus.set_default_bus(sdbus.sd_bus_open_system())
32+
33+
34+
async def get_settings_by_networkmanager(
35+
profile: ConnectionProfileDict,
36+
) -> ConnectionProfileDict:
37+
"""Add a connection and return the settings of NetworkManager
38+
Uses the helper method get_gettings_by_uuid() which uses get_settings()
39+
40+
Requires access and permissions to access a running NetworkManager.
41+
The added (dummy) connection is not saved and deleted immediately"""
42+
manager = SettingsManager()
43+
44+
# Temporarily add the new Wifi connection to NetworkManager:
45+
await manager.add_connection_unsaved(profile)
46+
47+
# Get the settings of the temporary connection from NetworkManager:
48+
uuid = profile["connection"]["uuid"][1]
49+
connection_profile_dict = await manager.get_settings_by_uuid(uuid)
50+
51+
# Remove the temporay connection (now that we have read its settings):
52+
await manager.delete_connection_by_uuid(uuid)
53+
return connection_profile_dict
54+
55+
56+
@pytest.mark.asyncio
57+
async def test_autoconnect_false_returned_by_networkmanager() -> None:
58+
"""Test get_settings_by_uuid() and check that autoconnect = False is set"""
59+
set_sdbus_default_bus()
60+
61+
# Set autoconnect = ("b", False) and expect it to be returned by NM:
62+
profile_test = profile_with_autoconnect_set_to(autoconnect=False).to_dbus()
63+
profile_from_manager = await get_settings_by_networkmanager(profile_test)
64+
# Assert that NetworkManager did return autoconnect=False (default is True)
65+
unittest.TestCase().assertTupleEqual(
66+
profile_from_manager["connection"]["autoconnect"], ("b", False)
67+
)
68+
69+
70+
@pytest.mark.asyncio
71+
async def test_autoconnect_true_not_returned_by_networkmanager() -> None:
72+
"""Check autoconnect=True (is default) is not returned by get_settings()"""
73+
set_sdbus_default_bus()
74+
75+
# Set autoconnect = ("b", True) and expect to not be returned by NM:
76+
# (NetworkManager does not return a property if it has the default value)
77+
profile_test = profile_with_autoconnect_set_to(autoconnect=True).to_dbus()
78+
unittest.TestCase().assertTupleEqual(
79+
profile_test["connection"]["autoconnect"], ("b", True)
80+
)
81+
# Assert that NetworkManager did not return autoconnect (default is True)
82+
profile_from_manager = await get_settings_by_networkmanager(profile_test)
83+
assert "autoconnect" not in profile_from_manager["connection"]
84+
85+
86+
def test_autoconnect_true_returned_by_settings_dict_when_requested() -> None:
87+
"""to_settings_dict(defaults=True) returns autoconnect=True (is default)"""
88+
profile_from_dbus_dict = profile_with_autoconnect_set_to(True)
89+
settings_dict = profile_from_dbus_dict.to_settings_dict(defaults=True)
90+
assert settings_dict["connection"]["autoconnect"] is True
91+
92+
93+
def test_autoconnect_true_not_returned_by_to_settings_dict() -> None:
94+
"""to_settings_dict(defaults=False) does not return autoconnect=True"""
95+
96+
# Like checked by test_autoconnect_true_not_returned_by_networkmanager()
97+
# above (which checks that get_settings) does not return autoconnect=True
98+
# (because it is the default), test the same for .to_settings_dict():
99+
profile_from_dbus_dict = profile_with_autoconnect_set_to(True)
100+
settings_dict = profile_from_dbus_dict.to_settings_dict(defaults=False)
101+
102+
# Assert that networkmanager did not return autoconnect (default is True)
103+
assert "autoconnect" not in settings_dict["connection"]
104+
105+
106+
async def check_to_settings_dict_profile_eqal_to_from_settings_dict() -> None:
107+
"""Async main function to run all tests when not run by pytest"""
108+
"""The tests can be run by pytest (and from IDEs by running this module)"""
109+
set_sdbus_default_bus()
110+
await test_autoconnect_false_returned_by_networkmanager()
111+
await test_autoconnect_true_not_returned_by_networkmanager()
112+
test_autoconnect_true_returned_by_settings_dict_when_requested()
113+
test_autoconnect_true_not_returned_by_to_settings_dict()
114+
115+
116+
if __name__ == "__main__":
117+
"""Main function to run all tests when not run by pytest"""
118+
asyncio.run(check_to_settings_dict_profile_eqal_to_from_settings_dict())

0 commit comments

Comments
 (0)