Comment les erreurs Twisted Deferred sans erreurs peuvent-elles être testées avec un essai?

Question

J'ai un code Twisted qui crée plusieurs chaînes de différés. Certains d'entre eux peuvent échouer sans avoir un errback qui les remet sur la chaîne de rappel. Je n'ai pas été en mesure d'écrire un test unitaire pour ce code - l'échec différé provoque l'échec du test une fois le code de test terminé. Comment puis-je écrire un test d'unité de passage pour ce code? Est-il prévu que chaque différé qui pourrait échouer en fonctionnement normal devrait avoir un errback à la fin de la chaîne qui le remet sur la chaîne de rappel?

La même chose se produit en cas d'échec de différé dans une liste différée, sauf si je crée la liste différée avec consumeErrors. C'est le cas même lorsque DeferredList est créé avec fireOnOneErrback et reçoit un errback qui le remet sur la chaîne de rappel. Y a-t-il des implications pour consumeErrors en plus de supprimer les échecs de test et la journalisation des erreurs? Est-ce que tous les différés qui peuvent échouer sans erreur peuvent être mis dans une liste différée?

tests Exemple de code exemple:

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

Il y a deux choses importantes au sujet de procès liés à cette question. Tout d'abord, une méthode de test ne passera pas si un échec est enregistré alors qu'il est en cours d'exécution. Les reports qui sont collectés avec un résultat d'échec provoquent l'échec de l'enregistrement. Deuxièmement, une méthode de test qui renvoie un différé ne passera pas si le différé se déclenche avec un échec.

Cela signifie que ni ces tests peuvent passer:

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

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

Ceci est important car le premier cas, le cas d'une poubelle étant différée collectés avec un résultat d'échec, signifie qu'une erreur est survenue que personne ne manipulé. C'est en quelque sorte similaire à la façon dont Python signalera une trace de pile si une exception atteint le niveau supérieur de votre programme.

De même, le deuxième cas est un filet de sécurité fourni par essai. Si une méthode de test synchrone déclenche une exception, le test ne passe pas. Donc, si une méthode de test d'essai renvoie un différé, le différé doit avoir un résultat de réussite pour que le test réussisse.

Il existe cependant des outils pour traiter chacun de ces cas. Après tout, si vous ne pouviez pas avoir un test de passage pour une API qui renvoyait un différé ayant parfois déclenché un échec, vous ne pourriez jamais tester votre code d'erreur. Ce serait une situation plutôt triste. :)

Ainsi, le plus utile des deux outils pour traiter ceci est TestCase.assertFailure. Ceci est une aide pour les tests qui veulent retourner un Reporté qui va au feu avec un échec:

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

Ce test passera parce que d ne se déclenche avec un échec enroulant un ValueError. Si d avait déclenché avec un résultat de réussite ou avec un échec d'emballage d'un autre type d'exception, le test échouait toujours.

Ensuite, il y a TestCase.flushLoggedErrors. C'est pour lorsque vous testez une API supposé pour enregistrer une erreur. Après tout, vous voulez parfois informer un administrateur qu'il y a un problème.Cela vous permet d'inspecter les échecs consignés pour vous assurer que votre code de journalisation fonctionne correctement. Il indique également au procès de ne pas s'inquiéter des choses que vous avez vidées, ainsi ils ne feront plus échouer le test. (L'appel gc.collect() est là parce que l'erreur n'est pas consignée jusqu'à ce que le différé soit récupéré.) Sur CPython, il sera collecté immédiatement en raison du comportement du GC de comptage de références, mais sur Jython ou PyPy ou toute autre exécution de Python. sans compter la référence, vous ne pouvez pas compter dessus.)

De plus, comme le garbage collection peut arriver à peu près à tout moment, vous pourriez parfois trouver qu'un de vos tests échoue parce qu'une erreur est enregistrée par un différé créé par un précédemment test étant garbage collecté lors de l'exécution du test plus tard. Cela signifie presque toujours que votre code de gestion des erreurs est incomplet d'une certaine façon - il vous manque un errback, ou vous avez échoué à enchaîner deux Deferred quelque part, ou vous laissez votre méthode de test finir avant que la tâche commence réellement - mais la manière dont l'erreur est signalée rend parfois difficile le suivi du code incriminé. L'option --force-gc de l'essai peut aider avec ceci. Il provoque le procès d'appeler le garbage collector entre chaque méthode de test. Cela ralentira vos tests de manière significative, mais cela devrait entraîner l'enregistrement de l'erreur sur le test qui le déclenche réellement, et non un test ultérieur arbitraire.