Use pytest

pydbus/tests/publish.py

@@ -3,10 +3,10 @@
 from threading import Thread
 import sys
 
-done = 0
-loop = GLib.MainLoop()
+from pydbus.tests.util import ClientPool, ClientThread
 
-class TestObject(object):
+
+class DummyObject(object):
 	'''
 <node>
 	<interface name='net.lvht.Foo1'>
@@ -23,40 +23,33 @@ def __init__(self, id):
 
 	def HelloWorld(self, a, b):
 		res = self.id + ": " + a + str(b)
-		global done
-		done += 1
-		if done == 2:
-			loop.quit()
-		print(res)
 		return res
 
-bus = SessionBus()
 
-with bus.publish("net.lew21.pydbus.Test", TestObject("Main"), ("Lol", TestObject("Lol"))):
-	remoteMain = bus.get("net.lew21.pydbus.Test")
-	remoteLol = bus.get("net.lew21.pydbus.Test", "Lol")
+def test_multiple_requests():
+	loop = GLib.MainLoop()
+	bus = SessionBus()
 
-	def t1_func():
-		print(remoteMain.HelloWorld("t", 1))
+	with bus.publish("net.lew21.pydbus.Test", DummyObject("Main"), ("Lol", DummyObject("Lol"))):
+		remoteMain = bus.get("net.lew21.pydbus.Test")
+		remoteLol = bus.get("net.lew21.pydbus.Test", "Lol")
 
-	def t2_func():
-		print(remoteLol.HelloWorld("t", 2))
+		def t1_func():
+			return remoteMain.HelloWorld("t", 1)
 
-	t1 = Thread(None, t1_func)
-	t2 = Thread(None, t2_func)
-	t1.daemon = True
-	t2.daemon = True
+		def t2_func():
+			return remoteLol.HelloWorld("t", 2)
 
-	def handle_timeout():
-		print("ERROR: Timeout.")
-		sys.exit(1)
+		pool = ClientPool(loop.quit)
+		t1 = ClientThread(t1_func, loop, pool)
+		t2 = ClientThread(t2_func, loop, pool)
 
-	GLib.timeout_add_seconds(2, handle_timeout)
+		GLib.timeout_add_seconds(2, loop.quit)
 
-	t1.start()
-	t2.start()
+		t1.start()
+		t2.start()
 
-	loop.run()
+		loop.run()
 
-	t1.join()
-	t2.join()
+		assert t1.result == "Main: t1"
+		assert t2.result == "Lol: t2"

pydbus/tests/publish_multiface.py

@@ -1,12 +1,14 @@
 from pydbus import SessionBus
 from gi.repository import GLib
-from threading import Thread
+from threading import Thread, Lock
 import sys
+import time
 
-done = 0
-loop = GLib.MainLoop()
+import pytest
 
-class TestObject(object):
+from pydbus.tests.util import ClientThread
+
+class DummyObject(object):
 	'''
 <node>
 	<interface name='net.lew21.pydbus.tests.Iface1'>
@@ -21,40 +23,90 @@ class TestObject(object):
 	</interface>
 </node>
 	'''
+
+	def __init__(self):
+		self.done = []
+
 	def Method1(self):
-		global done
-		done += 1
-		if done == 2:
-			loop.quit()
-		return "M1"
+		self.done += ["Method1"]
+		return self.done[-1]
 
 	def Method2(self):
-		global done
-		done += 1
-		if done == 2:
-			loop.quit()
-		return "M2"
+		self.done += ["Method2"]
+		return self.done[-1]
 
-bus = SessionBus()
 
-with bus.publish("net.lew21.pydbus.tests.expose_multiface", TestObject()):
-	remote = bus.get("net.lew21.pydbus.tests.expose_multiface")
+@pytest.fixture
+def defaults():
+	loop = GLib.MainLoop()
+	loop.cancelled = False
+	bus = SessionBus()
 
-	def t1_func():
-		print(remote.Method1())
-		print(remote.Method2())
+	obj = DummyObject()
+	with bus.publish("net.lew21.pydbus.tests.expose_multiface", obj):
+		yield loop, obj, bus.get("net.lew21.pydbus.tests.expose_multiface")
 
-	t1 = Thread(None, t1_func)
-	t1.daemon = True
 
-	def handle_timeout():
-		print("ERROR: Timeout.")
-		sys.exit(1)
+def run(loop, func):
+	thread = ClientThread(func, loop)
+	GLib.timeout_add_seconds(2, loop.quit)
 
-	GLib.timeout_add_seconds(2, handle_timeout)
+	thread.start()
+	loop.run()
 
-	t1.start()
+	try:
+		return thread.result
+	except ValueError:
+		pytest.fail('Unable to finish thread')
 
-	loop.run()
 
-	t1.join()
+def test_using_multiface(defaults):
+	def thread_func():
+		results = []
+		results += [remote.Method1()]
+		results += [remote.Method2()]
+		return results
+
+	loop, obj, remote = defaults
+
+	result = run(loop, thread_func)
+
+	assert result == ["Method1", "Method2"]
+	assert obj.done == ["Method1", "Method2"]
+
+
+@pytest.mark.parametrize("interface, method", [
+	("net.lew21.pydbus.tests.Iface1", "Method1"),
+	("net.lew21.pydbus.tests.Iface2", "Method2"),
+])
+def test_using_specific_interface(defaults, interface, method):
+	def thread_func():
+		return getattr(remote, method)()
+
+	loop, obj, remote = defaults
+	remote = remote[interface]
+
+	result = run(loop, thread_func)
+
+	assert result == method
+	assert obj.done == [method]
+
+
+@pytest.mark.parametrize("interface, method", [
+	("net.lew21.pydbus.tests.Iface1", "Method2"),
+	("net.lew21.pydbus.tests.Iface2", "Method1"),
+])
+def test_using_wrong_interface(defaults, interface, method):
+	def thread_func():
+		with pytest.raises(AttributeError) as e:
+			getattr(remote, method)()
+		return e
+
+	loop, obj, remote = defaults
+	remote = remote[interface]
+
+	result = run(loop, thread_func)
+
+	assert str(result.value) == "'{}' object has no attribute '{}'".format(
+		interface, method)
+	assert obj.done == []

pydbus/tests/publish_properties.py

@@ -3,69 +3,59 @@
 from threading import Thread
 import sys
 
-done = 0
-loop = GLib.MainLoop()
+import pytest
 
-class TestObject(object):
+from pydbus.tests.util import ClientPool, ClientThread
+
+
+class DummyObject(object):
 	'''
 <node>
 	<interface name='net.lew21.pydbus.tests.publish_properties'>
 		<property name="Foobar" type="s" access="readwrite"/>
 		<property name="Foo" type="s" access="read"/>
 		<property name="Bar" type="s" access="write"/>
