/usr/lib/python3/dist-packages/testtools/twistedsupport/_deferred.py is in python3-testtools 2.3.0-3ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | # Copyright (c) testtools developers. See LICENSE for details.
"""Utilities for Deferreds."""
from functools import partial
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,
)
|