¿Cómo se pueden probar los errores de Twisted Deferred sin errores con prueba?

Question

Tengo un código Twisted que crea múltiples cadenas de Deferreds. Algunos de estos pueden fallar sin tener un error que los vuelva a poner en la cadena de devolución de llamada. No he podido escribir una prueba unitaria para este código; el error diferido hace que la prueba falle después de que se haya completado el código de prueba. ¿Cómo puedo escribir una prueba de unidad de paso para este código? ¿Se espera que cada Deferred que podría fallar en la operación normal tenga un error al final de la cadena que lo vuelva a poner en la cadena de devolución de llamada?

Lo mismo ocurre cuando hay un Deferred diferido en una Lista diferida, a menos que cree la Lista diferida con consumirErrores. Este es el caso incluso cuando DeferredList se crea con fireOnOneErrback y recibe un error que lo coloca de nuevo en la cadena de devolución de llamada. ¿Hay alguna implicación para consumirErrores además de suprimir las fallas de prueba y el registro de errores? ¿Debería cada Deferred que puede fallar sin un error colocarse una Lista Diferida?

ejemplo prueba de ejemplo de código:

from twisted.trial import unittest 
from twisted.internet import defer 

def get_dl(**kwargs): 
    "Return a DeferredList with a failure and any kwargs given." 
    return defer.DeferredList(
     [defer.succeed(True), defer.fail(ValueError()), defer.succeed(True)], 
     **kwargs) 

def two_deferreds(): 
    "Create a failing Deferred, and create and return a succeeding Deferred." 
    d = defer.fail(ValueError()) 
    return defer.succeed(True) 


class DeferredChainTest(unittest.TestCase): 

    def check_success(self, result): 
     "If we're called, we're on the callback chain."   
     self.fail() 

    def check_error(self, failure): 
     """ 
     If we're called, we're on the errback chain. 
     Return to put us back on the callback chain. 
     """ 
     return True 

    def check_error_fail(self, failure): 
     """ 
     If we're called, we're on the errback chain. 
     """ 
     self.fail()   

    # This fails after all callbacks and errbacks have been run, with the 
    # ValueError from the failed defer, even though we're 
    # not on the errback chain. 
    def test_plain(self): 
     """ 
     Test that a DeferredList without arguments is on the callback chain. 
     """ 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl().addErrback(self.check_error_fail) 

    # This fails after all callbacks and errbacks have been run, with the 
    # ValueError from the failed defer, even though we're 
    # not on the errback chain. 
    def test_fire(self): 
     """ 
     Test that a DeferredList with fireOnOneErrback errbacks on failure, 
     and that an errback puts it back on the callback chain. 
     """ 
     # check_success asserts that we don't callback. 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl(fireOnOneErrback=True).addCallbacks(
      self.check_success, self.check_error).addErrback(
      self.check_error_fail) 

    # This succeeds. 
    def test_consume(self): 
     """ 
     Test that a DeferredList with consumeErrors errbacks on failure, 
     and that an errback puts it back on the callback chain. 
     """ 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl(consumeErrors=True).addErrback(self.check_error_fail) 

    # This succeeds. 
    def test_fire_consume(self): 
     """ 
     Test that a DeferredList with fireOnOneCallback and consumeErrors 
     errbacks on failure, and that an errback puts it back on the 
     callback chain. 
     """ 
     # check_success asserts that we don't callback. 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl(fireOnOneErrback=True, consumeErrors=True).addCallbacks(
      self.check_success, self.check_error).addErrback(
      self.check_error_fail) 

    # This fails after all callbacks and errbacks have been run, with the 
    # ValueError from the failed defer, even though we're 
    # not on the errback chain. 
    def test_two_deferreds(self): 
     # check_error_fail asserts that we are on the callback chain.   
     return two_deferreds().addErrback(self.check_error_fail) 

Answer 1

Hay dos cosas importantes acerca de juicio relacionados con esta cuestión.

Primero, un método de prueba no pasará si se registra una falla mientras se está ejecutando. Los diferidos que son basura recolectada con un resultado de falla provocan que se registre la falla.

En segundo lugar, un método de prueba que devuelve un aplazado no pasará si el aplazado dispara con una falla.

Esto significa que ninguna de estas pruebas puede pasar:

def test_logit(self): 
    defer.fail(Exception("oh no")) 

def test_returnit(self): 
    return defer.fail(Exception("oh no")) 

Esto es importante porque el primer caso, el caso de un ser basura diferido recogido con un resultado Fracaso, significa que ocurrió un error que nadie manejado. Es similar a la forma en que Python informará un seguimiento de pila si una excepción alcanza el nivel superior de su programa.

Del mismo modo, el segundo caso es una red de seguridad provista por prueba. Si un método de prueba síncrona genera una excepción, la prueba no pasa. Por lo tanto, si un método de prueba arroja un Deferred, el Deferred debe tener un resultado exitoso para que la prueba pase.

Sin embargo, existen herramientas para tratar cada uno de estos casos. Después de todo, si no puede realizar una prueba de aprobación para una API que devolvió un Deferred que emitió un error a veces, entonces nunca podría probar su código de error. Esa sería una situación bastante triste. :)

Por lo tanto, la herramienta más útil de las dos herramientas para hacer frente a esto es TestCase.assertFailure. Este es un ayudante para las pruebas que desea devolver un diferido que va a disparar con un fracaso:

def test_returnit(self): 
    d = defer.fail(ValueError("6 is a bad value")) 
    return self.assertFailure(d, ValueError) 

Esta prueba pasará porque d hace fuego con un fracaso envolver una ValueError. Si d ha disparado con un resultado exitoso o con un error al envolver algún otro tipo de excepción, entonces la prueba aún fallará.

A continuación, hay TestCase.flushLoggedErrors. Esto es para cuando prueba una API que es supuesta para registrar un error. Después de todo, a veces desea informar a un administrador que hay un problema.

def test_logit(self): 
    defer.fail(ValueError("6 is a bad value")) 
    gc.collect() 
    self.assertEquals(self.flushLoggedErrors(ValueError), 1) 

Esto le permite inspeccionar las fallas que se registraron para asegurarse de que su código de registro funciona correctamente. También le dice a la prueba que no se preocupe por las cosas que usted encendió, por lo que ya no harán que la prueba falle. (La llamada gc.collect() está ahí porque el error no se registra hasta que se recopile la basura diferida. En CPython, será basura recolectada inmediatamente debido al comportamiento del recuento de GC de referencia. Sin embargo, en Jython o PyPy o cualquier otro tiempo de ejecución de Python sin contar referencias, no puede confiar en eso.)

Además, dado que la recolección de basura puede ocurrir prácticamente en cualquier momento, a veces puede encontrar que una de sus pruebas falla porque se registra un error por un aplazamiento creado por una prueba anterior siendo la basura recolectada durante la ejecución de la prueba posterior. Esto casi siempre significa que su código de manejo de errores está incompleto de alguna manera - le falta un error, o no logró encadenar dos Deferreds en algún lugar, o está dejando que su método de prueba termine antes de que la tarea que comenzó realmente termine - pero la forma en que se informa el error a veces hace que sea difícil rastrear el código ofensivo. La opción Trial --force-gc puede ayudar con esto. Hace que la prueba invoque al recolector de basura entre cada método de prueba. Esto ralentizará significativamente sus pruebas, pero debería provocar que el error se registre contra la prueba que en realidad lo está desencadenando, no una prueba posterior arbitraria.