Add a test for NO_REPLY method calls

See https://gitlab.freedesktop.org/dbus/dbus-python/issues/26

Signed-off-by: Simon McVittie <smcv@collabora.com>

2 files changed, 80 insertions, 0 deletions

diff --git a/test/test-client.py b/test/test-client.py
index 41ad826..59e432c 100755
--- a/test/test-client.py
+++ b/test/test-client.py
@@ -41,6 +41,7 @@ except ImportError:
 import dbus
 import _dbus_bindings
 import dbus.glib
+import dbus.lowlevel
 import dbus.service
 
 from dbus._compat import is_py2, is_py3
@@ -83,6 +84,11 @@ NAME = "org.freedesktop.DBus.TestSuitePythonService"
 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
 
+# A random string that we should not transmit on the bus as a result of
+# the NO_REPLY flag
+SHOULD_NOT_HAPPEN = u'a1c04a41-cf98-4923-8487-ddaeeb3f02d1'
+
+
 class TestDBusBindings(unittest.TestCase):
     def setUp(self):
         self.bus = dbus.SessionBus()
@@ -199,6 +205,56 @@ class TestDBusBindings(unittest.TestCase):
         print("Delta: %f" % (b - a))
         self.assertTrue(True)
 
+    def testNoReply(self):
+        failures = []
+        report = []
+        main_loop = gobject.MainLoop()
+
+        def message_filter(conn, m):
+            print('Message filter received message: %r, %r' % (m, m.get_args_list()))
+
+            if conn is not self.bus:
+                failures.append('Message filter called on unexpected bus')
+
+            for a in m.get_args_list():
+                if isinstance(a, unicode):
+                    if SHOULD_NOT_HAPPEN in a:
+                        failures.append('Had an unexpected reply')
+                    elif a == 'TestNoReply report':
+                        report.extend(m.get_args_list())
+                        main_loop.quit()
+
+            return dbus.lowlevel.HANDLER_RESULT_NOT_YET_HANDLED
+
+        self.bus.add_message_filter(message_filter)
+
+        message = dbus.lowlevel.MethodCallMessage(NAME, OBJECT, IFACE, 'TestNoReply')
+        message.append(True)
+        message.append(False)
+        message.set_no_reply(True)
+        self.bus.send_message(message)
+
+        message = dbus.lowlevel.MethodCallMessage(NAME, OBJECT, IFACE, 'TestNoReply')
+        message.append(False)
+        message.append(False)
+        message.set_no_reply(True)
+        self.bus.send_message(message)
+
+        message = dbus.lowlevel.MethodCallMessage(NAME, OBJECT, IFACE, 'TestNoReply')
+        message.append(True)
+        message.append(True)
+        self.bus.send_message(message)
+
+        main_loop.run()
+        self.bus.remove_message_filter(message_filter)
+
+        if failures:
+            self.assertTrue(False, failures)
+
+        self.assertEqual(report[0], 'TestNoReply report')
+        self.assertEqual(report[1], 1)
+        self.assertEqual(report[2], 1)
+
     def testAsyncCalls(self):
         #test sending python types and getting them back async
         print("\n********* Testing Async Calls ***********")
diff --git a/test/test-service.py b/test/test-service.py
index 4289455..2799185 100755
--- a/test/test-service.py
+++ b/test/test-service.py
@@ -69,6 +69,11 @@ NAME = "org.freedesktop.DBus.TestSuitePythonService"
 IFACE = "org.freedesktop.DBus.TestSuiteInterface"
 OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
 
+# A random string that we should not transmit on the bus as a result of
+# the NO_REPLY flag
+SHOULD_NOT_HAPPEN = 'a1c04a41-cf98-4923-8487-ddaeeb3f02d1'
+
+
 class RemovableObject(dbus.service.Object):
     # Part of test for https://bugs.freedesktop.org/show_bug.cgi?id=10457
     @dbus.service.method(IFACE, in_signature='', out_signature='b')
@@ -144,6 +149,8 @@ class TestObject(dbus.service.Object, TestInterface):
                                       object_path + '/Multi2')
         self._multi.add_to_connection(bus_name.get_bus(),
                                       object_path + '/Multi2/3')
+        self._times_no_reply_succeeded = 0
+        self._times_no_reply_failed = 0
 
     """ Echo whatever is sent
     """
@@ -341,6 +348,23 @@ class TestObject(dbus.service.Object, TestInterface):
     def RaiseValueError(self):
         raise ValueError('Wrong!')
 
+    @dbus.service.method(IFACE, in_signature='bb', out_signature='sii')
+    def TestNoReply(self, succeed, report):
+        """Callers are meant to call this with the NO_REPLY flag,
+        unless they call it with report=true."""
+
+        if report:
+            return ('TestNoReply report',
+                    self._times_no_reply_succeeded,
+                    self._times_no_reply_failed)
+
+        if succeed:
+            self._times_no_reply_succeeded += 1
+            return (SHOULD_NOT_HAPPEN, 0, 0)
+        else:
+            self._times_no_reply_failed += 1
+            raise ValueError(SHOULD_NOT_HAPPEN)
+
     @dbus.service.method(IFACE, in_signature='', out_signature='')
     def RaiseDBusExceptionNoTraceback(self):
         class ServerError(dbus.DBusException):