system/NetworkManager
system/NetworkManager
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/NetworkManager/NetworkManager synced 2024-07-22 10:46:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

tools: move libnm-glib's fake NM service implementations here

Browse source 
Move libnm-glib's test-fake-nm.py and test-remote-settings-service.py
to tools/, merge them together into a single program, and fix a few
bugs (notably some missing signal emissions in the Settings service).

Although they are currently only used by libnm-glib's tests, they are
generic enough that they could be used by other code in the future
(and in particular, they will be used by libnm's tests as well).
This commit is contained in:
Dan Winship 2014-07-22 18:22:23 -04:00
parent 763ce29e58
commit 5432ef5e52
4 changed files with 125 additions and 166 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
libnm-glib/tests/Makefile.am
View file

@ -35,18 +35,15 @@ test_remote_settings_client_LDADD = \
###########################################
TEST_NM_BIN = test-fake-nm.py
TEST_RSS_BIN = test-remote-settings-service.py
EXTRA_DIST = $(TEST_RSS_BIN) $(TEST_NM_BIN)
TEST_NM_SERVICE = $(top_srcdir)/tools/test-networkmanager-service.py
check-local: test-nm-client test-remote-settings-client
if test -z "$$DBUS_SESSION_BUS_ADDRESS" ; then \
dbus-launch --exit-with-session $(abs_builddir)/test-nm-client $(abs_srcdir) $(TEST_NM_BIN); \
dbus-launch --exit-with-session $(abs_builddir)/test-remote-settings-client $(abs_srcdir) $(TEST_RSS_BIN); \
dbus-launch --exit-with-session $(abs_builddir)/test-nm-client $(abs_srcdir) $(TEST_NM_SERVICE); \
dbus-launch --exit-with-session $(abs_builddir)/test-remote-settings-client $(abs_srcdir) $(TEST_NM_SERVICE); \
else \
$(abs_builddir)/test-nm-client $(abs_srcdir) $(TEST_NM_BIN); \
$(abs_builddir)/test-remote-settings-client $(abs_srcdir) $(TEST_RSS_BIN); \
$(abs_builddir)/test-nm-client $(abs_srcdir) $(TEST_NM_SERVICE); \
$(abs_builddir)/test-remote-settings-client $(abs_srcdir) $(TEST_NM_SERVICE); \
fi;
endif

150
libnm-glib/tests/test-remote-settings-service.py
View file

@ -1,150 +0,0 @@
#!/usr/bin/env python
# -*- Mode: python; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
from __future__ import print_function
from gi.repository import GLib, GObject
import sys
import dbus
import dbus.service
import dbus.mainloop.glib
IFACE_SETTINGS = 'org.freedesktop.NetworkManager.Settings'
IFACE_CONNECTION = 'org.freedesktop.NetworkManager.Settings.Connection'
IFACE_DBUS = 'org.freedesktop.DBus'
class UnknownInterfaceException(dbus.DBusException):
_dbus_error_name = IFACE_DBUS + '.UnknownInterface'
class UnknownPropertyException(dbus.DBusException):
_dbus_error_name = IFACE_DBUS + '.UnknownProperty'
class PermissionDeniedException(dbus.DBusException):
_dbus_error_name = IFACE_SETTINGS + '.PermissionDenied'
mainloop = GObject.MainLoop()
class Connection(dbus.service.Object):
def __init__(self, bus, object_path, settings, remove_func):
dbus.service.Object.__init__(self, bus, object_path)
self.path = object_path
self.settings = settings
self.remove_func = remove_func
self.visible = True
self.props = {}
self.props['Unsaved'] = False
# Properties interface
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, iface):
if iface != IFACE_CONNECTION:
raise UnknownInterfaceException()
return self.props
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, iface, name):
if iface != IFACE_CONNECTION:
raise UnknownInterfaceException()
if not name in self.props.keys():
raise UnknownPropertyException()
return self.props[name]
# Connection methods
@dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='', out_signature='a{sa{sv}}')
def GetSettings(self):
if not self.visible:
raise PermissionDeniedException()
return self.settings
@dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='b', out_signature='')
def SetVisible(self, vis):
self.visible = vis
self.Updated()
@dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='', out_signature='')
def Delete(self):
self.remove_func(self)
self.Removed()
@dbus.service.signal(IFACE_CONNECTION, signature='')
def Removed(self):
pass
@dbus.service.signal(IFACE_CONNECTION, signature='')
def Updated(self):
pass
class Settings(dbus.service.Object):
def __init__(self, bus, object_path):
dbus.service.Object.__init__(self, bus, object_path)
self.connections = {}
self.bus = bus
self.counter = 1
self.props = {}
self.props['Hostname'] = "foobar.baz"
self.props['CanModify'] = True
@dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='', out_signature='ao')
def ListConnections(self):
connections = []
return self.connections.keys()
@dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='a{sa{sv}}', out_signature='o')
def AddConnection(self, settings):
path = "/org/freedesktop/NetworkManager/Settings/Connection/{0}".format(self.counter)
self.counter = self.counter + 1
self.connections[path] = Connection(self.bus, path, settings, self.delete_connection)
print("Added connection {0}".format(path))
return path
def delete_connection(self, connection):
del self.connections[connection.path]
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, iface):
if iface != IFACE_SETTINGS:
raise UnknownInterfaceException()
return self.props
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, iface, name):
if iface != IFACE_SETTINGS:
raise UnknownInterfaceException()
if not name in self.props.keys():
raise UnknownPropertyException()
return self.props[name]
@dbus.service.signal(IFACE_SETTINGS, signature='o')
def NewConnection(self, path):
pass
@dbus.service.method(IFACE_SETTINGS, in_signature='', out_signature='')
def Quit(self):
mainloop.quit()
def quit_cb(user_data):
mainloop.quit()
def main():
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()
obj = Settings(bus, "/org/freedesktop/NetworkManager/Settings")
if not bus.request_name("org.freedesktop.NetworkManager"):
sys.exit(1)
print("Service started")
GLib.timeout_add_seconds(20, quit_cb, None)
try:
mainloop.run()
except Exception as e:
pass
print("Service stopped")
sys.exit(0)
if __name__ == '__main__':
main()

