Merge pull request #22 from bernhardkaindl/settings-helper-from-simple-settings-dict

Add from_settings_dict() to read a dict without signatures

Author: igo95862

pytest.ini

@@ -0,0 +1,17 @@
+[pytest]
+#
+# Newer version of pytest-asyncio issue a DeprecationWarning unless
+# asyncio_mode is specified. This is because it is planned to change
+# the legacy default of automatically decorating async fixtures from
+# automatc to strict (only all functions and fixtures have to be marked).
+#
+# But only very recent versions of pytest-asyncio versions support
+# the asyncio_mode configuration setting and then issue a config warning,
+# which can't be filtered.
+#
+# Thus, filter the Deprecationwarning of pytest-asyncio for some time
+# until all users can be assumed to use the latest pytest-asyncio.
+#
+filterwarnings =
+    ignore:The 'asyncio_mode' default value will change to 'strict' in future, please explicitly use 'asyncio_mode=strict' or 'asyncio_mode=auto' in pytest configuration file.:DeprecationWarning
+

sdbus_async/networkmanager/settings/base.py

@@ -48,6 +48,21 @@ def from_dbus(
         }
         return cls(**unvarianted_options)
 
+    @classmethod
+    def from_dict(cls,
+                  plain_dict: Dict[str, Any]
+                  ) -> NetworkManagerSettingsMixin:
+        options = {}
+        for dataclass_field in fields(cls):
+            dbus_name = dataclass_field.metadata["dbus_name"]
+            if dbus_name in plain_dict:
+                value = plain_dict[dbus_name]
+                if dataclass_field.metadata["dbus_type"] == 'aa{sv}':
+                    inner_class = cls.setting_name_to_inner_class(dbus_name)
+                    value = [inner_class.from_dict(item) for item in value]
+                options[dataclass_field.name] = value
+        return cls(**options)
+
     @classmethod
     @lru_cache(maxsize=None)
     def setting_name_reverse_mapping(cls) -> Dict[str, str]:

sdbus_async/networkmanager/settings/profile.py

@@ -366,6 +366,22 @@ def from_dbus(cls, dbus_dict: NetworkManagerConnectionProperties
             raise e
         return cls(**unvarianted_options)
 
+    @classmethod
+    def from_settings_dict(
+        cls, settings_dict: Dict[str, Dict[str, Any]]
+    ) -> ConnectionProfile:
+        """Return a ConnectionProfile created from a simple settings dict
+        A simple settings dict uses the same keys as from_dbus() and to_dbus()
+        but without the dbus variable signatures used by NetworkManader.py
+
+        This means a simple settings dict does not use the underscore in keys
+        like the attributes of this class have to use and use "id" and "type".
+        """
+        unvarianted_options: Dict[str, Any] = {
+            SETTING_DBUS_NAME_TO_NAME[k]: SETTING_TO_CLASS[k].from_dict(v)
+            for k, v in settings_dict.items()}
+        return cls(**unvarianted_options)
+
 
 SETTING_DBUS_NAME_TO_NAME: Dict[str, str] = {
     f.metadata['dbus_name']: f.name

tests/settings/test_from_dict_async.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python
+# SPDX-License-Identifier: LGPL-2.1-or-later
+import asyncio
+import contextlib
+import sdbus
+import pytest
+from sdbus_async.networkmanager import (
+    ConnectionProfile,
+    NetworkConnectionSettings,
+    NetworkManagerSettings as SettingsManager,
+    NmSettingsInvalidConnectionError,
+)
+
+# All test coroutines will be treated as marked.
+
+
+def test_wifi_wpa_psk_simple_from_dict() -> ConnectionProfile:
+    """Parse connection and ipv4 settings from dbus using ConnectionProfile"""
+    profile = ConnectionProfile.from_settings_dict(
+        {
+            "connection": {
+                "id": "WirelessWpaPskConnection",
+                "type": "802-11-wireless",
+                "uuid": "16ea7af1-0e35-4036-831e-ced975f48510",
+                "autoconnect": False,
+            },
+            "ipv4": {"method": "auto"},
+            "ipv6": {"method": "disabled"},
+            "802-11-wireless": {
+                "security": "802-11-wireless-security",
+                "ssid": b"CafeSSID",
+            },
+            "802-11-wireless-security": {"key-mgmt": "wpa-psk"},
+        }
+    )
+    assert profile.connection.connection_id == "WirelessWpaPskConnection"
+    assert profile.connection.uuid == "16ea7af1-0e35-4036-831e-ced975f48510"
+    assert profile.connection.connection_type == "802-11-wireless"
+    assert profile.connection.autoconnect is False
+    assert profile.ipv4
+    assert profile.ipv4.method == "auto"
+    assert profile.ipv4.address_data is None
+    assert profile.ipv6
+    assert profile.ipv6.method == "disabled"
+    return profile
+
+
+async def delete_connection_by_uuid(nmset: SettingsManager, uuid: str) -> None:
+    dpath = await nmset.get_connection_by_uuid(uuid)
+    connection_settings = NetworkConnectionSettings(dpath)
+    await connection_settings.delete()
+
+
+# @pytest_asyncio.fixture
+@pytest.mark.asyncio
+async def test_add_wifi_wpa_psk_simple_from_dict() -> None:
+    """Test adding a wireless WPA PSK connection profile(unsaved) from dict"""
+    profile = test_wifi_wpa_psk_simple_from_dict()
+
+    # If we add many connections passing the same id, things get messy. Check:
+    sdbus.set_default_bus(sdbus.sd_bus_open_system())
+    settings = SettingsManager()
+    assert profile.connection.uuid
+    with contextlib.suppress(NmSettingsInvalidConnectionError):
+        await settings.get_connection_by_uuid(profile.connection.uuid)
+        print(f"Deleting existing connection with {profile.connection.uuid}")
+        await delete_connection_by_uuid(settings, profile.connection.uuid)
+    await settings.add_connection_unsaved(profile.to_dbus())
+    await delete_connection_by_uuid(settings, profile.connection.uuid)
+
+
+if __name__ == "__main__":
+    """The tests can be run by pytest (and from IDEs by running this module)"""
+    test_wifi_wpa_psk_simple_from_dict()
+
+    # Test using ConnectionProfile.from_dict() to create a ConnectionProfile
+    # which can be used to add Wireless WPA PSK connection profile from dict
+    # to a running NetworkManager. Requires access and permissions to access
+    # a running NetworkManager. The added connection is not saved and deleted
+    # immediately (and does not have autoconnect enabled):
+    asyncio.run(test_add_wifi_wpa_psk_simple_from_dict())