From 3870063d2b91cdd175ea854cab2d5c0ce3377b2d Mon Sep 17 00:00:00 2001 From: Martin Vidner Date: Sat, 22 Aug 2009 10:25:44 +0200 Subject: [PATCH] Added nm-mock, a fake NetworkManager. It lives on the session bus. Added --session and factored out the bus access to allow easy switching. Now the tests can be run even without a real NM running. --- cnetworkmanager | 27 +++-- dbusclient/__init__.py | 1 + networkmanager/accesspoint.py | 5 +- networkmanager/activeconnection.py | 5 +- networkmanager/applet/__init__.py | 5 +- networkmanager/applet/connection.py | 5 +- networkmanager/applet/service/__init__.py | 5 +- networkmanager/applet/service/connection.py | 3 +- networkmanager/base.py | 21 ++++ networkmanager/device.py | 13 +- networkmanager/monitor.py | 3 +- networkmanager/networkmanager.py | 5 +- nm-mock | 182 ++++++++++++++++++++++++++++ 13 files changed, 252 insertions(+), 28 deletions(-) create mode 100644 networkmanager/base.py create mode 100755 nm-mock diff --git a/cnetworkmanager b/cnetworkmanager index 49b9c7d..8117f35 100755 --- a/cnetworkmanager +++ b/cnetworkmanager @@ -1,7 +1,12 @@ #!/usr/bin/python VERSION = "0.21.1" +# must be set before we ask for signals +from dbus.mainloop.glib import DBusGMainLoop +DBusGMainLoop(set_as_default=True) + import sys +import os import time import dbus from networkmanager import NetworkManager @@ -10,10 +15,8 @@ from networkmanager.applet import NetworkManagerSettings, SYSTEM_SERVICE, USER_S from networkmanager.applet.service import NetworkManagerUserSettings import networkmanager.applet.settings as settings from networkmanager.util import Table +import networkmanager.base -# must be set before we ask for signals -from dbus.mainloop.glib import DBusGMainLoop -DBusGMainLoop(set_as_default=True) # for calling quit import gobject loop = gobject.MainLoop() @@ -103,10 +106,18 @@ op.add_option("--wpa-pass", metavar="KEY", help="use this WPA passphrase") +op.add_option("--session", + action="store_true", default=False, + help="Connect to the session bus for testing") + (options, args) = op.parse_args() Table.terse = options.terse +# TODO: CNM_OPTS, like nmcli +if options.session or os.getenv("CNETWORKMANAGER_TEST") != None: + networkmanager.base.NM_BUS = dbus.SessionBus() + nm = NetworkManager() true_choices = ["1", "on", "yes", "true"] @@ -330,12 +341,12 @@ if options.activate_connection != None: ######## demo ########## -from dbusclient import DBusMio -mio = DBusMio(dbus.SystemBus(), "org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager") -i = mio.Introspect() -d = mio.GetDevices() - if options.demo: + from dbusclient import DBusMio + mio = DBusMio(networkmanager.base.Bus(), "org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager") + i = mio.Introspect() + d = mio.GetDevices() + nm = NetworkManager() # TODO: generic signal (adapt cnm monitor), print name and args diff --git a/dbusclient/__init__.py b/dbusclient/__init__.py index 1ece9ae..5e6362a 100644 --- a/dbusclient/__init__.py +++ b/dbusclient/__init__.py @@ -37,6 +37,7 @@ class DBusMio(dbus.proxies.ProxyObject): # FIXME common for this class, all classes? self.__default_interface = kwargs.pop("default_interface", None) +# print "OP:", object_path super(DBusMio, self).__init__(conn, bus_name, object_path, introspect, follow_name_owner_changes, **kwargs) def __getattr__(self, name): diff --git a/networkmanager/accesspoint.py b/networkmanager/accesspoint.py index f9ed824..172a57a 100644 --- a/networkmanager/accesspoint.py +++ b/networkmanager/accesspoint.py @@ -2,6 +2,7 @@ import dbus from dbusclient import DBusClient from dbusclient.func import * +from base import Base import util class Mode(util.Enum): @@ -10,7 +11,7 @@ class Mode(util.Enum): ADHOC = 1 INFRA = 2 -class AccessPoint(DBusClient): +class AccessPoint(Base): """ Signals: @@ -53,7 +54,7 @@ class AccessPoint(DBusClient): IFACE = "org.freedesktop.NetworkManager.AccessPoint" def __init__(self, opath): - super(AccessPoint, self).__init__(dbus.SystemBus(), self.SERVICE, opath, default_interface=self.IFACE) + super(AccessPoint, self).__init__(self.SERVICE, opath, default_interface=self.IFACE) AccessPoint._add_adaptors( # PropertiesChanged = SA(identity), diff --git a/networkmanager/activeconnection.py b/networkmanager/activeconnection.py index c01b9cc..43e3517 100644 --- a/networkmanager/activeconnection.py +++ b/networkmanager/activeconnection.py @@ -6,9 +6,10 @@ from dbusclient.func import * from applet import Connection from device import Device from accesspoint import AccessPoint +from base import Base from util import Enum -class ActiveConnection(DBusClient): +class ActiveConnection(Base): """ Signals: PropertiesChanged ( a{sv}: properties ) @@ -29,7 +30,7 @@ class ActiveConnection(DBusClient): IFACE = "org.freedesktop.NetworkManager.Connection.Active" def __init__(self, opath): - super(ActiveConnection, self).__init__(dbus.SystemBus(), self.SERVICE, opath, default_interface=self.IFACE) + super(ActiveConnection, self).__init__(self.SERVICE, opath, default_interface=self.IFACE) class State(Enum): UNKNOWN = 0 diff --git a/networkmanager/applet/__init__.py b/networkmanager/applet/__init__.py index 2d260b7..ef00b67 100644 --- a/networkmanager/applet/__init__.py +++ b/networkmanager/applet/__init__.py @@ -6,6 +6,7 @@ which is the nested map returned by NMS.Connection.GetSettings""" import dbus from dbusclient import DBusClient from dbusclient.func import * +from networkmanager.base import Base from networkmanager.applet.connection import Connection __all__ = ["Applet", "Connection",] @@ -16,7 +17,7 @@ USER_SERVICE = "org.freedesktop.NetworkManagerUserSettings" # TODO NMS.System, not in spec -class NetworkManagerSettings(DBusClient): +class NetworkManagerSettings(Base): """NetworkManagerSettings The NM Settings client library @@ -34,7 +35,7 @@ class NetworkManagerSettings(DBusClient): def __init__(self, service): # default_interface because knetworkmanager doesnt provide introspection - super(NetworkManagerSettings, self).__init__(dbus.SystemBus(), service, self.OPATH, default_interface = self.IFACE) + super(NetworkManagerSettings, self).__init__(service, self.OPATH, default_interface = self.IFACE) # need instance specific adaptors for user/system conn factories self._add_adaptor("methods", "ListConnections", MA(seq_adaptor(self._create_connection))) diff --git a/networkmanager/applet/connection.py b/networkmanager/applet/connection.py index 2d769e3..80aeb2e 100644 --- a/networkmanager/applet/connection.py +++ b/networkmanager/applet/connection.py @@ -2,10 +2,11 @@ import dbus from dbusclient import DBusClient +from networkmanager.base import Base #from dbusclient.func import * -class Connection(DBusClient): +class Connection(Base): """NetworkManagerSettings.Connection (including Secrets) Methods: @@ -24,7 +25,7 @@ class Connection(DBusClient): SECRETS_IFACE = "org.freedesktop.NetworkManagerSettings.Connection.Secrets" def __init__(self, service, opath): - super(Connection, self).__init__(dbus.SystemBus(), service, opath, default_interface=self.IFACE) + super(Connection, self).__init__(service, opath, default_interface=self.IFACE) # no adaptors necessary, it seems Connection._add_adaptors( diff --git a/networkmanager/applet/service/__init__.py b/networkmanager/applet/service/__init__.py index 9a4d91f..050c6b9 100644 --- a/networkmanager/applet/service/__init__.py +++ b/networkmanager/applet/service/__init__.py @@ -2,10 +2,11 @@ import dbus import dbus.service import _dbus_bindings from connection import Connection +from networkmanager.base import Bus from networkmanager.applet import USER_SERVICE, SYSTEM_SERVICE def service_pid(name): - bus = dbus.SystemBus() + bus = Bus() DBS = 'org.freedesktop.DBus' DBI = DBS dbo = bus.get_object(DBS, '/') @@ -18,7 +19,7 @@ def service_pid(name): class NetworkManagerSettings(dbus.service.Object): # conmaps is a list def __init__(self, conmaps, requested_name = None): - bus = dbus.SystemBus() + bus = Bus() opath = "/org/freedesktop/NetworkManagerSettings" bus_name = None if requested_name != None: diff --git a/networkmanager/applet/service/connection.py b/networkmanager/applet/service/connection.py index 3a3b399..5d7f1b5 100644 --- a/networkmanager/applet/service/connection.py +++ b/networkmanager/applet/service/connection.py @@ -1,12 +1,13 @@ import dbus import os +from networkmanager.base import Bus from networkmanager.applet.settings import Settings # server analog of cConnection class Connection(dbus.service.Object): def __init__(self, opath, conmap): assert isinstance(conmap, dict) - bus = dbus.SystemBus() + bus = Bus() dbus.service.Object.__init__(self, bus, opath) self.settings = Settings(conmap) diff --git a/networkmanager/base.py b/networkmanager/base.py new file mode 100644 index 0000000..f0730e3 --- /dev/null +++ b/networkmanager/base.py @@ -0,0 +1,21 @@ +import dbus +from dbusclient import DBusClient + +"""The bus where NetworkManager objects live. + +It can be globally changed to a debugging bus for a mock NM service. +Having that as a parameter would complicate the API for the normal +use case, so it is factored out here. +""" +NM_BUS = dbus.SystemBus() +#print "Global Bus:", NM_BUS +def Bus(): +# print "Bus:", NM_BUS + return NM_BUS + +class Base(DBusClient): + def __init__(self, bus_name=None, object_path=None, introspect=True, follow_name_owner_changes=False, **kwargs): + conn = Bus() + super(Base, self).__init__(conn, bus_name, object_path, introspect, follow_name_owner_changes, **kwargs) + + diff --git a/networkmanager/device.py b/networkmanager/device.py index 27ce5d5..eab7469 100644 --- a/networkmanager/device.py +++ b/networkmanager/device.py @@ -6,6 +6,7 @@ from dbusclient import DBusClient from dbusclient.func import * from util import Enum, Flags from accesspoint import AccessPoint, Mode # for Wireless +from base import Base class Ip4Address(object): def __init__(self, int32): @@ -18,7 +19,7 @@ class Ip4Address(object): i32 /= 256 return ".".join(ret) -class IP4Config(DBusClient): +class IP4Config(Base): """ Properties: Addresses - aau - (read) @@ -31,7 +32,7 @@ class IP4Config(DBusClient): IFACE = "org.freedesktop.NetworkManager.IP4Config" def __init__(self, opath): - super(IP4Config, self).__init__(dbus.SystemBus(), self.SERVICE, opath, default_interface = self.IFACE) + super(IP4Config, self).__init__(self.SERVICE, opath, default_interface = self.IFACE) IP4Config._add_adaptors( Addresses = PA(identity), #TODO @@ -40,7 +41,7 @@ IP4Config._add_adaptors( Routes = PA(identity), #TODO ) -class DHCP4Config(DBusClient): +class DHCP4Config(Base): """ Signals: PropertiesChanged ( a{sv}: properties ) @@ -52,14 +53,14 @@ class DHCP4Config(DBusClient): IFACE = "org.freedesktop.NetworkManager.DHCP4Config" def __init__(self, opath): - super(DHCP4Config, self).__init__(dbus.SystemBus(), self.SERVICE, opath, default_interface = self.IFACE) + super(DHCP4Config, self).__init__(self.SERVICE, opath, default_interface = self.IFACE) #DHCP4Config._add_adaptors( # PropertiesChanged = SA(identity), # Options = PA(identity), # ) -class Device(DBusClient): +class Device(Base): """networkmanager device Signals: @@ -161,7 +162,7 @@ class Device(DBusClient): def __init__(self, opath): """Inits the base class, unlike _create""" - super(Device, self).__init__(dbus.SystemBus(), self.SERVICE, opath, default_interface = self.IFACE) + super(Device, self).__init__(self.SERVICE, opath, default_interface = self.IFACE) _constructors = {} diff --git a/networkmanager/monitor.py b/networkmanager/monitor.py index 1232930..a65e3d9 100644 --- a/networkmanager/monitor.py +++ b/networkmanager/monitor.py @@ -4,6 +4,7 @@ import dbus import dbusclient.monitor import networkmanager import device +from base import Bus class Monitor(dbusclient.monitor.Monitor): """A crude signal monitor for NetworkManager. @@ -13,7 +14,7 @@ class Monitor(dbusclient.monitor.Monitor): """ def __init__(self): - super(Monitor, self).__init__(dbus.SystemBus()) + super(Monitor, self).__init__(Bus()) self.watch( self.propc_h, diff --git a/networkmanager/networkmanager.py b/networkmanager/networkmanager.py index ffa721d..04f1405 100644 --- a/networkmanager/networkmanager.py +++ b/networkmanager/networkmanager.py @@ -5,6 +5,7 @@ from dbusclient import DBusClient, object_path from dbusclient.func import * from device import Device from activeconnection import ActiveConnection +from base import Base from util import Enum # gratuitous convertor to test writable properties @@ -15,7 +16,7 @@ def english_to_bool(v): return False return v -class NetworkManager(DBusClient): +class NetworkManager(Base): """networkmanager The NM client library @@ -47,7 +48,7 @@ class NetworkManager(DBusClient): IFACE = "org.freedesktop.NetworkManager" def __init__(self): - super(NetworkManager, self).__init__(dbus.SystemBus(), self.SERVICE, self.OPATH, default_interface=self.IFACE) + super(NetworkManager, self).__init__(self.SERVICE, self.OPATH, default_interface=self.IFACE) class State(Enum): diff --git a/nm-mock b/nm-mock new file mode 100755 index 0000000..c1e002b --- /dev/null +++ b/nm-mock @@ -0,0 +1,182 @@ +#!/usr/bin/python + +# TODO: adapt nm lib to connect to session bus too, and try this mock + +import dbus +import dbus.service +import _dbus_bindings +from dbus.mainloop.glib import DBusGMainLoop +import gobject + +import networkmanager +import networkmanager.device +from networkmanager.base import Bus + +class DBusMock(dbus.service.Object): + def __init__(self, bus, opath, bus_name=None): + print "INIT:", self.__class__, opath + if bus_name != None: + NEE = dbus.exceptions.NameExistsException + try: + bus_name = dbus.service.BusName(bus_name, bus, + replace_existing=True, + do_not_queue=True) + except NEE: + raise NEE("%s (pid %d)" % (bus_name, self.service_pid(bus_name))) + super(DBusMock, self).__init__(bus, opath, bus_name) + self.properties = {} + + @staticmethod + def service_pid(name): + bus = Bus() + DBS = 'org.freedesktop.DBus' + DBI = DBS + dbo = bus.get_object(DBS, '/') + dbi = dbus.Interface(dbo, DBI) + owner = dbi.GetNameOwner(name) + pid = dbi.GetConnectionUnixProcessID(owner) + return pid + + + @dbus.service.method(dbus_interface="org.freedesktop.DBus.Properties", + in_signature="ss", out_signature="v") + def Get(self, interface, prop_name): + # real NM ignores the interface too + value = self.properties.get(prop_name, 42) + print "Get", self.__dbus_object_path__, prop_name, "->", value + return value + +class NetworkManagerMock(DBusMock): + def __init__(self, bus): + opath = "/org/freedesktop/NetworkManager" + bus_name = "org.freedesktop.NetworkManager" + super(NetworkManagerMock, self).__init__(bus, opath, bus_name) + + self.properties.update({ + "State": networkmanager.NetworkManager.State.CONNECTED, + "ActiveConnections": ["/AC1"], + }) + + + @dbus.service.method(dbus_interface="org.freedesktop.NetworkManager", + in_signature="", out_signature="ao") + def GetDevices(self): + return ["/D1", "/D5"] + +class NetworkManagerSettingsMock(DBusMock): + def __init__(self, bus): + opath = "/org/freedesktop/NetworkManagerSettings" + bus_name = "org.freedesktop.NetworkManagerUserSettings" + super(NetworkManagerSettingsMock, self).__init__(bus, opath, bus_name) + + @dbus.service.method(dbus_interface="org.freedesktop.NetworkManagerSettings", + in_signature="", out_signature="ao") + def ListConnections(self): + return ["/C1", "/C2"] + +class ConnectionMock(DBusMock): + def __init__(self, bus, opath): + super(ConnectionMock, self).__init__(bus, opath) + + @dbus.service.method(dbus_interface="org.freedesktop.NetworkManagerSettings.Connection", + in_signature="", out_signature="a{sa{sv}}") + def GetSettings(self): + return { + "connection": { + "id": "mockid", + "type": "mocktype" + }, + "section1": { + "key1": "value1", + "key2": "value2", + }, + "section2": { + "key3": "value4", + "key4": "value4", + }, + } + + +class DeviceMock(DBusMock): + def __init__(self, bus, opath): + super(DeviceMock, self).__init__(bus, opath) + + self.properties.update({ + "Udi": "/hal/udi", + "Driver": "mock driver", + "HwAddress": "11:22:33:44:55:66", + "Ip4Config": "/IC1", + }) + +class WiredDeviceMock(DeviceMock): + def __init__(self, bus, opath): + super(WiredDeviceMock, self).__init__(bus, opath) + + self.properties.update({ + "DeviceType": networkmanager.Device.DeviceType.ETHERNET, + "Interface": "emock0", + }) + +class WirelessDeviceMock(DeviceMock): + def __init__(self, bus, opath): + super(WirelessDeviceMock, self).__init__(bus, opath) + + self.properties.update({ + "DeviceType": networkmanager.Device.DeviceType.WIRELESS, + "ActiveAccessPoint": "/AP1", + "Interface": "wmock0", + }) + + @dbus.service.method(dbus_interface="org.freedesktop.NetworkManager.Device.Wireless", + in_signature="", out_signature="ao") + def GetAccessPoints(self): + return ["/AP1", "/AP2"] + +class ActiveConnectionMock(DBusMock): + def __init__(self, bus, opath): + super(ActiveConnectionMock, self).__init__(bus, opath) + + self.properties.update({ + "ServiceName": "org.freedesktop.NetworkManagerUserSettings", + "Connection": "/C1", + "SpecificObject": "/AP1", + "Devices": ["/D5"], + }) + +class AccessPointMock(DBusMock): + def __init__(self, bus, opath): + super(AccessPointMock, self).__init__(bus, opath) + + self.properties.update({ + "Ssid": dbus.ByteArray("mock-ssid"), + "HwAddress": "11:22:33:44:55:66", + }) + +class Ip4ConfigMock(DBusMock): + def __init__(self, bus, opath): + super(Ip4ConfigMock, self).__init__(bus, opath) + + self.properties.update({ + "Addresses": [(42, 43, 44)], + "Nameservers": [42, 43, 44], + "Domains": ["mockdom1", "mockdom2"], + "Routes": dbus.Array([], signature="au"), # format? fix nmcli + }) + +if __name__ == "__main__": + DBusGMainLoop(set_as_default=True) + + networkmanager.base.NM_BUS = dbus.SessionBus() + bus = Bus() + nmm = NetworkManagerMock(bus) + nms = NetworkManagerSettingsMock(bus) + c1 = ConnectionMock(bus, "/C1") + c2 = ConnectionMock(bus, "/C2") + ac = ActiveConnectionMock(bus, "/AC1") + d1 = WiredDeviceMock(bus, "/D1") + d5 = WirelessDeviceMock(bus, "/D5") + ap1 = AccessPointMock(bus, "/AP1") + ap2 = AccessPointMock(bus, "/AP2") + ic1 = Ip4ConfigMock(bus, "/IC1") + loop = gobject.MainLoop() + loop.run() -- 2.11.4.GIT