In che modo è possibile testare gli errori deviati e deviati senza errori?

Question

Ho qualche codice Twisted che crea più catene di differite. Alcuni di questi potrebbero fallire senza avere una battuta d'arresto che li riporta alla catena di callback. Non sono stato in grado di scrivere un test unitario per questo codice: il mancato rinvio fa sì che il test non riesca dopo che il codice del test è stato completato. Come posso scrivere un test di unità passeggero per questo codice? È previsto che ogni Deferred che potrebbe fallire nel normale funzionamento dovrebbe avere una battuta d'arresto alla fine della catena che la reinserisce nella catena di callback?

La stessa cosa accade quando c'è un Deferred fallito in una DeferredList, a meno che non crei la DeferredList con ConsumeErrors. Questo è il caso anche quando DeferredList viene creato con fireOnOneErrback e viene restituito un errore che lo rimette nella catena di callback. Ci sono implicazioni per ConsumeErrors oltre a sopprimere i fallimenti dei test e la registrazione degli errori? Ogni Deferred che può fallire senza un errback deve essere inserito in DeferredList?

test esempio di codice di esempio:

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

Ci sono due cose importanti su di prova relativi a questa domanda.

In primo luogo, un metodo di prova non passa se viene registrato un errore mentre è in esecuzione. I rinvii che sono garbage collection con un risultato Failure causano il Failure da registrare.

In secondo luogo, un metodo di prova che restituisce un rinvio non passerà se il differito si attiva con un errore.

Questo significa che nessuno di questi test può passare:

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

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

Questo è importante perché il primo caso, il caso di un essere garbage anticipate raccolto con un risultato Fallimento, significa che v'è un errore che nessuno maneggiato. È simile al modo in cui Python segnalerà una traccia dello stack se un'eccezione raggiunge il livello più alto del tuo programma.

Allo stesso modo, il secondo caso è una rete di sicurezza fornita da prova. Se un metodo di prova sincrono solleva un'eccezione, il test non passa. Quindi, se un metodo di test di prova restituisce un rinvio, il differito deve avere un risultato positivo per il test da superare.

Esistono tuttavia strumenti per gestire ciascuno di questi casi. Dopotutto, se non si poteva avere un test di passaggio per un'API che restituiva a volte un rinvio infornato con un errore, allora non si poteva mai testare il codice di errore. Sarebbe una situazione piuttosto triste. :)

Quindi, il più utile dei due strumenti per gestire questo è TestCase.assertFailure. Si tratta di un aiuto per le prove che vogliono restituire una differita che sta andando a fuoco con un fallimento:

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

Questo test passerà perché d fa fuoco con un guasto avvolgendo un ValueError. Se d è stato attivato con un risultato positivo o con un errore che ha involucro di un altro tipo di eccezione, il test fallirebbe comunque.

Successivamente, c'è TestCase.flushLoggedErrors. Questo è per quando stai testando un'API che è supposto per registrare un errore. Dopotutto, a volte vuoi informare un amministratore che c'è un problema.

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

Ciò consente di ispezionare gli errori registrati per assicurarsi che il codice di registrazione funzioni correttamente. Indica inoltre al processo di non preoccuparsi delle cose che hai scaricato, quindi non causeranno più il fallimento del test. (La chiamata gc.collect() è presente perché l'errore non viene registrato fino a quando il Deferred non viene raccolto. Su CPython, verrà immediatamente raccolto a causa del comportamento di conteggio dei riferimenti GC. Tuttavia, su Jython o PyPy o qualsiasi altro runtime di Python senza contare i riferimenti, non ci si può fidare di questo.)

Inoltre, poiché la raccolta dei dati inutili può accadere praticamente in qualsiasi momento, a volte è possibile rilevare che uno dei test non riesce perché un errore viene registrato da un differito creato da un test precedente viene sottoposto a garbage collection durante l'esecuzione del test successivo. Questo significa quasi sempre che il codice di gestione degli errori è incompleto in qualche modo: ti manca un errore o non riesci a mettere insieme due Deferred da qualche parte, o stai lasciando che il tuo metodo di prova finisca prima che l'attività che ha effettivamente terminato - ma il modo in cui viene segnalato l'errore a volte rende difficile rintracciare il codice incriminato. L'opzione di prova --force-gc può essere d'aiuto. Fa sì che il processo invochi il garbage collector tra ogni metodo di prova. Ciò rallenterà significativamente i test, ma dovrebbe causare l'accesso dell'errore al test che lo sta effettivamente attivando, non un test successivo arbitrario.