Skip to content

Commit

Permalink
Merge branch 'trunk' into 11679-loopingcall-vs-clock-tests
Browse files Browse the repository at this point in the history
  • Loading branch information
exarkun committed Sep 22, 2022
2 parents 4825f8b + b1a90b4 commit 8ee190c
Show file tree
Hide file tree
Showing 6 changed files with 98 additions and 48 deletions.
8 changes: 3 additions & 5 deletions src/twisted/trial/_dist/workerreporter.py
Expand Up @@ -10,24 +10,22 @@
"""

from types import TracebackType
from typing import Callable, List, Optional, Sequence, Tuple, Type, TypeVar, Union
from typing import Callable, List, Optional, Sequence, Type, TypeVar
from unittest import TestCase as PyUnitTestCase

from attrs import Factory, define
from typing_extensions import Literal, TypeAlias
from typing_extensions import Literal

from twisted.internet.defer import Deferred, maybeDeferred
from twisted.protocols.amp import AMP
from twisted.python.failure import Failure
from twisted.python.reflect import qual
from twisted.trial._dist import managercommands
from twisted.trial.reporter import TestResult
from ..reporter import TrialFailure
from .stream import chunk, stream

T = TypeVar("T")
ExcInfo: TypeAlias = Tuple[Type[BaseException], BaseException, TracebackType]
XUnitFailure = Union[ExcInfo, Tuple[None, None, None]]
TrialFailure = Union[XUnitFailure, Failure]


async def addError(
Expand Down
Empty file.
Empty file.
9 changes: 8 additions & 1 deletion src/twisted/trial/reporter.py
Expand Up @@ -15,10 +15,13 @@
import unittest as pyunit
import warnings
from collections import OrderedDict
from typing import TYPE_CHECKING, List, Tuple
from types import TracebackType
from typing import TYPE_CHECKING, List, Tuple, Type, Union

from zope.interface import implementer

from typing_extensions import TypeAlias

from twisted.python import log, reflect
from twisted.python.components import proxyForInterface
from twisted.python.failure import Failure
Expand All @@ -33,6 +36,10 @@
except ImportError:
TestProtocolClient = None

ExcInfo: TypeAlias = Tuple[Type[BaseException], BaseException, TracebackType]
XUnitFailure = Union[ExcInfo, Tuple[None, None, None]]
TrialFailure = Union[XUnitFailure, Failure]


def _makeTodo(value: str) -> "Todo":
"""
Expand Down
36 changes: 36 additions & 0 deletions src/twisted/trial/test/__init__.py
Expand Up @@ -4,3 +4,39 @@
"""
Unit tests for the Trial unit-testing framework.
"""

from hypothesis import HealthCheck, settings


def _activateHypothesisProfile():
"""
Load a Hypothesis profile appropriate for a Twisted test suite.
"""
deterministic = settings(
# Disable the deadline. It is too hard to guarantee that a particular
# piece of Python code will always run in less than some fixed amount
# of time. Hardware capabilities, the OS scheduler, the Python
# garbage collector, and other factors all combine to make substantial
# outliers possible. Such failures are a distraction from development
# and a hassle on continuous integration environments.
deadline=None,
suppress_health_check=[
# With the same reasoning as above, disable the Hypothesis time
# limit on data generation by example search strategies.
HealthCheck.too_slow,
],
# When a developer is working on one set of changes, or continuous
# integration system is testing them, it is disruptive for Hypothesis
# to discover a bug in pre-existing code. This is just what
# Hypothesis will do by default, by exploring a pseudo-randomly
# different set of examples each time. Such failures are a
# distraction from development and a hassle in continuous integration
# environments.
derandomize=True,
)

settings.register_profile("twisted_trial_test_profile_deterministic", deterministic)
settings.load_profile("twisted_trial_test_profile_deterministic")


_activateHypothesisProfile()
93 changes: 51 additions & 42 deletions src/twisted/trial/test/test_reporter.py
Expand Up @@ -14,15 +14,22 @@
from inspect import getmro
from io import BytesIO, StringIO
from typing import Type
from unittest import TestCase as StdlibTestCase, expectedFailure
from unittest import (
TestCase as StdlibTestCase,
TestSuite as PyUnitTestSuite,
expectedFailure,
)

from twisted.python import log, reflect
from hamcrest import assert_that, equal_to, has_item, has_length

