core implementation

code bomb implementing the bulk of the osmo-gsm-tester

Change-Id: I53610becbf643ed51b90cfd9debc6992fe211ec9

3 files changed, 172 insertions, 0 deletions

diff --git a/selftest/dbus_test/dbus_server.py b/selftest/dbus_test/dbus_server.py
new file mode 100755
index 0000000..222b28b
--- /dev/null
+++ b/selftest/dbus_test/dbus_server.py
@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+
+# Based on http://stackoverflow.com/questions/22390064/use-dbus-to-just-send-a-message-in-python
+
+# Python DBUS Test Server
+# runs until the Quit() method is called via DBUS
+
+from gi.repository import GLib
+from pydbus import SessionBus
+
+loop = GLib.MainLoop()
+
+class MyDBUSService(object):
+	"""
+		<node>
+			<interface name='net.lew21.pydbus.ClientServerExample'>
+				<method name='Hello'>
+					<arg type='s' name='response' direction='out'/>
+				</method>
+				<method name='EchoString'>
+					<arg type='s' name='a' direction='in'/>
+					<arg type='s' name='response' direction='out'/>
+				</method>
+				<method name='Quit'/>
+			</interface>
+		</node>
+	"""
+
+	def Hello(self):
+		"""returns the string 'Hello, World!'"""
+		return "Hello, World!"
+
+	def EchoString(self, s):
+		"""returns whatever is passed to it"""
+		return s
+
+	def Quit(self):
+		"""removes this object from the DBUS connection and exits"""
+		loop.quit()
+
+bus = SessionBus()
+bus.publish("net.lew21.pydbus.ClientServerExample", MyDBUSService())
+loop.run()
+
diff --git a/selftest/dbus_test/ofono_client.py b/selftest/dbus_test/ofono_client.py
new file mode 100755
index 0000000..6b60f98
--- /dev/null
+++ b/selftest/dbus_test/ofono_client.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+
+'''
+Power on and off some modem on ofono, while running the glib main loop in a
+thread and receiving modem state changes by dbus signals.
+'''
+
+from pydbus import SystemBus, Variant
+import time
+import threading
+import pprint
+
+from gi.repository import GLib
+loop = GLib.MainLoop()
+
+def propchanged(*args, **kwargs):
+        print('-> PROP CHANGED: %r %r' % (args, kwargs))
+
+class GlibMainloop(threading.Thread):
+        def run(self):
+                loop.run()
+
+ml = GlibMainloop()
+ml.start()
+
+try:
+        bus = SystemBus()
+
+        print('\n- list modems')
+        root = bus.get("org.ofono", '/')
+        print(root.Introspect())
+	modems = sorted(root.GetModems())
+        pprint.pprint(modems)
+
+        first_modem_path = modems[0][0]
+        print('\n- first modem %r' % first_modem_path)
+        modem = bus.get("org.ofono", first_modem_path)
+        modem.PropertyChanged.connect(propchanged)
+
+        print(modem.Introspect())
+        print(modem.GetProperties())
+
+        print('\n- set Powered = True')
+        modem.SetProperty('Powered', Variant('b', True))
+        print('call returned')
+        print(modem.GetProperties())
+
+        time.sleep(1)
+
+        print('\n- set Powered = False')
+        modem.SetProperty('Powered', Variant('b', False))
+        print('call returned')
+
+        print(modem.GetProperties())
+finally:
+        loop.quit()
+ml.join()
diff --git a/selftest/dbus_test/ofono_client_one_thread.py b/selftest/dbus_test/ofono_client_one_thread.py
new file mode 100644
index 0000000..96d54bc
--- /dev/null
+++ b/selftest/dbus_test/ofono_client_one_thread.py
@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+
+'''
+Power on and off some modem on ofono, while running the glib main loop in a
+thread and receiving modem state changes by dbus signals.
+'''
+
+from pydbus import SystemBus, Variant
+import time
+import pprint
+
+from gi.repository import GLib
+glib_main_loop = GLib.MainLoop()
+glib_main_ctx = glib_main_loop.get_context()
+
+def propchanged(*args, **kwargs):
+        print('-> PROP CHANGED: %r %r' % (args, kwargs))
+
+
+def pump():
+    global glib_main_ctx
+    print('pump?')
+    while glib_main_ctx.pending():
+        print('* pump')
+        glib_main_ctx.iteration()
+
+def wait(condition):
+    pump()
+    while not condition():
+        time.sleep(.1)
+        pump()
+
+bus = SystemBus()
+
+print('\n- list modems')
+root = bus.get("org.ofono", '/')
+print(root.Introspect())
+modems = sorted(root.GetModems())
+pprint.pprint(modems)
+pump()
+
+first_modem_path = modems[0][0]
+print('\n- first modem %r' % first_modem_path)
+modem = bus.get("org.ofono", first_modem_path)
+modem.PropertyChanged.connect(propchanged)
+
+print(modem.Introspect())
+print(modem.GetProperties())
+
+print('\n- set Powered = True')
+modem.SetProperty('Powered', Variant('b', True))
+print('call returned')
+print('- pump dbus events')
+pump()
+pump()
+print('sleep 1')
+time.sleep(1)
+pump()
+
+
+print('- modem properties:')
+print(modem.GetProperties())
+
+
+print('\n- set Powered = False')
+modem.SetProperty('Powered', Variant('b', False))
+print('call returned')
+
+print(modem.GetProperties())
+
+# vim: tabstop=4 shiftwidth=4 expandtab