Use pytest

pydbus/tests/context.py

@@ -1,39 +1,40 @@
 from pydbus import SessionBus, connect
 import os
 
-DBUS_SESSION_BUS_ADDRESS = os.getenv("DBUS_SESSION_BUS_ADDRESS")
+import pytest
 
-with connect(DBUS_SESSION_BUS_ADDRESS) as bus:
-	bus.dbus
 
-del bus._dbus
-try:
-	bus.dbus
-	assert(False)
-except RuntimeError:
-	pass
+def test_remove_dbus():
+	DBUS_SESSION_BUS_ADDRESS = os.getenv("DBUS_SESSION_BUS_ADDRESS")
+
+	with connect(DBUS_SESSION_BUS_ADDRESS) as bus:
+		bus.dbus
 
-with SessionBus() as bus:
-	pass
+	del bus._dbus
+	with pytest.raises(RuntimeError):
+		bus.dbus
 
-# SessionBus() and SystemBus() are not closed automatically, so this should work:
-bus.dbus
 
-with bus.request_name("net.lew21.Test"):
-	pass
+def test_use_exited_bus():
+	"""Test using a bus instance after its context manager."""
+	with SessionBus() as bus:
+		pass
+
+	# SessionBus() and SystemBus() are not closed automatically, so this should work:
+	bus.dbus
 
-with bus.request_name("net.lew21.Test"):
-	pass
+	with bus.request_name("net.lew21.Test"):
+		pass
 
-with bus.request_name("net.lew21.Test"):
-	try:
-		bus.request_name("net.lew21.Test")
-		assert(False)
-	except RuntimeError:
+	with bus.request_name("net.lew21.Test"):
 		pass
 
-with bus.watch_name("net.lew21.Test"):
-	pass
+	with bus.request_name("net.lew21.Test"):
+		with pytest.raises(RuntimeError):
+			bus.request_name("net.lew21.Test")
 
-with bus.subscribe(sender="net.lew21.Test"):
-	pass
+	with bus.watch_name("net.lew21.Test"):
+		pass
+
+	with bus.subscribe(sender="net.lew21.Test"):
+		pass

pydbus/tests/identifier.py

@@ -1,19 +1,13 @@
-from __future__ import print_function
+import pytest
+
 from pydbus.identifier import filter_identifier
-import sys
 
-tests = [
+@pytest.mark.parametrize("input, output", [
 	("abc", "abc"),
 	("a_bC", "a_bC"),
 	("a-b_c", "a_b_c"),
 	("a@bc", "abc"),
 	("!@#$%^&*", ""),
-]
-
-ret = 0
-for input, output in tests:
-	if not filter_identifier(input) == output:
-		print("ERROR: filter(" + input + ") returned: " + filter_identifier(input), file=sys.stderr)
-		ret = 1
-
-sys.exit(ret)
+])
+def test_filter_identifier(input, output):
+	assert filter_identifier(input) == output

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 == []