mirror of
https://gitlab.freedesktop.org/dbus/dbus.git
synced 2025-12-25 11:40:10 +01:00
236c7b7384
* python/dbus_bindings.pyx: Memory management foo (global): remove hacky _user_data_references global list (GIL_safe_cunregister_function_handler): userdata now stuffed into tuples. Unref user_data (GIL_safe_cmessage_function_handler): userdata now stuffed into tuples (Connection::__del__): Remove and replace with __dealloc__ method (Connection::add_filter): Stuff user_data into a tuple. Use Py_INCREF to keep tuple from being deallocated instead of the global var hack (Connection::register_object_path): Stuff user_data into a tuple. Use Py_INCREF to keep tuple from being deallocated instead of the global var hack (Connection::register_fallback): Stuff user_data into a tuple. Use Py_INCREF to keep tuple from being deallocated instead of the global var hack (GIL_safe_pending_call_notification): Don't unref the message because it gets unreffed when going out of scope. Py_XDECREF the user_data (PendingCall::__del__): Remove and replace with __dealloc__ method (PendingCall::set_notify): ref the pending call because we will need it to stick around for when the notify callback gets called (Message::__del__): Remove and replace with __dealloc__ method * python/dbus_glib_bindings.pyx (init_gthreads): Changed to gthreads_init to match up with the dbus call * python/glib.py (init_threads): Changed to threads_init to match up with gobject.threads_init(). init_threads is kept for backwards compat but will most likely be deprecated in the future * test/python/test-client.py: - revamp to use Python's unittest functionality - add async call tests - setup threads in glib and dbus so we make sure locks are working
100 lines
3.7 KiB
Python
Executable file
100 lines
3.7 KiB
Python
Executable file
#!/usr/bin/env python
|
|
import sys
|
|
import os
|
|
import unittest
|
|
|
|
builddir = os.environ["DBUS_TOP_BUILDDIR"]
|
|
pydir = builddir + "/python"
|
|
|
|
sys.path.insert(0, pydir)
|
|
sys.path.insert(0, pydir + "/.libs")
|
|
|
|
import dbus
|
|
import dbus_bindings
|
|
import gobject
|
|
import dbus.glib
|
|
|
|
if not dbus.__file__.startswith(pydir):
|
|
raise Exception("DBus modules are not being picked up from the package")
|
|
|
|
if not dbus_bindings.__file__.startswith(pydir):
|
|
raise Exception("DBus modules are not being picked up from the package")
|
|
|
|
class TestDBusBindings(unittest.TestCase):
|
|
def setUp(self):
|
|
self.bus = dbus.SessionBus()
|
|
self.remote_object = self.bus.get_object("org.freedesktop.DBus.TestSuitePythonService", "/org/freedesktop/DBus/TestSuitePythonObject")
|
|
self.iface = dbus.Interface(self.remote_object, "org.freedesktop.DBus.TestSuiteInterface")
|
|
|
|
self.test_types_vals = [1, 12323231, 3.14159265, 99999999.99,
|
|
"dude", "123", "What is all the fuss about?", "gob@gob.com",
|
|
[1,2,3], ["how", "are", "you"], [1.23,2.3], [1], ["Hello"],
|
|
(1,2,3), (1,), (1,"2",3), ("2", "what"), ("you", 1.2),
|
|
{1:"a", 2:"b"}, {"a":1, "b":2}, {1:1.1, 2:2.2}, {1.1:"a", 1.2:"b"},
|
|
[[1,2,3],[2,3,4]], [["a","b"],["c","d"]],
|
|
([1,2,3],"c", 1.2, ["a","b","c"], {"a": (1,"v"), "b": (2,"d")})
|
|
]
|
|
|
|
def testInterfaceKeyword(self):
|
|
#test dbus_interface parameter
|
|
print self.remote_object.Echo("dbus_interface on Proxy test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
|
|
print self.iface.Echo("dbus_interface on Interface test Passed", dbus_interface = "org.freedesktop.DBus.TestSuiteInterface")
|
|
self.assert_(True)
|
|
|
|
def testIntrospection(self):
|
|
#test introspection
|
|
print "\n********* Introspection Test ************"
|
|
print self.remote_object.Introspect(dbus_interface="org.freedesktop.DBus.Introspectable")
|
|
print "Introspection test passed"
|
|
self.assert_(True)
|
|
|
|
def testPythonTypes(self):
|
|
#test sending python types and getting them back
|
|
print "\n********* Testing Python Types ***********"
|
|
|
|
for send_val in self.test_types_vals:
|
|
print "Testing %s"% str(send_val)
|
|
recv_val = self.iface.Echo(send_val)
|
|
self.assertEquals(send_val, recv_val)
|
|
|
|
def testAsyncCalls(self):
|
|
#test sending python types and getting them back async
|
|
print "\n********* Testing Async Calls ***********"
|
|
|
|
|
|
main_loop = gobject.MainLoop()
|
|
class async_check:
|
|
def __init__(self, test_controler, expected_result, do_exit):
|
|
self.expected_result = expected_result
|
|
self.do_exit = do_exit
|
|
self.test_controler = test_controler
|
|
|
|
def callback(self, val):
|
|
if self.do_exit:
|
|
main_loop.quit()
|
|
|
|
self.test_controler.assertEquals(val, self.expected_result)
|
|
|
|
def error_handler(error):
|
|
print error
|
|
if self.do_exit:
|
|
main_loop.quit()
|
|
|
|
self.test_controler.assert_(val, False)
|
|
|
|
last_type = self.test_types_vals[-1]
|
|
for send_val in self.test_types_vals:
|
|
print "Testing %s"% str(send_val)
|
|
check = async_check(self, send_val, last_type == send_val)
|
|
recv_val = self.iface.Echo(send_val,
|
|
reply_handler = check.callback,
|
|
error_handler = check.error_handler)
|
|
|
|
main_loop.run()
|
|
|
|
if __name__ == '__main__':
|
|
gobject.threads_init()
|
|
dbus.glib.init_threads()
|
|
|
|
unittest.main()
|
|
|