Merge pull request #691 from twisted/9006-coro-result-assertions

successResultOf, failureResultOf, assertNoResult should accept coroutines
This commit is contained in:
Wilfredo Sánchez Vega 2019-07-22 11:54:59 -07:00 committed by GitHub
commit c2fbd43161
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 518 additions and 73 deletions

View File

@ -10,6 +10,7 @@ from __future__ import absolute_import, division
from twisted.python.compat import _PY35PLUS, _PY3, execfile
from twisted.python.filepath import FilePath
from twisted.trial import unittest
if _PY35PLUS:
@ -19,18 +20,16 @@ if _PY35PLUS:
execfile(_path.path, _g)
AwaitTests = _g["AwaitTests"]
else:
from twisted.trial.unittest import TestCase
class AwaitTests(TestCase):
class AwaitTests(unittest.SynchronousTestCase):
"""
A dummy class to show that this test file was discovered but the tests
are unable to be ran in this version of Python.
are unable to be run in this version of Python.
"""
skip = "async/await is not available before Python 3.5"
def test_notAvailable(self):
"""
A skipped test to show that this was not ran because the Python is
A skipped test to show that this was not run because the Python is
too old.
"""
@ -42,18 +41,16 @@ if _PY3:
execfile(_path.path, _g)
YieldFromTests = _g["YieldFromTests"]
else:
from twisted.trial.unittest import TestCase
class YieldFromTests(TestCase):
class YieldFromTests(unittest.SynchronousTestCase):
"""
A dummy class to show that this test file was discovered but the tests
are unable to be ran in this version of Python.
are unable to be run in this version of Python.
"""
skip = "yield from is not available before Python 3"
def test_notAvailable(self):
"""
A skipped test to show that this was not ran because the Python is
A skipped test to show that this was not run because the Python is
too old.
"""

View File

@ -0,0 +1,2 @@
twisted.trial.successResultOf, twisted.trial.failureResultOf, and
twisted.trial.assertNoResult accept coroutines as well as Deferreds.

View File