1
tools/Makefile.am
View file

@ -3,4 +3,5 @@ EXTRA_DIST = \
debug-helper.py \
doc-generator.xsl \
run-test-valgrind.sh \
test-networkmanager-service.py \
test-sudo-wrapper.sh

127
libnm-glib/tests/test-fake-nm.py → tools/test-networkmanager-service.py
View file

@ -3,15 +3,14 @@
from __future__ import print_function
from gi.repository import GLib, GObject
from gi.repository import GLib
import sys
import dbus
import dbus.service
import dbus.mainloop.glib
import random
mainloop = GObject.MainLoop()
quit_id = 0
mainloop = GLib.MainLoop()
# NM State
NM_STATE_UNKNOWN = 0
@ -687,7 +686,7 @@ class NetworkManager(ExportedObj):
def AddWiredDevice(self, ifname):
for d in self.devices:
if d.iface == ifname:
raise PermissionDeniedError("Device already added")
raise PermissionDeniedException("Device already added")
dev = WiredDevice(self._bus, ifname)
self.add_device(dev)
return dbus.ObjectPath(dev.path)
@ -696,7 +695,7 @@ class NetworkManager(ExportedObj):
def AddWifiDevice(self, ifname):
for d in self.devices:
if d.iface == ifname:
raise PermissionDeniedError("Device already added")
raise PermissionDeniedException("Device already added")
dev = WifiDevice(self._bus, ifname)
self.add_device(dev)
return dbus.ObjectPath(dev.path)
@ -705,7 +704,7 @@ class NetworkManager(ExportedObj):
def AddWimaxDevice(self, ifname):
for d in self.devices:
if d.iface == ifname:
raise PermissionDeniedError("Device already added")
raise PermissionDeniedException("Device already added")
dev = WimaxDevice(self._bus, ifname)
self.add_device(dev)
return dbus.ObjectPath(dev.path)
@ -748,6 +747,117 @@ class NetworkManager(ExportedObj):
return
raise UnknownDeviceException("Device not found")
###################################################################
IFACE_CONNECTION = 'org.freedesktop.NetworkManager.Settings.Connection'
class Connection(dbus.service.Object):
def __init__(self, bus, object_path, settings, remove_func):
dbus.service.Object.__init__(self, bus, object_path)
self.path = object_path
self.settings = settings
self.remove_func = remove_func
self.visible = True
self.props = {}
self.props['Unsaved'] = False
# Properties interface
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, iface):
if iface != IFACE_CONNECTION:
raise UnknownInterfaceException()
return self.props
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, iface, name):
if iface != IFACE_CONNECTION:
raise UnknownInterfaceException()
if not name in self.props.keys():
raise UnknownPropertyException()
return self.props[name]
# Connection methods
@dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='', out_signature='a{sa{sv}}')
def GetSettings(self):
if not self.visible:
raise PermissionDeniedException()
return self.settings
@dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='b', out_signature='')
def SetVisible(self, vis):
self.visible = vis
self.Updated()
@dbus.service.method(dbus_interface=IFACE_CONNECTION, in_signature='', out_signature='')
def Delete(self):
self.remove_func(self)
self.Removed()
@dbus.service.signal(IFACE_CONNECTION, signature='')
def Removed(self):
pass
@dbus.service.signal(IFACE_CONNECTION, signature='')
def Updated(self):
pass
###################################################################
IFACE_SETTINGS = 'org.freedesktop.NetworkManager.Settings'
class Settings(dbus.service.Object):
def __init__(self, bus, object_path):
dbus.service.Object.__init__(self, bus, object_path)
self.connections = {}
self.bus = bus
self.counter = 1
self.props = {}
self.props['Hostname'] = "foobar.baz"
self.props['CanModify'] = True
@dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='', out_signature='ao')
def ListConnections(self):
return self.connections.keys()
@dbus.service.method(dbus_interface=IFACE_SETTINGS, in_signature='a{sa{sv}}', out_signature='o')
def AddConnection(self, settings):
path = "/org/freedesktop/NetworkManager/Settings/Connection/{0}".format(self.counter)
self.counter = self.counter + 1
self.connections[path] = Connection(self.bus, path, settings, self.delete_connection)
self.NewConnection(path)
self.PropertiesChanged({ 'connections': dbus.Array(self.connections.keys(), 'o') })
return path
def delete_connection(self, connection):
del self.connections[connection.path]
self.PropertiesChanged({ 'connections': dbus.Array(self.connections.keys(), 'o') })
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='s', out_signature='a{sv}')
def GetAll(self, iface):
if iface != IFACE_SETTINGS:
raise UnknownInterfaceException()
return self.props
@dbus.service.method(dbus_interface=dbus.PROPERTIES_IFACE, in_signature='ss', out_signature='v')
def Get(self, iface, name):
if iface != IFACE_SETTINGS:
raise UnknownInterfaceException()
if not name in self.props.keys():
raise UnknownPropertyException()
return self.props[name]
@dbus.service.signal(IFACE_SETTINGS, signature='o')
def NewConnection(self, path):
pass
@dbus.service.signal(IFACE_SETTINGS, signature='a{sv}')
def PropertiesChanged(self, path):
pass
@dbus.service.method(IFACE_SETTINGS, in_signature='', out_signature='')
def Quit(self):
mainloop.quit()
###################################################################
def quit_cb(user_data):
mainloop.quit()
@ -757,12 +867,13 @@ def main():
random.seed()
bus = dbus.SessionBus()
nm = NetworkManager(bus, "/org/freedesktop/NetworkManager")
manager = NetworkManager(bus, "/org/freedesktop/NetworkManager")
settings = Settings(bus, "/org/freedesktop/NetworkManager/Settings")
if not bus.request_name("org.freedesktop.NetworkManager"):
sys.exit(1)
# quit after inactivity to ensure we don't stick around if tests fail
quit_id = GLib.timeout_add_seconds(20, quit_cb, None)
GLib.timeout_add_seconds(20, quit_cb, None)
try:
mainloop.run()