如何为用Python编写的DBUS服务编写功能测试？

(标题是:"如何为用Python编写的DBUS服务编写单元测试？")

我已经开始使用dbus-python编写DBUS服务,但是我在编写测试用例时遇到了麻烦.

这是我正在尝试创建的测试示例.请注意,我在setUp()中放置了一个GLib事件循环,这就是问题所在:

import unittest

import gobject
import dbus
import dbus.service
import dbus.glib

class MyDBUSService(dbus.service.Object):
    def __init__(self):
        bus_name = dbus.service.BusName('test.helloservice', bus = dbus.SessionBus())
        dbus.service.Object.__init__(self, bus_name, '/test/helloservice')

    @dbus.service.method('test.helloservice')
    def hello(self):
        return "Hello World!"


class BaseTestCase(unittest.TestCase):

    def setUp(self):
        myservice = MyDBUSService()
        loop = gobject.MainLoop()
        loop.run()
        # === Test blocks here ===

    def testHelloService(self):
        bus = dbus.SessionBus()
        helloservice = bus.get_object('test.helloservice', '/test/helloservice')
        hello = helloservice.get_dbus_method('hello', 'test.helloservice')
        assert hello() == "Hello World!"

if __name__ == '__main__':
    unittest.main()

我的问题是DBUS实现要求您启动一个事件循环,以便它可以开始调度事件.常见的方法是使用GLib的gobject.MainLoop().start()(虽然我没有嫁给这种方法,如果有人有更好的建议).如果您没有启动事件循环,该服务仍会阻止,您也无法查询它.

如果我在测试中启动我的服务,事件循环会阻止测试完成.我知道该服务正在运行,因为我可以使用qdbus工具从外部查询服务,但我无法在启动它的测试中自动执行此操作.

我正在考虑在测试中进行某种处理以处理这个问题,但我希望有人可能有一个更整洁的解决方案,或者至少是一个很好的起点,我将如何编写这样的测试.

在Ali A的帖子的帮助下,我设法解决了我的问题.阻塞事件循环需要启动到一个单独的进程中,以便它可以在不阻塞测试的情况下侦听事件.

请注意我的问题标题包含一些不正确的术语,我试图编写功能测试,而不是单元测试.我知道这种区别,但直到后来才意识到我的错误.

我在我的问题中调整了这个例子.它松散地类似于"test_pidavim.py"示例,但使用导入"dbus.glib"来处理glib循环依赖,而不是在所有DBusGMainLoop中编码:

import unittest

import os
import sys
import subprocess
import time

import dbus
import dbus.service
import dbus.glib
import gobject

class MyDBUSService(dbus.service.Object):

    def __init__(self):
        bus_name = dbus.service.BusName('test.helloservice', bus = dbus.SessionBus())
        dbus.service.Object.__init__(self, bus_name, '/test/helloservice')

    def listen(self):
        loop = gobject.MainLoop()
        loop.run()

    @dbus.service.method('test.helloservice')
    def hello(self):
        return "Hello World!"


class BaseTestCase(unittest.TestCase):

    def setUp(self):
        env = os.environ.copy()
        self.p = subprocess.Popen(['python', './dbus_practice.py', 'server'], env=env)
        # Wait for the service to become available
        time.sleep(1)
        assert self.p.stdout == None
        assert self.p.stderr == None

    def testHelloService(self):
        bus = dbus.SessionBus()
        helloservice = bus.get_object('test.helloservice', '/test/helloservice')
        hello = helloservice.get_dbus_method('hello', 'test.helloservice')
        assert hello() == "Hello World!"

    def tearDown(self):
        # terminate() not supported in Python 2.5
        #self.p.terminate()
        os.kill(self.p.pid, 15)

if __name__ == '__main__':

    arg = ""
    if len(sys.argv) > 1:
        arg = sys.argv[1]

    if arg == "server":
        myservice = MyDBUSService()
        myservice.listen()

    else:
        unittest.main()