@ -150,6 +150,8 @@ class NewStyleOnly(object):
module = namedAny(self.module)
except ImportError as e:
raise unittest.SkipTest("Not importable: {}".format(e))
except SyntaxError as e:
raise unittest.SkipTest("Invalid syntax: {}".format(e))
oldStyleClasses = []
@ -174,6 +176,47 @@ class NewStyleOnly(object):
class NewStyleOnlyTests(unittest.TestCase):
"""
Tests for L{NewStyleOnly}, because test turtles all the way down.
"""
def test_importError(self):
"""
L{NewStyleOnly.test_newStyleClassesOnly} raises L{unittest.SkipTest}
if in the module is not importable.
"""
self._verifySkipOnImportError(ImportError)
def test_syntaxError(self):
"""
L{NewStyleOnly.test_newStyleClassesOnly} raises L{unittest.SkipTest}
if in the module is not parseable.
"""
self._verifySkipOnImportError(SyntaxError)
def _verifySkipOnImportError(self, exceptionType):
"""
L{NewStyleOnly.test_newStyleClassesOnly} raises L{unittest.SkipTest}
if in the module is not importable due to the given exception type.
@param exceptionType: The exception raised by namedAny.
@type exceptionType: exception class
"""
def namedAny(*args, **kwargs):
raise exceptionType("Oh no")
import twisted.test.test_nooldstyle as me
self.patch(me, "namedAny", namedAny)
nso = NewStyleOnly()
nso.module = "foo"
self.assertRaises(unittest.SkipTest, nso.test_newStyleClassesOnly)
def _buildTestClasses(_locals):
"""
Build the test classes that use L{NewStyleOnly}, one class per module.

View File

@ -18,7 +18,9 @@ from twisted.python import failure, log, monkey
from twisted.python.reflect import fullyQualifiedName
from twisted.python.util import runWithWarningsSuppressed
from twisted.python.deprecate import (
getDeprecationWarningString, warnAboutFunction)
getDeprecationWarningString, warnAboutFunction
)
from twisted.internet.defer import ensureDeferred
from twisted.trial import itrial, util
@ -686,19 +688,26 @@ class _Assertions(pyunit.TestCase, object):
@return: The result of C{deferred}.
"""
deferred = ensureDeferred(deferred)
result = []
deferred.addBoth(result.append)
if not result:
self.fail(
"Success result expected on %r, found no result instead" % (
deferred,))
elif isinstance(result[0], failure.Failure):
"Success result expected on {!r}, found no result instead"
.format(deferred)
)
result = result[0]
if isinstance(result, failure.Failure):
self.fail(
"Success result expected on %r, "
"found failure result instead:\n%s" % (
deferred, result[0].getTraceback()))
else:
return result[0]
"Success result expected on {!r}, "
"found failure result instead:\n{}"
.format(deferred, result.getTraceback())
)
return result
def failureResultOf(self, deferred, *expectedExceptionTypes):
@ -726,29 +735,44 @@ class _Assertions(pyunit.TestCase, object):
@return: The failure result of C{deferred}.
@rtype: L{failure.Failure}
"""
deferred = ensureDeferred(deferred)
result = []
deferred.addBoth(result.append)
if not result:
self.fail(
"Failure result expected on %r, found no result instead" % (
deferred,))
elif not isinstance(result[0], failure.Failure):
"Failure result expected on {!r}, found no result instead"
.format(deferred)
)
result = result[0]
if not isinstance(result, failure.Failure):
self.fail(
"Failure result expected on %r, "
"found success result (%r) instead" % (deferred, result[0]))
elif (expectedExceptionTypes and
not result[0].check(*expectedExceptionTypes)):
"Failure result expected on {!r}, "
"found success result ({!r}) instead"
.format(deferred, result)
)
if (
expectedExceptionTypes and
not result.check(*expectedExceptionTypes)
):
expectedString = " or ".join([
'.'.join((t.__module__, t.__name__)) for t in
expectedExceptionTypes])
".".join((t.__module__, t.__name__))
for t in expectedExceptionTypes
])
self.fail(
"Failure of type (%s) expected on %r, "
"found type %r instead: %s" % (
expectedString, deferred, result[0].type,
result[0].getTraceback()))
else:
return result[0]
"Failure of type ({}) expected on {!r}, "
"found type {!r} instead: {}"
.format(
expectedString, deferred, result.type,
result.getTraceback()
)
)
return result
def assertNoResult(self, deferred):
@ -770,18 +794,23 @@ class _Assertions(pyunit.TestCase, object):
@raise SynchronousTestCase.failureException: If the
L{Deferred<twisted.internet.defer.Deferred>} has a result.
"""
deferred = ensureDeferred(deferred)
result = []
def cb(res):
result.append(res)
return res
deferred.addBoth(cb)
if result:
# If there is already a failure, the self.fail below will
# report it, so swallow it in the deferred
deferred.addErrback(lambda _: None)
self.fail(
"No result expected on %r, found %r instead" % (
deferred, result[0]))
"No result expected on {!r}, found {!r} instead"
.format(deferred, result[0])
)
def assertRegex(self, text, regex, msg=None):

View File