-		<method name='Quit'/>
 	</interface>
 </node>
 	'''
 	def __init__(self):
 		self.Foo = "foo"
 		self.Foobar = "foobar"
 
-	def Quit(self):
-		loop.quit()
-
-bus = SessionBus()
 
-with bus.publish("net.lew21.pydbus.tests.publish_properties", TestObject()):
-	remote = bus.get("net.lew21.pydbus.tests.publish_properties")
-	remote_iface = remote['net.lew21.pydbus.tests.publish_properties']
+def test_properties():
+	bus = SessionBus()
+	loop = GLib.MainLoop()
 
-	def t1_func():
-		for obj in [remote, remote_iface]:
-			assert(obj.Foo == "foo")
-			assert(obj.Foobar == "foobar")
-			obj.Foobar = "barfoo"
-			assert(obj.Foobar == "barfoo")
-			obj.Foobar = "foobar"
-			assert(obj.Foobar == "foobar")
-			obj.Bar = "rab"
+	with bus.publish("net.lew21.pydbus.tests.publish_properties", DummyObject()):
+		remote = bus.get("net.lew21.pydbus.tests.publish_properties")
+		remote_iface = remote['net.lew21.pydbus.tests.publish_properties']
 
-		remote.Foobar = "barfoo"
+		def t1_func():
+			for obj in [remote, remote_iface]:
+				assert(obj.Foo == "foo")
+				assert(obj.Foobar == "foobar")
+				obj.Foobar = "barfoo"
+				assert(obj.Foobar == "barfoo")
+				obj.Foobar = "foobar"
+				assert(obj.Foobar == "foobar")
+				obj.Bar = "rab"
 
-		try:
-			remote.Get("net.lew21.pydbus.tests.publish_properties", "Bar")
-			assert(False)
-		except GLib.GError:
-			pass
-		try:
-			remote.Set("net.lew21.pydbus.tests.publish_properties", "Foo", Variant("s", "haxor"))
-			assert(False)
-		except GLib.GError:
-			pass
-		assert(remote.GetAll("net.lew21.pydbus.tests.publish_properties") == {'Foobar': 'barfoo', 'Foo': 'foo'})
-		remote.Quit()
+			remote.Foobar = "barfoo"
 
-	t1 = Thread(None, t1_func)
-	t1.daemon = True
+			with pytest.raises(GLib.GError):
+				remote.Get("net.lew21.pydbus.tests.publish_properties", "Bar")
+			with pytest.raises(GLib.GError):
+				remote.Set("net.lew21.pydbus.tests.publish_properties", "Foo", Variant("s", "haxor"))
+			assert(remote.GetAll("net.lew21.pydbus.tests.publish_properties") == {'Foobar': 'barfoo', 'Foo': 'foo'})
 
-	def handle_timeout():
-		print("ERROR: Timeout.")
-		sys.exit(1)
+		t1 = ClientThread(t1_func, loop)
 
-	GLib.timeout_add_seconds(2, handle_timeout)
+		GLib.timeout_add_seconds(2, loop.quit)
 
-	t1.start()
+		t1.start()
 
-	loop.run()
+		loop.run()
 
-	t1.join()
+		# The result is not important, but it might reraise the assertion
+		t1.result