pruebas - ¿Cómo escribir una prueba funcional para un servicio DBUS escrito en Python?

tkinter examples

Question

(El título fue: "¿Cómo escribir una prueba unitaria para un servicio DBUS escrito en Python?")

Empecé a escribir un servicio DBUS usando dbus-python, pero tengo problemas para escribir un caso de prueba.

Aquí hay un ejemplo de la prueba que intento crear. Observe que he puesto un bucle de evento GLib en setUp (), aquí es donde el problema llega:

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()

Mi problema es que la implementación DBUS requiere que inicie un bucle de evento para que pueda comenzar a despachar eventos. El enfoque común es usar gobject.MainLoop (). Start () de GLib (aunque no estoy casado con este enfoque, si alguien tiene una mejor sugerencia). Si no inicia un bucle de evento, el servicio aún bloquea y tampoco puede consultarlo.

Si comienzo mi servicio en la prueba, el ciclo de eventos impide que la prueba se complete. Sé que el servicio está funcionando porque puedo consultar el servicio externamente usando la herramienta qdbus, pero no puedo automatizar esto dentro de la prueba que lo inicia.

Estoy considerando hacer algún tipo de proceso bifurcando dentro de la prueba para manejar esto, pero esperaba que alguien pudiera tener una solución más ordenada, o al menos un buen punto de partida para la forma en que escribiría una prueba como esta.

Answer 1

Con algo de ayuda de la publicación de Ali A, he logrado resolver mi problema. El ciclo de eventos de bloqueo debe iniciarse en un proceso separado, de modo que pueda escuchar los eventos sin bloquear la prueba.

Tenga en cuenta que el título de mi pregunta contenía una terminología incorrecta, estaba intentando escribir una prueba funcional, a diferencia de una prueba unitaria. Era consciente de la distinción, pero no me di cuenta de mi error hasta más tarde.

Ajusté el ejemplo en mi pregunta. Se parece mucho al ejemplo "test_pidavim.py", pero utiliza una importación para "dbus.glib" para manejar las dependencias del ciclo glib en lugar de codificar en todas las cosas de 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()

Answer 2

Puede que esté un poco fuera de mi alcance aquí, ya que no conozco a Python y solo entiendo algo de lo que es este "dbus" mágico, pero si lo entiendo correctamente, se requiere crear un entorno de prueba bastante inusual con runloops, extended setup / desmontaje, y así sucesivamente.

La respuesta a tu problema es usar burla . Cree una clase abstracta que defina su interfaz, y luego construya un objeto a partir de eso para usar en su código actual. Para propósitos de prueba, usted construye un objeto simulado que se comunica a través de esa misma interfaz, pero tiene un comportamiento que usted definiría para los propósitos de prueba. Puede usar este enfoque para "simular" el objeto dbus que se ejecuta a través de un bucle de evento, hacer algún trabajo, etc., y luego simplemente concentrarse en probar cómo su clase debe reaccionar ante el resultado del "trabajo" realizado por ese objeto.

Answer 3

Solo necesita asegurarse de que está manejando su ciclo principal correctamente.

def refresh_ui():
while gtk.events_pending():
gtk.main_iteration_do(False)

Esto ejecutará el bucle principal de gtk hasta que haya terminado de procesar todo, en lugar de simplemente ejecutarlo y bloquearlo.

Para ver un ejemplo completo de esto en la práctica, pruebe las unidades de una interfaz dbus, vaya aquí: http://pida.co.uk/trac/browser/pida/editors/vim/test_pidavim.py

Answer 4

Solución simple: no pruebe la unidad a través de dbus.

En su lugar, escriba sus pruebas unitarias para llamar a sus métodos directamente. Eso encaja más naturalmente con la naturaleza de las pruebas unitarias.

Es posible que también desee algunas pruebas de integración automatizadas, que comprueben que se ejecutan a través de dbus, pero no necesitan ser tan completas ni ejecutarse de forma aislada. Puede tener una configuración que inicie una instancia real de su servidor, en un proceso separado.

Answer 5

También puede iniciar el enlace principal en un hilo separado muy simplemente dentro de su método setUp.

Algo como esto:

import threading
class BaseTestCase(unittest.TestCase):
def setUp(self):
myservice = MyDBUSService()
self.loop = gobject.MainLoop()
threading.Thread(name=''glib mainloop'', target=self.loop.run)
def tearDown(self):
self.loop.quit()

Answer 6

Echa un vistazo a la biblioteca python-dbusmock .

Oculta la fea lógica del subproceso detrás de tus ojos, por lo que no tienes que preocuparte por eso en tus pruebas.