@ -0,0 +1,308 @@
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
from twisted.python.failure import Failure
from twisted.trial import unittest
from twisted.internet.defer import Deferred, fail
class ResultOfCoroutineAssertionsTests(unittest.SynchronousTestCase):
"""
Tests for L{SynchronousTestCase.successResultOf},
L{SynchronousTestCase.failureResultOf}, and
L{SynchronousTestCase.assertNoResult} when given a coroutine.
"""
result = object()
exception = Exception("Bad times")
failure = Failure(exception)
async def successResult(self):
return self.result
async def noCurrentResult(self):
await Deferred()
async def raisesException(self):
raise self.exception
def test_withoutResult(self):
"""
L{SynchronousTestCase.successResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
with no current result.
"""
self.assertRaises(
self.failureException, self.successResultOf, self.noCurrentResult()
)
def test_successResultOfWithException(self):
"""
L{SynchronousTestCase.successResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
that raises an exception.
"""
self.assertRaises(
self.failureException, self.successResultOf, self.raisesException()
)
def test_successResultOfWithFailureHasTraceback(self):
"""
L{SynchronousTestCase.successResultOf} raises a
L{SynchronousTestCase.failureException} that has the original failure
traceback when called with a coroutine with a failure result.
"""
try:
self.successResultOf(self.raisesException())
except self.failureException as e:
self.assertIn(self.failure.getTraceback(), str(e))
test_successResultOfWithFailureHasTraceback.todo = (
"Tracebacks aren't preserved by exceptions later wrapped in Failures"
)
def test_failureResultOfWithoutResult(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
with no current result.
"""
self.assertRaises(
self.failureException, self.failureResultOf, self.noCurrentResult()
)
def test_failureResultOfWithSuccess(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
with a success result.
"""
self.assertRaises(
self.failureException, self.failureResultOf, self.successResult()
)
def test_failureResultOfWithWrongFailure(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
that raises an exception that was not expected.
"""
self.assertRaises(
self.failureException,
self.failureResultOf, self.raisesException(), KeyError
)
def test_failureResultOfWithWrongExceptionOneExpectedException(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
that raises an exception with a failure type that was not expected, and
the L{SynchronousTestCase.failureException} message contains the
expected exception type.
"""
try:
self.failureResultOf(self.raisesException(), KeyError)
except self.failureException as e:
self.assertIn(
"Failure of type ({0}.{1}) expected on".format(
KeyError.__module__, KeyError.__name__
),
str(e)
)
def test_failureResultOfWithWrongExceptionOneExpectedExceptionHasTB(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
that raises an exception with a failure type that was not expected, and
the L{SynchronousTestCase.failureException} message contains the
original exception traceback.
"""
try:
self.failureResultOf(self.raisesException(), KeyError)
except self.failureException as e:
self.assertIn(self.failure.getTraceback(), str(e))
test_failureResultOfWithWrongExceptionOneExpectedExceptionHasTB.todo = (
"Tracebacks aren't preserved by exceptions later wrapped in Failures"
)
def test_failureResultOfWithWrongExceptionMultiExpectedExceptions(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
that raises an exception of a type that was not expected, and the
L{SynchronousTestCase.failureException} message contains expected
exception types in the error message.
"""
try:
self.failureResultOf(self.raisesException(), KeyError, IOError)
except self.failureException as e:
self.assertIn(
"Failure of type ({0}.{1} or {2}.{3}) expected on".format(
KeyError.__module__, KeyError.__name__,
IOError.__module__, IOError.__name__,
),
str(e)
)
def test_failureResultOfWithWrongExceptionMultiExpectedExceptionsHasTB(
self
):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a coroutine
that raises an exception of a type that was not expected, and the
L{SynchronousTestCase.failureException} message contains the original
exception traceback in the error message.
"""
try:
self.failureResultOf(self.raisesException(), KeyError, IOError)
except self.failureException as e:
self.assertIn(self.failure.getTraceback(), str(e))
test_failureResultOfWithWrongExceptionMultiExpectedExceptionsHasTB.todo = (
"Tracebacks aren't preserved by exceptions later wrapped in Failures"
)
def test_successResultOfWithSuccessResult(self):
"""
When passed a coroutine which currently has a result (ie, if converted
into a L{Deferred}, L{Deferred.addCallback} would cause the added
callback to be called before C{addCallback} returns),
L{SynchronousTestCase.successResultOf} returns that result.
"""
self.assertIdentical(
self.result, self.successResultOf(self.successResult())
)
def test_failureResultOfWithExpectedException(self):
"""
When passed a coroutine which currently has an exception result (ie, if
converted into a L{Deferred}, L{Deferred.addErrback} would cause the
added errback to be called before C{addErrback} returns),
L{SynchronousTestCase.failureResultOf} returns a L{Failure} containing
that exception, if the exception type is expected.
"""
self.assertEqual(
self.failure.value,
self.failureResultOf(
self.raisesException(), self.failure.type, KeyError
).value
)
def test_failureResultOfWithException(self):
"""
When passed a coroutine which currently has an exception result (ie, if
converted into a L{Deferred}, L{Deferred.addErrback} would cause the
added errback to be called before C{addErrback} returns),
L{SynchronousTestCase.failureResultOf} returns returns a L{Failure}
containing that exception.
"""
self.assertEqual(
self.failure.value,
self.failureResultOf(self.raisesException()).value
)
def test_assertNoResultSuccess(self):
"""
When passed a coroutine which currently has a success result (see
L{test_withSuccessResult}), L{SynchronousTestCase.assertNoResult}
raises L{SynchronousTestCase.failureException}.
"""
self.assertRaises(
self.failureException, self.assertNoResult, self.successResult()
)
def test_assertNoResultFailure(self):
"""
When passed a coroutine which currently has an exception result (see
L{test_withFailureResult}), L{SynchronousTestCase.assertNoResult}
raises L{SynchronousTestCase.failureException}.
"""
self.assertRaises(
self.failureException, self.assertNoResult, self.raisesException()
)
def test_assertNoResult(self):
"""
When passed a coroutine with no current result,
L{SynchronousTestCase.assertNoResult} does not raise an exception.
"""
self.assertNoResult(self.noCurrentResult())
def test_assertNoResultPropagatesSuccess(self):
"""
When passed a coroutine awaiting a L{Deferred} with no current result,
which is then fired with a success result,
L{SynchronousTestCase.assertNoResult} doesn't modify the result of the
L{Deferred}.
"""
d = Deferred()
async def noCurrentResult():
return await d
c = noCurrentResult()
self.assertNoResult(d)
d.callback(self.result)
self.assertEqual(self.result, self.successResultOf(c))
def test_assertNoResultPropagatesLaterFailure(self):
"""
When passed a coroutine awaiting a L{Deferred} with no current result,
which is then fired with a L{Failure} result,
L{SynchronousTestCase.assertNoResult} doesn't modify the result of the
L{Deferred}.
"""
f = Failure(self.exception)
d = Deferred()
async def noCurrentResult():
return await d
c = noCurrentResult()
self.assertNoResult(d)
d.errback(f)
self.assertEqual(f.value, self.failureResultOf(c).value)
def test_assertNoResultSwallowsImmediateFailure(self):
"""
When passed a L{Deferred} which currently has a L{Failure} result,
L{SynchronousTestCase.assertNoResult} changes the result of the
L{Deferred} to a success.
"""
d = fail(self.failure)
async def raisesException():
return await d
c = raisesException()
try:
self.assertNoResult(d)
except self.failureException:
pass
self.assertEqual(None, self.successResultOf(c))

View File

@ -14,9 +14,11 @@ demonstrated to work earlier in the file are used by those later in the file
from __future__ import division, absolute_import
import sys
import warnings
import unittest as pyunit
from twisted.python.filepath import FilePath
from twisted.python.util import FancyEqMixin
from twisted.python.reflect import (
prefixedMethods, accumulateMethods, fullyQualifiedName)
@ -957,17 +959,20 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
L{SynchronousTestCase.failureResultOf}, and
L{SynchronousTestCase.assertNoResult}.
"""
result = object()
failure = Failure(Exception("Bad times"))
def test_withoutSuccessResult(self):
def test_withoutResult(self):
"""
L{SynchronousTestCase.successResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
with no current result.
"""
self.assertRaises(
self.failureException, self.successResultOf, Deferred())
self.failureException, self.successResultOf, Deferred()
)
def test_successResultOfWithFailure(self):
@ -977,7 +982,8 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
with a failure result.
"""
self.assertRaises(
self.failureException, self.successResultOf, fail(self.failure))
self.failureException, self.successResultOf, fail(self.failure)
)
def test_successResultOfWithFailureHasTraceback(self):
@ -992,14 +998,15 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
self.assertIn(self.failure.getTraceback(), str(e))
def test_withoutFailureResult(self):
def test_failureResultOfWithoutResult(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
with no current result.
"""
self.assertRaises(
self.failureException, self.failureResultOf, Deferred())
self.failureException, self.failureResultOf, Deferred()
)
def test_failureResultOfWithSuccess(self):
@ -1009,58 +1016,90 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
with a success result.
"""
self.assertRaises(
self.failureException, self.failureResultOf, succeed(self.result))
self.failureException, self.failureResultOf, succeed(self.result)
)
def test_failureResultOfWithWrongFailure(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
with a failure type that was not expected.
that fails with an exception type that was not expected.
"""
self.assertRaises(
self.failureException, self.failureResultOf, fail(self.failure),
KeyError)
self.failureException,
self.failureResultOf, fail(self.failure), KeyError
)
def test_failureResultOfWithWrongFailureOneExpectedFailure(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
with a failure type that was not expected, and the
that fails with an exception type that was not expected, and the
L{SynchronousTestCase.failureException} message contains the expected
exception type.
"""
try:
self.failureResultOf(fail(self.failure), KeyError)
except self.failureException as e:
self.assertIn(
"Failure of type ({0}.{1}) expected on".format(
KeyError.__module__, KeyError.__name__
),
str(e)
)
def test_failureResultOfWithWrongFailureOneExpectedFailureHasTB(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
that fails with an exception type that was not expected, and the
L{SynchronousTestCase.failureException} message contains the original
failure traceback as well as the expected failure type
failure traceback.
"""
try:
self.failureResultOf(fail(self.failure), KeyError)
except self.failureException as e:
self.assertIn(self.failure.getTraceback(), str(e))
self.assertIn(
"Failure of type ({0}.{1}) expected on".format(
KeyError.__module__, KeyError.__name__),
str(e))
def test_failureResultOfWithWrongFailureMultiExpectedFailure(self):
def test_failureResultOfWithWrongFailureMultiExpectedFailures(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
with a failure type that was not expected, and the
with an exception type that was not expected, and the
L{SynchronousTestCase.failureException} message contains the expected
exception types in the error message.
"""
try:
self.failureResultOf(fail(self.failure), KeyError, IOError)
except self.failureException as e:
self.assertIn(
"Failure of type ({0}.{1} or {2}.{3}) expected on".format(
KeyError.__module__, KeyError.__name__,
IOError.__module__, IOError.__name__,
),
str(e)
)
def test_failureResultOfWithWrongFailureMultiExpectedFailuresHasTB(self):
"""
L{SynchronousTestCase.failureResultOf} raises
L{SynchronousTestCase.failureException} when called with a L{Deferred}
with an exception type that was not expected, and the
L{SynchronousTestCase.failureException} message contains the original
failure traceback as well as the expected failure types in the error
message
failure traceback in the error message.
"""
try:
self.failureResultOf(fail(self.failure), KeyError, IOError)
except self.failureException as e:
self.assertIn(self.failure.getTraceback(), str(e))
self.assertIn(
"Failure of type ({0}.{1} or {2}.{3}) expected on".format(
KeyError.__module__, KeyError.__name__,
IOError.__module__, IOError.__name__),
str(e))
def test_withSuccessResult(self):
def test_successResultOfWithSuccessResult(self):
"""
When passed a L{Deferred} which currently has a result (ie,
L{Deferred.addCallback} would cause the added callback to be called
@ -1071,20 +1110,22 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
self.result, self.successResultOf(succeed(self.result)))
def test_withExpectedFailureResult(self):
def test_failureResultOfWithExpectedFailureResult(self):
"""
When passed a L{Deferred} which currently has a L{Failure} result (ie,
L{Deferred.addErrback} would cause the added errback to be called
before C{addErrback} returns), L{SynchronousTestCase.failureResultOf}
returns that L{Failure} if that L{Failure}'s type is expected.
returns that L{Failure} if its contained exception type is expected.
"""
self.assertIdentical(
self.failure,
self.failureResultOf(fail(self.failure), self.failure.type,
KeyError))
self.failureResultOf(
fail(self.failure), self.failure.type, KeyError
)
)
def test_withFailureResult(self):
def test_failureResultOfWithFailureResult(self):
"""
When passed a L{Deferred} which currently has a L{Failure} result
(ie, L{Deferred.addErrback} would cause the added errback to be called
@ -1092,32 +1133,36 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
returns that L{Failure}.
"""
self.assertIdentical(
self.failure, self.failureResultOf(fail(self.failure)))
self.failure, self.failureResultOf(fail(self.failure))
)
def test_assertNoResultSuccess(self):
"""
When passed a L{Deferred} which currently has a success result (see
L{test_withSuccessResult}), L{SynchronousTestCase.assertNoResult} raises
L{SynchronousTestCase.failureException}.
L{test_withSuccessResult}), L{SynchronousTestCase.assertNoResult}
raises L{SynchronousTestCase.failureException}.
"""
self.assertRaises(
self.failureException, self.assertNoResult, succeed(self.result))
self.failureException, self.assertNoResult, succeed(self.result)
)
def test_assertNoResultFailure(self):
"""
When passed a L{Deferred} which currently has a failure result (see
L{test_withFailureResult}), L{SynchronousTestCase.assertNoResult} raises
L{SynchronousTestCase.failureException}.
L{test_withFailureResult}), L{SynchronousTestCase.assertNoResult}
raises L{SynchronousTestCase.failureException}.
"""
self.assertRaises(
self.failureException, self.assertNoResult, fail(self.failure))
self.failureException, self.assertNoResult, fail(self.failure)
)
def test_assertNoResult(self):
"""
When passed a L{Deferred} with no current result,
L{SynchronousTestCase.assertNoResult} does not raise an exception.
"""
self.assertNoResult(Deferred())
@ -1161,6 +1206,27 @@ class ResultOfAssertionsTests(unittest.SynchronousTestCase):
if sys.version_info >= (3, 5):
_p = FilePath(__file__).parent().child("_assertiontests.py.3only")
with _p.open() as _f:
exec(_f.read())
del _p, _f
else:
class ResultOfCoroutineAssertionsTests(unittest.SynchronousTestCase):
"""
A dummy class to show that this test file was discovered but the tests
are unable to be run in this version of Python.
"""
skip = "async/await is not available before Python 3.5"
def test_notAvailable(self):
"""
A skipped test to show that this was not run because the Python is
too old.
"""
class AssertionNamesTests(unittest.SynchronousTestCase):
"""
Tests for consistency of naming within TestCase assertion methods