[Python-modules-commits] [txfixtures] 02/05: backport-testtools-twistedsupport

Free Ekanayaka freee at moszumanska.debian.org
Thu Feb 2 09:42:20 UTC 2017


This is an automated email from the git hooks/post-receive script.

freee pushed a commit to branch patch-queue/master
in repository txfixtures.

commit 15c1270ab23e2c31026b71abe0c73e6dcfecb893
Author: Free Ekanayaka <freee at debian.org>
Date:   Thu Jan 5 10:20:25 2017 +0000

    backport-testtools-twistedsupport
    
    Include a backport of testtools.twistedsupport, needed to run
    txfixtures unit tests.
    
    Gbp-Pq: Name 0001-backport-testtools-twistedsupport.patch
---
 txfixtures/_testtools.py | 290 +++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 290 insertions(+)

diff --git a/txfixtures/_testtools.py b/txfixtures/_testtools.py
new file mode 100644
index 0000000..654373e
--- /dev/null
+++ b/txfixtures/_testtools.py
@@ -0,0 +1,290 @@
+# Copyright (c) testtools developers. See LICENSE for details.
+
+"""Matchers that operate on synchronous Deferreds.
+
+A "synchronous" Deferred is one that does not need the reactor or any other
+asynchronous process in order to fire.
+
+Normal application code can't know when a Deferred is going to fire, because
+that is generally left up to the reactor. Unit tests can (and should!) provide
+fake reactors, or don't use the reactor at all, so that Deferreds fire
+synchronously.
+
+These matchers allow you to make assertions about when and how Deferreds fire,
+and about what values they fire with.
+"""
+from functools import partial
+
+from testtools.compat import _u
+from testtools.matchers import Mismatch
+
+from testtools.content import TracebackContent
+
+
+class DeferredNotFired(Exception):
+    """Raised when we extract a result from a Deferred that's not fired yet."""
+
+    def __init__(self, deferred):
+        msg = "%r has not fired yet." % (deferred,)
+        super(DeferredNotFired, self).__init__(msg)
+
+
+def extract_result(deferred):
+    """Extract the result from a fired deferred.
+
+    It can happen that you have an API that returns Deferreds for
+    compatibility with Twisted code, but is in fact synchronous, i.e. the
+    Deferreds it returns have always fired by the time it returns.  In this
+    case, you can use this function to convert the result back into the usual
+    form for a synchronous API, i.e. the result itself or a raised exception.
+
+    As a rule, this function should not be used when operating with
+    asynchronous Deferreds (i.e. for normal use of Deferreds in application
+    code). In those cases, it is better to add callbacks and errbacks as
+    needed.
+    """
+    failures = []
+    successes = []
+    deferred.addCallbacks(successes.append, failures.append)
+    if len(failures) == 1:
+        failures[0].raiseException()
+    elif len(successes) == 1:
+        return successes[0]
+    else:
+        raise DeferredNotFired(deferred)
+
+
+class ImpossibleDeferredError(Exception):
+    """Raised if a Deferred somehow triggers both a success and a failure."""
+
+    def __init__(self, deferred, successes, failures):
+        msg = ('Impossible condition on %r, got both success (%r) and '
+               'failure (%r)')
+        super(ImpossibleDeferredError, self).__init__(
+            msg % (deferred, successes, failures))
+
+
+def on_deferred_result(deferred, on_success, on_failure, on_no_result):
+    """Handle the result of a synchronous ``Deferred``.
+
+    If ``deferred`` has fire successfully, call ``on_success``.
+    If ``deferred`` has failed, call ``on_failure``.
+    If ``deferred`` has not yet fired, call ``on_no_result``.
+
+    The value of ``deferred`` will be preserved, so that other callbacks and
+    errbacks can be added to ``deferred``.
+
+    :param Deferred[A] deferred: A synchronous Deferred.
+    :param Callable[[Deferred[A], A], T] on_success: Called if the Deferred
+        fires successfully.
+    :param Callable[[Deferred[A], Failure], T] on_failure: Called if the
+        Deferred fires unsuccessfully.
+    :param Callable[[Deferred[A]], T] on_no_result: Called if the Deferred has
+        not yet fired.
+
+    :raises ImpossibleDeferredError: If the Deferred somehow
+        triggers both a success and a failure.
+    :raises TypeError: If the Deferred somehow triggers more than one success,
+        or more than one failure.
+
+    :return: Whatever is returned by the triggered callback.
+    :rtype: ``T``
+    """
+    successes = []
+    failures = []
+
+    def capture(value, values):
+        values.append(value)
+        return value
+
+    deferred.addCallbacks(
+        partial(capture, values=successes),
+        partial(capture, values=failures),
+    )
+
+    if successes and failures:
+        raise ImpossibleDeferredError(deferred, successes, failures)
+    elif failures:
+        [failure] = failures
+        return on_failure(deferred, failure)
+    elif successes:
+        [result] = successes
+        return on_success(deferred, result)
+    else:
+        return on_no_result(deferred)
+
+
+def failure_content(failure):
+    """Create a Content object for a Failure.
+
+    :param Failure failure: The failure to create content for.
+    :rtype: ``Content``
+    """
+    return TracebackContent(
+        (failure.type, failure.value, failure.getTracebackObject()),
+        None,
+    )
+
+
+class _NoResult(object):
+    """Matches a Deferred that has not yet fired."""
+
+    @staticmethod
+    def _got_result(deferred, result):
+        return Mismatch(
+            _u('No result expected on %r, found %r instead'
+               % (deferred, result)))
+
+    def match(self, deferred):
+        """Match ``deferred`` if it hasn't fired."""
+        return on_deferred_result(
+            deferred,
+            on_success=self._got_result,
+            on_failure=self._got_result,
+            on_no_result=lambda _: None,
+        )
+
+
+_NO_RESULT = _NoResult()
+
+
+def has_no_result():
+    """Match a Deferred that has not yet fired.
+
+    For example, this will pass::
+
+        assert_that(defer.Deferred(), has_no_result())
+
+    But this will fail:
+
+    >>> assert_that(defer.succeed(None), has_no_result())
+    Traceback (most recent call last):
+      ...
+      File "testtools/assertions.py", line 22, in assert_that
+        raise MismatchError(matchee, matcher, mismatch, verbose)
+    testtools.matchers._impl.MismatchError: No result expected on <Deferred at ... current result: None>, found None instead
+
+    As will this:
+
+    >>> assert_that(defer.fail(RuntimeError('foo')), has_no_result())
+    Traceback (most recent call last):
+      ...
+      File "testtools/assertions.py", line 22, in assert_that
+        raise MismatchError(matchee, matcher, mismatch, verbose)
+    testtools.matchers._impl.MismatchError: No result expected on <Deferred at ... current result: <twisted.python.failure.Failure <type 'exceptions.RuntimeError'>>>, found <twisted.python.failure.Failure <type 'exceptions.RuntimeError'>> instead
+    """
+    return _NO_RESULT
+
+
+class _Succeeded(object):
+    """Matches a Deferred that has fired successfully."""
+
+    def __init__(self, matcher):
+        """Construct a ``_Succeeded`` matcher."""
+        self._matcher = matcher
+
+    @staticmethod
+    def _got_failure(deferred, failure):
+        deferred.addErrback(lambda _: None)
+        return Mismatch(
+            _u('Success result expected on %r, found failure result '
+               'instead: %r' % (deferred, failure)),
+            {'traceback': failure_content(failure)},
+        )
+
+    @staticmethod
+    def _got_no_result(deferred):
+        return Mismatch(
+            _u('Success result expected on %r, found no result '
+               'instead' % (deferred,)))
+
+    def match(self, deferred):
+        """Match against the successful result of ``deferred``."""
+        return on_deferred_result(
+            deferred,
+            on_success=lambda _, value: self._matcher.match(value),
+            on_failure=self._got_failure,
+            on_no_result=self._got_no_result,
+        )
+
+
+def succeeded(matcher):
+    """Match a Deferred that has fired successfully.
+
+    For example::
+
+        fires_with_the_answer = succeeded(Equals(42))
+        deferred = defer.succeed(42)
+        assert_that(deferred, fires_with_the_answer)
+
+    This assertion will pass. However, if ``deferred`` had fired with a
+    different value, or had failed, or had not fired at all, then it would
+    fail.
+
+    Use this instead of
+    :py:meth:`twisted.trial.unittest.SynchronousTestCase.successResultOf`.
+
+    :param matcher: A matcher to match against the result of a
+        :class:`~twisted.internet.defer.Deferred`.
+    :return: A matcher that can be applied to a synchronous
+        :class:`~twisted.internet.defer.Deferred`.
+    """
+    return _Succeeded(matcher)
+
+
+class _Failed(object):
+    """Matches a Deferred that has failed."""
+
+    def __init__(self, matcher):
+        self._matcher = matcher
+
+    def _got_failure(self, deferred, failure):
+        # We have handled the failure, so suppress its output.
+        deferred.addErrback(lambda _: None)
+        return self._matcher.match(failure)
+
+    @staticmethod
+    def _got_success(deferred, success):
+        return Mismatch(
+            _u('Failure result expected on %r, found success '
+               'result (%r) instead' % (deferred, success)))
+
+    @staticmethod
+    def _got_no_result(deferred):
+        return Mismatch(
+            _u('Failure result expected on %r, found no result instead'
+               % (deferred,)))
+
+    def match(self, deferred):
+        return on_deferred_result(
+            deferred,
+            on_success=self._got_success,
+            on_failure=self._got_failure,
+            on_no_result=self._got_no_result,
+        )
+
+
+def failed(matcher):
+    """Match a Deferred that has failed.
+
+    For example::
+
+        error = RuntimeError('foo')
+        fails_at_runtime = failed(
+            AfterPreprocessing(lambda f: f.value, Equals(error)))
+        deferred = defer.fail(error)
+        assert_that(deferred, fails_at_runtime)
+
+    This assertion will pass. However, if ``deferred`` had fired successfully,
+    had failed with a different error, or had not fired at all, then it would
+    fail.
+
+    Use this instead of
+    :py:meth:`twisted.trial.unittest.SynchronousTestCase.failureResultOf`.
+
+    :param matcher: A matcher to match against the result of a failing
+        :class:`~twisted.internet.defer.Deferred`.
+    :return: A matcher that can be applied to a synchronous
+        :class:`~twisted.internet.defer.Deferred`.
+    """
+    return _Failed(matcher)

-- 
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/python-modules/packages/txfixtures.git



More information about the Python-modules-commits mailing list