from twisted.python import log
from twisted.python.failure import Failure
from twisted.python.reflect import qual
from twisted.trial import itrial, reporter, runner, unittest, util
from twisted.trial.reporter import UncleanWarningsReporterWrapper, _ExitWrapper
from twisted.trial.test import erroneous, sample
from twisted.trial.unittest import SkipTest, Todo, makeTodo
from .._dist.test.matchers import isFailure, matches_result, similarFrame
from .matchers import after


class BrokenStream:
Expand Down Expand Up @@ -132,14 +139,14 @@ class ErrorReportingTests(StringTest):
def setUp(self):
self.loader = runner.TestLoader()
self.output = StringIO()
self.result = reporter.Reporter(self.output)
self.result: reporter.Reporter = reporter.Reporter(self.output)

def getOutput(self, suite):
result = self.getResult(suite)
result.done()
return self.output.getvalue()

def getResult(self, suite):
def getResult(self, suite: PyUnitTestSuite) -> reporter.Reporter:
suite.run(self.result)
return self.result

Expand Down Expand Up @@ -207,51 +214,48 @@ def test_doctestError(self):
expect = [self.doubleSeparator, re.compile(r"\[(ERROR|FAIL)\]")]
self.stringComparison(expect, output.splitlines())

def test_hiddenException(self):
def test_hiddenException(self) -> None:
"""
Check that errors in C{DelayedCall}s get reported, even if the
test already has a failure.
When a function scheduled using L{IReactorTime.callLater} in a
test method raises an exception that exception is added to the test
result as an error.
This happens even if the test also fails and the test failure is also
added to the test result as a failure.
Only really necessary for testing the deprecated style of tests that
use iterate() directly. See
L{erroneous.DelayedCall.testHiddenException} for more details.
"""
from twisted.internet import reactor

if reflect.qual(reactor).startswith("twisted.internet.asyncioreactor"):
raise self.skipTest(
"This test does not work on the asyncio reactor, as the "
"traceback comes from inside asyncio, not Twisted."
)

test = erroneous.DelayedCall("testHiddenException")
output = self.getOutput(test).splitlines()
errorQual = qual(RuntimeError)
match = [
self.doubleSeparator,
"[FAIL]",
"Traceback (most recent call last):",
re.compile(
r"^\s+File .*erroneous\.py., line \d+, in " "testHiddenException$"

result = self.getResult(PyUnitTestSuite([test]))
assert_that(
result, matches_result(errors=has_length(1), failures=has_length(1))
)
[(actualCase, error)] = result.errors
assert_that(test, equal_to(actualCase))
assert_that(
error,
isFailure(
type=equal_to(RuntimeError),
value=after(str, equal_to("something blew up")),
frames=has_item(similarFrame("go", "erroneous.py")), # type: ignore[arg-type]
),
re.compile(
r'^\s+self\.fail\("Deliberate failure to mask the '
r'hidden exception"\)$'
)

[(actualCase, failure)] = result.failures
assert_that(test, equal_to(actualCase))
assert_that(
failure,
isFailure(
type=equal_to(test.failureException),
value=after(
str, equal_to("Deliberate failure to mask the hidden exception")
),
frames=has_item(similarFrame("testHiddenException", "erroneous.py")), # type: ignore[arg-type]
),
"twisted.trial.unittest.FailTest: "
"Deliberate failure to mask the hidden exception",
"twisted.trial.test.erroneous.DelayedCall.testHiddenException",
self.doubleSeparator,
"[ERROR]",
"Traceback (most recent call last):",
re.compile(r"^\s+File .* in runUntilCurrent"),
re.compile(r"^\s+.*"),
re.compile(r'^\s+File .*erroneous\.py", line \d+, in go'),
re.compile(r"^\s+raise RuntimeError\(self.hiddenExceptionMsg\)"),
errorQual + ": something blew up",
"twisted.trial.test.erroneous.DelayedCall.testHiddenException",
]
self.stringComparison(match, output)
)


class UncleanWarningWrapperErrorReportingTests(ErrorReportingTests):
Expand All @@ -263,7 +267,12 @@ class UncleanWarningWrapperErrorReportingTests(ErrorReportingTests):
def setUp(self):
self.loader = runner.TestLoader()
self.output = StringIO()
self.result = UncleanWarningsReporterWrapper(reporter.Reporter(self.output))
self.reporter: reporter.Reporter = reporter.Reporter(self.output)
self.result = UncleanWarningsReporterWrapper(self.reporter)

def getResult(self, suite: PyUnitTestSuite) -> reporter.Reporter:
suite.run(self.result)
return self.reporter


class TracebackHandlingTests(unittest.SynchronousTestCase):
Expand Down

0 comments on commit 8ee190c

Please sign in to comment.