Как можно проверить ошибки Twisted Deferred без ошибок, используя пробную версию?

Question

У меня есть некоторый витой код, который создает несколько цепочек отложенных.Некоторые из них могут потерпеть неудачу без ошибки, которая возвращает их в цепочку обратных вызовов.Я не смог написать модульный тест для этого кода - сбой Deferred приводит к сбою теста после завершения кода теста.Как я могу написать прохождение модульного теста для этого кода?Ожидается ли, что у каждого Отложенного, который может потерпеть неудачу в нормальной работе, должен быть ошибочный конец в конце цепочки, который помещает его обратно в цепочку обратных вызовов?если только я не создаю DeferredList с помощью receiveErrors.Это имеет место даже в том случае, когда DeferredList создается с помощью fireOnOneErrback и ему предоставляется ошибка, которая возвращает его в цепочку обратных вызовов.Есть ли какие-либо последствия для потребления ошибок, кроме подавления ошибок теста и регистрации ошибок?Должен ли каждый Deferred, который может потерпеть неудачу без ошибки, быть помещенным в DeferredList?

Примеры тестов примера кода:

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

Есть две важные вещи в пробной версии, связанные с этим вопросом.

Во-первых, тестовый метод не пройдет, если во время работы будет зарегистрирован сбой.Отложенные, которые являются мусором, собранным с результатом Отказа, приводят к регистрации Отказа.

Во-вторых, метод тестирования, который возвращает Отложенный, не пройдет, если Отложенный выстрелит с Отказом.что ни один из этих тестов не может пройти:

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

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

Это важно, потому что первый случай, когда Deferred является мусором, собираемым с результатом Failure, означает, что произошла ошибка, которую никто не обрабатывал.Это похоже на то, как Python сообщит о трассировке стека, если исключение достигнет верхнего уровня вашей программы.

Аналогично, второй случай - это сеть безопасности, предоставляемая пробной версией.Если синхронный метод теста вызывает исключение, тест не проходит.Поэтому, если метод пробного теста возвращает Отложенный, Отложенный должен иметь успешный результат для прохождения теста.

Однако существуют инструменты для работы с каждым из этих случаев.В конце концов, если у вас не может быть проходного теста для API, который возвращает Deferred, который иногда запускается с ошибкой, то вы никогда не сможете проверить свой код ошибки.Это была бы довольно печальная ситуация.:)

Итак, более полезным из двух инструментов для решения этой проблемы является TestCase.assertFailure.Это помощник для тестов, которые хотят вернуть Deferred, который собирается сработать с ошибкой:

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

Этот тест пройдет, потому что d сработает с ошибкой, заключающей ValueError.Если бы d сработал с успешным результатом или с ошибкой, заключающей в себе какой-либо другой тип исключения, то тест все равно не прошел бы.

Далее, есть TestCase.flushLoggedErrors.Это для тестирования API, который предполагается для регистрации ошибки.В конце концов, иногда вы действительно хотите сообщить администратору, что есть проблема.

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

Это позволяет вам проверять сбои, которые были зарегистрированы, чтобы убедиться, что ваш код регистрации работает правильно.Он также говорит пробной версии, что не стоит беспокоиться о вещах, которые вы сбросили, так что они больше не приведут к провалу теста.(Вызов gc.collect() происходит потому, что ошибка не регистрируется до тех пор, пока Deferred не будет собран сборщиком мусора. На CPython он будет сразу же собран сборщиком мусора из-за поведения GC подсчета ссылок. Однако на Jython, PyPy или любом другомСреда выполнения Python без подсчета ссылок, вы не можете на это полагаться.)

Кроме того, поскольку сборка мусора может происходить практически в любое время, иногда вы можете обнаружить, что один из ваших тестов завершился неудачей, поскольку ошибка регистрируетсяa Deferred, созданный тестом ранее , является мусором, собранным во время выполнения более позднего теста.Это в значительной степени всегда означает, что ваш код обработки ошибок каким-то образом неполон - у вас отсутствует ошибка, или вы не сумели где-то соединить два Deferreds, или вы даете своему тестовому методу завершиться до того, как задача, которую он запустил, действительно завершится - носпособ сообщения об ошибке иногда затрудняет отслеживание кода, нарушающего работу.Опция Trial --force-gc может помочь с этим.Это вызывает пробный запуск сборщика мусора между каждым методом тестирования.Это значительно замедлит ваши тесты, но должно привести к тому, что ошибка будет регистрироваться в сравнении с тестом, который фактически ее запускает, а не произвольным последующим тестом.