# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation; either version 3 of the License, or (at your option) any
# later version.  See http://www.gnu.org/copyleft/lgpl.html for the full text
# of the license.

__author__ = "Iftikhar Ahmad"
__copyright__ = """
(c) 2012 Canonical Ltd.
(c) 2017 - 2022 Martin Pitt <martin@piware.de>
"""

import os
import re
import shutil
import subprocess
import sys
import tracemalloc
import unittest

import dbus
import dbus.mainloop.glib
from gi.repository import GLib

import dbusmock
from dbusmock.templates.networkmanager import (
    CSETTINGS_IFACE,
    MANAGER_IFACE,
    SETTINGS_IFACE,
    SETTINGS_OBJ,
    DeviceState,
    InfrastructureMode,
    NM80211ApSecurityFlags,
    NMActiveConnectionState,
    NMConnectivityState,
    NMState,
)
from packaging.version import Version

tracemalloc.start(25)
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

have_nmcli = shutil.which("nmcli")


@unittest.skipUnless(have_nmcli, "nmcli not installed")
class TestNetworkManager(dbusmock.DBusTestCase):
    """Test mocking NetworkManager"""

    @classmethod
    def setUpClass(cls):
        cls.start_system_bus()
        cls.dbus_con = cls.get_dbus(True)

        os.environ["G_DEBUG"] = "fatal-warnings,fatal-criticals"

        # prepare environment which avoids translations
        cls.lang_env = os.environ.copy()
        try:
            del cls.lang_env["LANG"]
        except KeyError:
            pass
        try:
            del cls.lang_env["LANGUAGE"]
        except KeyError:
            pass
        cls.lang_env["LC_MESSAGES"] = "C"

    def setUp(self):
        (self.p_mock, self.obj_networkmanager) = self.spawn_server_template(
            "networkmanager", {"NetworkingEnabled": True, "WwanEnabled": False}, stdout=subprocess.PIPE
        )
        self.dbusmock = dbus.Interface(self.obj_networkmanager, dbusmock.MOCK_IFACE)
        self.settings = dbus.Interface(self.dbus_con.get_object(MANAGER_IFACE, SETTINGS_OBJ), SETTINGS_IFACE)

    def tearDown(self):
        self.p_mock.stdout.close()
        self.p_mock.terminate()
        self.p_mock.wait()

    def read_general(self):
        return subprocess.check_output(["nmcli", "--nocheck", "general"], env=self.lang_env, text=True)

    def read_networking(self):
        return subprocess.check_output(["nmcli", "--nocheck", "networking"], env=self.lang_env, text=True)

    def read_connection(self):
        return subprocess.check_output(["nmcli", "--nocheck", "connection"], env=self.lang_env, text=True)

    def read_active_connection(self):
        return subprocess.check_output(
            ["nmcli", "--nocheck", "connection", "show", "--active"], env=self.lang_env, text=True
        )

    def read_device(self):
        return subprocess.check_output(["nmcli", "--nocheck", "dev"], env=self.lang_env, text=True)

    def read_device_wifi(self):
        return subprocess.check_output(
            ["nmcli", "--nocheck", "dev", "wifi", "list", "--rescan", "no"],
            env=self.lang_env,
            text=True,
        )

    def test_one_eth_disconnected(self):
        self.dbusmock.AddEthernetDevice("mock_Ethernet1", "eth0", DeviceState.DISCONNECTED)
        out = self.read_device()
        self.assertRegex(out, r"eth0.*\sdisconnected")

    def test_one_eth_connected(self):
        self.dbusmock.AddEthernetDevice("mock_Ethernet1", "eth0", DeviceState.ACTIVATED)
        out = self.read_device()
        self.assertRegex(out, r"eth0.*\sconnected")

    def test_two_eth(self):
        # test with numeric state value
        self.dbusmock.AddEthernetDevice("mock_Ethernet1", "eth0", 30)
        self.dbusmock.AddEthernetDevice("mock_Ethernet2", "eth1", DeviceState.ACTIVATED)
        out = self.read_device()
        self.assertRegex(out, r"eth0.*\sdisconnected")
        self.assertRegex(out, r"eth1.*\sconnected")

    def test_wifi_without_access_points(self):
        self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        out = self.read_device()
        self.assertRegex(out, r"wlan0.*\sconnected")

    def test_eth_and_wifi(self):
        self.dbusmock.AddEthernetDevice("mock_Ethernet1", "eth0", DeviceState.DISCONNECTED)
        self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        out = self.read_device()
        self.assertRegex(out, r"eth0.*\sdisconnected")
        self.assertRegex(out, r"wlan0.*\sconnected")

    def test_one_wifi_with_accesspoints(self):
        wifi = self.dbusmock.AddWiFiDevice("mock_WiFi2", "wlan0", DeviceState.ACTIVATED)
        self.dbusmock.AddAccessPoint(
            wifi,
            "Mock_AP1",
            "AP_1",
            "00:23:F8:7E:12:BB",
            InfrastructureMode.NM_802_11_MODE_ADHOC,
            2425,
            5400,
            82,
            NM80211ApSecurityFlags.NM_802_11_AP_SEC_NONE,
        )
        self.dbusmock.AddAccessPoint(
            wifi,
            "Mock_AP3",
            "AP_3",
            "00:23:F8:7E:12:BC",
            InfrastructureMode.NM_802_11_MODE_INFRA,
            2425,
            5400,
            82,
            NM80211ApSecurityFlags.NM_802_11_AP_SEC_KEY_MGMT_PSK,
        )
        out = self.read_device()
        aps = self.read_device_wifi()
        self.assertRegex(out, r"wlan0.*\sconnected")
        self.assertRegex(aps, r"AP_1.*\sAd-Hoc")
        self.assertRegex(aps, r"AP_3.*\sInfra")

        # connect to non-existing wifi
        res = subprocess.run(["nmcli", "dev", "wifi", "connect", "nonexisting"], check=False, stderr=subprocess.PIPE)
        self.assertNotEqual(res.returncode, 0)
        self.assertRegex(res.stderr, b"No network.*nonexisting")
        self.assertRegex(self.read_device(), r"wlan0.*\sconnected\s+--")

        # TODO: for connecting to password protected Wifi we need to implement secrets agent
        # https://github.com/martinpitt/python-dbusmock/issues/216
        out = subprocess.check_output(["nmcli", "--version"], text=True)
        m = re.search(r"([1-9.]+[0-9])", out)
        assert m, "could not parse version from " + out
        if Version(m.group(1)) >= Version("1.49.3"):
            self.skipTest("https://github.com/martinpitt/python-dbusmock/issues/216")

        # connect to existing wifi with password
        subprocess.check_call(
            ["timeout", "--signal=KILL", "5", "nmcli", "dev", "wifi", "connect", "AP_3", "password", "s3kr1t"]
        )
        self.assertRegex(self.read_device(), r"wlan0.*\sconnected\s+AP_3")

        # connect to existing wifi without password
        subprocess.check_call(["nmcli", "dev", "wifi", "connect", "AP_1"])
        self.assertRegex(self.read_device(), r"wlan0.*\sconnected\s+AP_1")

    def test_two_wifi_with_accesspoints(self):
        wifi1 = self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        wifi2 = self.dbusmock.AddWiFiDevice("mock_WiFi2", "wlan1", DeviceState.UNAVAILABLE)
        self.dbusmock.AddAccessPoint(
            wifi1,
            "Mock_AP0",
            "AP_0",
            "00:23:F8:7E:12:BA",
            InfrastructureMode.NM_802_11_MODE_UNKNOWN,
            2425,
            5400,
            82,
            0x400,
        )
        self.dbusmock.AddAccessPoint(
            wifi2,
            "Mock_AP1",
            "AP_1",
            "00:23:F8:7E:12:BB",
            InfrastructureMode.NM_802_11_MODE_ADHOC,
            2425,
            5400,
            82,
            NM80211ApSecurityFlags.NM_802_11_AP_SEC_KEY_MGMT_PSK,
        )
        self.dbusmock.AddAccessPoint(
            wifi2,
            "Mock_AP3",
            "AP_2",
            "00:23:F8:7E:12:BC",
            InfrastructureMode.NM_802_11_MODE_INFRA,
            2425,
            5400,
            82,
            0x400,
        )
        out = self.read_device()
        aps = self.read_device_wifi()
        self.assertRegex(out, r"wlan0.*\sconnected")
        self.assertRegex(out, r"wlan1.*\sunavailable")
        self.assertRegex(aps, r"AP_0.*\s(Unknown|N/A)")
        self.assertRegex(aps, r"AP_1.*\sAd-Hoc")
        self.assertRegex(aps, r"AP_2.*\sInfra")

    def test_wifi_with_connection(self):
        wifi1 = self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        ap1 = self.dbusmock.AddAccessPoint(
            wifi1,
            "Mock_AP1",
            "The_SSID",
            "00:23:F8:7E:12:BB",
            InfrastructureMode.NM_802_11_MODE_ADHOC,
            2425,
            5400,
            82,
            NM80211ApSecurityFlags.NM_802_11_AP_SEC_KEY_MGMT_PSK,
        )
        con1 = self.dbusmock.AddWiFiConnection(wifi1, "Mock_Con1", "The_SSID", "wpa-psk")

        self.assertRegex(self.read_connection(), r"The_SSID.*\s(802-11-wireless|wifi)")
        self.assertEqual(ap1, "/org/freedesktop/NetworkManager/AccessPoint/Mock_AP1")
        self.assertEqual(con1, "/org/freedesktop/NetworkManager/Settings/Mock_Con1")

        settings = subprocess.check_output(
            ["nmcli", "--nocheck", "connection", "show", "The_SSID"], env=self.lang_env, text=True
        )
        self.assertRegex(settings, r"ipv4.method:\s*auto")
        self.assertRegex(settings, r"ipv4.gateway:\s*--")
        self.assertRegex(settings, r"ipv6.method:\s*auto")
        self.assertRegex(settings, r"ipv4.gateway:\s*--")

    def test_global_state(self):
        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_CONNECTED_GLOBAL)
        self.assertRegex(self.read_general(), r"connected.*\sfull")

        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_CONNECTED_SITE)
        self.assertRegex(self.read_general(), r"connected \(site only\).*\sfull")

        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_CONNECTED_LOCAL)
        self.assertRegex(self.read_general(), r"connected \(local only\).*\sfull")

        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_CONNECTING)
        self.assertRegex(self.read_general(), r"connecting.*\sfull")

        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_DISCONNECTING)
        self.assertRegex(self.read_general(), r"disconnecting.*\sfull")

        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_DISCONNECTED)
        self.assertRegex(self.read_general(), r"disconnected.*\sfull")

        self.dbusmock.SetGlobalConnectionState(NMState.NM_STATE_ASLEEP)
        self.assertRegex(self.read_general(), r"asleep.*\sfull")

    def test_connectivity_state(self):
        self.dbusmock.SetConnectivity(NMConnectivityState.NM_CONNECTIVITY_FULL)
        self.assertRegex(self.read_general(), r"connected.*\sfull")

        self.dbusmock.SetConnectivity(NMConnectivityState.NM_CONNECTIVITY_LIMITED)
        self.assertRegex(self.read_general(), r"connected.*\slimited")

        self.dbusmock.SetConnectivity(NMConnectivityState.NM_CONNECTIVITY_PORTAL)
        self.assertRegex(self.read_general(), r"connected.*\sportal")

        self.dbusmock.SetConnectivity(NMConnectivityState.NM_CONNECTIVITY_NONE)
        self.assertRegex(self.read_general(), r"connected.*\snone")

    def test_networking(self):
        self.dbusmock.SetNetworkingEnabled(False)
        self.assertRegex(self.read_networking(), "disabled")

        self.dbusmock.SetNetworkingEnabled(True)
        self.assertRegex(self.read_networking(), "enabled")

    def test_wifi_with_active_connection(self):
        wifi1 = self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        ap1 = self.dbusmock.AddAccessPoint(
            wifi1,
            "Mock_AP1",
            "The_SSID",
            "00:23:F8:7E:12:BB",
            InfrastructureMode.NM_802_11_MODE_INFRA,
            2425,
            5400,
            82,
            NM80211ApSecurityFlags.NM_802_11_AP_SEC_KEY_MGMT_PSK,
        )
        con1 = self.dbusmock.AddWiFiConnection(wifi1, "Mock_Con1", "The_SSID", "")
        active_con1 = self.dbusmock.AddActiveConnection(
            [wifi1], con1, ap1, "Mock_Active1", NMActiveConnectionState.NM_ACTIVE_CONNECTION_STATE_ACTIVATED
        )

        self.assertEqual(ap1, "/org/freedesktop/NetworkManager/AccessPoint/Mock_AP1")
        self.assertEqual(con1, "/org/freedesktop/NetworkManager/Settings/Mock_Con1")
        self.assertEqual(active_con1, "/org/freedesktop/NetworkManager/ActiveConnection/Mock_Active1")

        self.assertRegex(self.read_general(), r"connected.*\sfull")
        self.assertRegex(self.read_connection(), r"The_SSID.*\s(802-11-wireless|wifi)")
        self.assertRegex(self.read_active_connection(), r"The_SSID.*\s(802-11-wireless|wifi)")
        self.assertRegex(self.read_device_wifi(), "The_SSID")

        self.dbusmock.RemoveActiveConnection(wifi1, active_con1)

        self.assertRegex(self.read_connection(), r"The_SSID.*\s(802-11-wireless|wifi)")
        self.assertFalse(re.compile(r"The_SSID.*\s(802-11-wireless|wifi)").search(self.read_active_connection()))
        self.assertRegex(self.read_device_wifi(), "The_SSID")

        self.dbusmock.RemoveWifiConnection(wifi1, con1)

        self.assertFalse(re.compile(r"The_SSID.*\s(802-11-wireless|wifi)").search(self.read_connection()))
        self.assertRegex(self.read_device_wifi(), "The_SSID")

        self.dbusmock.RemoveAccessPoint(wifi1, ap1)
        self.assertFalse(re.compile(r"The_SSID").search(self.read_device_wifi()))

    def test_add_connection(self):
        self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        uuid = "11111111-1111-1111-1111-111111111111"
        settings = dbus.Dictionary(
            {
                "connection": dbus.Dictionary(
                    {"id": "test connection", "uuid": uuid, "type": "802-11-wireless"}, signature="sv"
                ),
                "802-11-wireless": dbus.Dictionary({"ssid": dbus.ByteArray(b"The_SSID")}, signature="sv"),
            },
            signature="sa{sv}",
        )
        con1 = self.settings.AddConnection(settings)

        self.assertEqual(con1, "/org/freedesktop/NetworkManager/Settings/0")
        self.assertRegex(self.read_connection(), uuid + r".*\s(802-11-wireless|wifi)")

        # Use the same settings, but this one will autoconnect.
        uuid2 = "22222222-2222-2222-2222-222222222222"
        settings["connection"]["autoconnect"] = dbus.Boolean(True)
        settings["connection"]["uuid"] = uuid2

        con2 = self.settings.AddConnection(settings)
        self.assertEqual(con2, "/org/freedesktop/NetworkManager/Settings/1")

        self.assertRegex(self.read_general(), r"connected.*\sfull")
        self.assertRegex(self.read_connection(), uuid2 + r".*\s(802-11-wireless|wifi)")
        self.assertRegex(self.read_active_connection(), uuid2 + r".*\s(802-11-wireless|wifi)")

    def test_update_connection(self):
        uuid = "133d8eb9-6de6-444f-8b37-f40bf9e33226"
        settings = dbus.Dictionary(
            {
                "connection": dbus.Dictionary(
                    {"id": "test wireless", "uuid": uuid, "type": "802-11-wireless"}, signature="sv"
                ),
                "802-11-wireless": dbus.Dictionary({"ssid": dbus.ByteArray(b"The_SSID")}, signature="sv"),
            },
            signature="sa{sv}",
        )

        con1 = self.settings.AddConnection(settings)
        con1_iface = dbus.Interface(self.dbus_con.get_object(MANAGER_IFACE, con1), CSETTINGS_IFACE)

        self.assertEqual(con1, "/org/freedesktop/NetworkManager/Settings/0")
        self.assertRegex(self.read_connection(), uuid + r".*\s(802-11-wireless|wifi)")

        new_settings = dbus.Dictionary(
            {
                "connection": dbus.Dictionary({"id": "test wired", "type": "802-3-ethernet"}, signature="sv"),
                "802-3-ethernet": dbus.Dictionary({"name": "802-3-ethernet"}, signature="sv"),
            },
            signature="sa{sv}",
        )

        con1_iface.Update(new_settings)
        self.assertRegex(self.read_connection(), uuid + r".*\s(ethernet|802-3-ethernet)")

    def test_remove_connection(self):
        wifi1 = self.dbusmock.AddWiFiDevice("mock_WiFi1", "wlan0", DeviceState.ACTIVATED)
        ap1 = self.dbusmock.AddAccessPoint(
            wifi1,
            "Mock_AP1",
            "The_SSID",
            "00:23:F8:7E:12:BB",
            InfrastructureMode.NM_802_11_MODE_INFRA,
            2425,
            5400,
            82,
            NM80211ApSecurityFlags.NM_802_11_AP_SEC_KEY_MGMT_PSK,
        )
        con1 = self.dbusmock.AddWiFiConnection(wifi1, "Mock_Con1", "The_SSID", "")
        self.dbusmock.AddActiveConnection(
            [wifi1], con1, ap1, "Mock_Active1", NMActiveConnectionState.NM_ACTIVE_CONNECTION_STATE_ACTIVATED
        )

        con1_i = dbus.Interface(self.dbus_con.get_object(MANAGER_IFACE, con1), CSETTINGS_IFACE)
        con1_i.Delete()

        self.assertRegex(self.read_general(), r"disconnected.*\sfull")
        self.assertFalse(re.compile(r"The_SSID.*\s802-11-wireless").search(self.read_active_connection()))
        self.assertRegex(self.read_device(), r"wlan0.*\sdisconnected")

    def test_add_remove_settings(self):
        connection = {
            "connection": {
                "timestamp": 1441979296,
                "type": "vpn",
                "id": "a",
                "uuid": "11111111-1111-1111-1111-111111111111",
            },
            "vpn": {"service-type": "org.freedesktop.NetworkManager.openvpn", "data": {"connection-type": "tls"}},
            "ipv4": {
                "routes": dbus.Array([], signature="o"),
                "never-default": True,
                "addresses": dbus.Array([], signature="o"),
                "dns": dbus.Array([], signature="o"),
                "method": "auto",
            },
            "ipv6": {
                "addresses": dbus.Array([], signature="o"),
                "ip6-privacy": 0,
                "dns": dbus.Array([], signature="o"),
                "never-default": True,
                "routes": dbus.Array([], signature="o"),
                "method": "auto",
            },
        }

        connectionA = self.settings.AddConnection(connection)
        connection["connection"]["id"] = "b"
        connection["connection"]["uuid"] = "11111111-1111-1111-1111-111111111112"
        connectionB = self.settings.AddConnection(connection)
        self.assertEqual(self.settings.ListConnections(), [connectionA, connectionB])

        connectionA_i = dbus.Interface(self.dbus_con.get_object(MANAGER_IFACE, connectionA), CSETTINGS_IFACE)
        connectionA_i.Delete()
        self.assertEqual(self.settings.ListConnections(), [connectionB])

        connection["connection"]["id"] = "c"
        connection["connection"]["uuid"] = "11111111-1111-1111-1111-111111111113"
        connectionC = self.settings.AddConnection(connection)
        self.assertEqual(self.settings.ListConnections(), [connectionB, connectionC])

    def test_add_update_settings(self):
        connection = {
            "connection": {
                "timestamp": 1441979296,
                "type": "vpn",
                "id": "a",
                "uuid": "11111111-1111-1111-1111-111111111111",
            },
            "vpn": {
                "service-type": "org.freedesktop.NetworkManager.openvpn",
                "data": dbus.Dictionary({"connection-type": "tls"}, signature="ss"),
            },
            "ipv4": {
                "routes": dbus.Array([], signature="o"),
                "never-default": True,
                "addresses": dbus.Array([], signature="o"),
                "dns": dbus.Array([], signature="o"),
                "method": "auto",
            },
            "ipv6": {
                "addresses": dbus.Array([], signature="o"),
                "ip6-privacy": 0,
                "dns": dbus.Array([], signature="o"),
                "never-default": True,
                "routes": dbus.Array([], signature="o"),
                "method": "auto",
            },
        }

        connectionA = self.settings.AddConnection(connection)
        self.assertEqual(self.settings.ListConnections(), [connectionA])

        connectionA_i = dbus.Interface(self.dbus_con.get_object(MANAGER_IFACE, connectionA), CSETTINGS_IFACE)
        connection["connection"]["id"] = "b"

        def do_update():
            connectionA_i.Update(connection)

        caught = []
        ml = GLib.MainLoop()

        def catch(*_, **kwargs):
            if (
                kwargs["interface"] == "org.freedesktop.NetworkManager.Settings.Connection"
                and kwargs["member"] == "Updated"
            ):
                caught.append(kwargs["path"])
                ml.quit()

        self.dbus_con.add_signal_receiver(
            catch, interface_keyword="interface", path_keyword="path", member_keyword="member"
        )

        GLib.timeout_add(200, do_update)
        # ensure that the loop quits even when we don't catch anything
        GLib.timeout_add(3000, ml.quit)
        ml.run()

        self.assertEqual(connectionA_i.GetSettings(), connection)
        self.assertEqual(caught, [connectionA])

    def test_settings_secrets(self):
        secrets = dbus.Dictionary(
            {
                "cert-pass": "certificate password",
                "password": "the password",
            },
            signature="ss",
        )

        connection = {
            "connection": {
                "timestamp": 1441979296,
                "type": "vpn",
                "id": "a",
                "uuid": "11111111-1111-1111-1111-111111111111",
            },
            "vpn": {
                "service-type": "org.freedesktop.NetworkManager.openvpn",
                "data": dbus.Dictionary(
                    {
                        "connection-type": "password-tls",
                        "remote": "remotey",
                        "ca": "/my/ca.crt",
                        "cert": "/my/cert.crt",
                        "cert-pass-flags": "1",
                        "key": "/my/key.key",
                        "password-flags": "1",
                    },
                    signature="ss",
                ),
                "secrets": secrets,
            },
            "ipv4": {
                "routes": dbus.Array([], signature="o"),
                "never-default": True,
                "addresses": dbus.Array([], signature="o"),
                "dns": dbus.Array([], signature="o"),
                "method": "auto",
            },
            "ipv6": {
                "addresses": dbus.Array([], signature="o"),
                "ip6-privacy": 0,
                "dns": dbus.Array([], signature="o"),
                "never-default": True,
                "routes": dbus.Array([], signature="o"),
                "method": "auto",
            },
        }

        connectionPath = self.settings.AddConnection(connection)
        self.assertEqual(self.settings.ListConnections(), [connectionPath])

        connection_i = dbus.Interface(self.dbus_con.get_object(MANAGER_IFACE, connectionPath), CSETTINGS_IFACE)

        # We expect there to be no secrets in the normal settings dict
        del connection["vpn"]["secrets"]
        self.assertEqual(connection_i.GetSettings(), connection)

        # Secrets request should contain just vpn section with the secrets in
        self.assertEqual(connection_i.GetSecrets("vpn"), {"vpn": {"secrets": secrets}})

    def test_get_conn_by_uuid(self):
        uuid = "133d8eb9-6de6-444f-8b37-f40bf9e33226"
        settings = dbus.Dictionary(
            {
                "connection": dbus.Dictionary(
                    {"id": "test wireless", "uuid": uuid, "type": "802-11-wireless"}, signature="sv"
                ),
                "802-11-wireless": dbus.Dictionary({"ssid": dbus.ByteArray(b"The_SSID")}, signature="sv"),
            },
            signature="sa{sv}",
        )
        connectionPath = self.settings.AddConnection(settings)
        self.assertEqual(self.settings.GetConnectionByUuid(uuid), connectionPath)

        with self.assertRaisesRegex(dbus.exceptions.DBusException, "No connection with the UUID was found."):
            self.settings.GetConnectionByUuid("123123123213213")


if __name__ == "__main__":
    unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout))