diff --git a/setup.cfg b/setup.cfg index 1c567871501..18c01766fab 100644 --- a/setup.cfg +++ b/setup.cfg @@ -51,6 +51,7 @@ packages = find: test = cython-test-exception-raiser >= 1.0.2, <2 PyHamcrest >= 1.9.0 + hypothesis ~= 6.0 ; List of dependencies required to build the documentation and test the ; release scripts and process. diff --git a/src/twisted/newsfragments/11616.misc b/src/twisted/newsfragments/11616.misc new file mode 100644 index 00000000000..e69de29bb2d diff --git a/src/twisted/trial/_dist/test/matchers.py b/src/twisted/trial/_dist/test/matchers.py index 44a34eb4038..d783fec8b24 100644 --- a/src/twisted/trial/_dist/test/matchers.py +++ b/src/twisted/trial/_dist/test/matchers.py @@ -5,8 +5,48 @@ Hamcrest matchers useful throughout the test suite. """ -from hamcrest import equal_to, has_length, has_properties +__all__ = [ + "matches_result", + "HasSum", + "IsSequenceOf", +] + +from typing import List, Sequence, Tuple, TypeVar + +from hamcrest import ( + contains_exactly, + contains_string, + equal_to, + has_length, + has_properties, + instance_of, +) +from hamcrest.core.base_matcher import BaseMatcher +from hamcrest.core.core.allof import AllOf +from hamcrest.core.description import Description from hamcrest.core.matcher import Matcher +from typing_extensions import Protocol + +from twisted.python.failure import Failure + +T = TypeVar("T") + + +class Semigroup(Protocol[T]): + """ + A type with an associative binary operator. + + Common examples of a semigroup are integers with addition and strings with + concatenation. + """ + + def __add__(self, other: T) -> T: + """ + This must be associative: a + (b + c) == (a + b) + c + """ + + +S = TypeVar("S", bound=Semigroup) def matches_result( @@ -30,3 +70,123 @@ def matches_result( "unexpectedSuccesses": unexpectedSuccesses, } ) + + +class HasSum(BaseMatcher[Sequence[S]]): + """ + Match a sequence the elements of which sum to a value matched by + another matcher. + + :ivar sumMatcher: The matcher which must match the sum. + :ivar zero: The zero value for the matched type. + """ + + def __init__(self, sumMatcher: Matcher[S], zero: S) -> None: + self.sumMatcher = sumMatcher + self.zero = zero + + def _sum(self, sequence: Sequence[S]) -> S: + if not sequence: + return self.zero + result = self.zero + for elem in sequence: + result = result + elem + return result + + def _matches(self, item: Sequence[S]) -> bool: + """ + Determine whether the sum of the sequence is matched. + """ + s = self._sum(item) + return self.sumMatcher.matches(s) + + def describe_mismatch(self, item: Sequence[S], description: Description) -> None: + """ + Describe the mismatch. + """ + s = self._sum(item) + description.append_description_of(self) + self.sumMatcher.describe_mismatch(s, description) + return None + + def describe_to(self, description: Description) -> None: + """ + Describe this matcher for error messages. + """ + description.append_text("a sequence with sum ") + description.append_description_of(self.sumMatcher) + description.append_text(", ") + + +class IsSequenceOf(BaseMatcher[Sequence[T]]): + """ + Match a sequence where every element is matched by another matcher. + + :ivar elementMatcher: The matcher which must match every element of the + sequence. + """ + + def __init__(self, elementMatcher: Matcher[T]) -> None: + self.elementMatcher = elementMatcher + + def _matches(self, item: Sequence[T]) -> bool: + """ + Determine whether every element of the sequence is matched. + """ + for elem in item: + if not self.elementMatcher.matches(elem): + return False + return True + + def describe_mismatch(self, item: Sequence[T], description: Description) -> None: + """ + Describe the mismatch. + """ + for idx, elem in enumerate(item): + if not self.elementMatcher.matches(elem): + description.append_description_of(self) + description.append_text(f"not sequence with element #{idx} {elem!r}") + + def describe_to(self, description: Description) -> None: + """ + Describe this matcher for error messages. + """ + description.append_text("a sequence containing only ") + description.append_description_of(self.elementMatcher) + description.append_text(", ") + + +def isFailure(**properties: Matcher[object]) -> Matcher[object]: + """ + Match an instance of L{Failure} with matching attributes. + """ + return AllOf( + instance_of(Failure), + has_properties(**properties), + ) + + +def similarFrame( + functionName: str, fileName: str +) -> Matcher[Sequence[Tuple[str, str, int, List[object], List[object]]]]: + """ + Match a tuple representation of a frame like those used by + L{twisted.python.failure.Failure}. + """ + # The frames depend on exact layout of the source + # code in files and on the filesystem so we won't + # bother being very precise here. Just verify we + # see some distinctive fragments. + # + # In particular, the last frame should be a tuple like + # + # (functionName, fileName, someint, [], []) + return contains_exactly( + equal_to(functionName), + contains_string(fileName), # type: ignore[arg-type] + instance_of(int), # type: ignore[arg-type] + # Unfortunately Failure makes them sometimes tuples, sometimes + # dict_items. + has_length(0), # type: ignore[arg-type] + has_length(0), # type: ignore[arg-type] + ) diff --git a/src/twisted/trial/_dist/test/test_matchers.py b/src/twisted/trial/_dist/test/test_matchers.py new file mode 100644 index 00000000000..35ac7070940 --- /dev/null +++ b/src/twisted/trial/_dist/test/test_matchers.py @@ -0,0 +1,188 @@ +""" +Tests for L{twisted.trial._dist.test.matchers}. +""" + +from typing import Callable, Sequence, Tuple, Type + +from hamcrest import anything, assert_that, contains, contains_string, equal_to, not_ +from hamcrest.core.matcher import Matcher +from hamcrest.core.string_description import StringDescription +from hypothesis import given +from hypothesis.strategies import ( + binary, + booleans, + integers, + just, + lists, + one_of, + sampled_from, + text, + tuples, +) + +from twisted.python.failure import Failure +from twisted.trial.unittest import SynchronousTestCase +from .matchers import HasSum, IsSequenceOf, S, isFailure, similarFrame + +Summer = Callable[[Sequence[S]], S] +concatInt = sum +concatStr = "".join +concatBytes = b"".join + + +class HasSumTests(SynchronousTestCase): + """ + Tests for L{HasSum}. + """ + + summables = one_of( + tuples(lists(integers()), just(concatInt)), + tuples(lists(text()), just(concatStr)), + tuples(lists(binary()), just(concatBytes)), + ) + + @given(summables) + def test_matches(self, summable: Tuple[Sequence[S], Summer[S]]) -> None: + """ + L{HasSum} matches a sequence if the elements sum to a value matched by + the parameterized matcher. + + :param summable: A tuple of a sequence of values to try to match and a + function which can compute the correct sum for that sequence. + """ + seq, sumFunc = summable + expected = sumFunc(seq) + zero = sumFunc([]) + matcher = HasSum(equal_to(expected), zero) + + description = StringDescription() + assert_that(matcher.matches(seq, description), equal_to(True)) + assert_that(str(description), equal_to("")) + + @given(summables) + def test_mismatches( + self, + summable: Tuple[ + Sequence[S], + Summer[S], + ], + ) -> None: + """ + L{HasSum} does not match a sequence if the elements do not sum to a + value matched by the parameterized matcher. + + :param summable: See L{test_matches}. + """ + seq, sumFunc = summable + zero = sumFunc([]) + # A matcher that never matches. + sumMatcher: Matcher[S] = not_(anything()) + matcher = HasSum(sumMatcher, zero) + + actualDescription = StringDescription() + assert_that(matcher.matches(seq, actualDescription), equal_to(False)) + + sumMatcherDescription = StringDescription() + sumMatcherDescription.append_description_of(sumMatcher) + actualStr = str(actualDescription) + assert_that(actualStr, contains_string("a sequence with sum")) + assert_that(actualStr, contains_string(str(sumMatcherDescription))) + + +class IsSequenceOfTests(SynchronousTestCase): + """ + Tests for L{IsSequenceOf}. + """ + + sequences = lists(booleans()) + + @given(integers(min_value=0, max_value=1000)) + def test_matches(self, numItems: int) -> None: + """ + L{IsSequenceOf} matches a sequence if all of the elements are + matched by the parameterized matcher. + + :param numItems: The length of a sequence to try to match. + """ + seq = [True] * numItems + matcher = IsSequenceOf(equal_to(True)) + + actualDescription = StringDescription() + assert_that(matcher.matches(seq, actualDescription), equal_to(True)) + assert_that(str(actualDescription), equal_to("")) + + @given(integers(min_value=0, max_value=1000), integers(min_value=0, max_value=1000)) + def test_mismatches(self, numBefore: int, numAfter: int) -> None: + """ + L{IsSequenceOf} does not match a sequence if any of the elements + are not matched by the parameterized matcher. + + :param numBefore: In the sequence to try to match, the number of + elements expected to match before an expected mismatch. + + :param numAfter: In the sequence to try to match, the number of + elements expected expected to match after an expected mismatch. + """ + # Hide the non-matching value somewhere in the sequence. + seq = [True] * numBefore + [False] + [True] * numAfter + matcher = IsSequenceOf(equal_to(True)) + + actualDescription = StringDescription() + assert_that(matcher.matches(seq, actualDescription), equal_to(False)) + actualStr = str(actualDescription) + assert_that(actualStr, contains_string("a sequence containing only")) + assert_that( + actualStr, contains_string(f"not sequence with element #{numBefore}") + ) + + +class IsFailureTests(SynchronousTestCase): + """ + Tests for L{isFailure}. + """ + + @given(sampled_from([ValueError, ZeroDivisionError, RuntimeError])) + def test_matches(self, excType: Type[BaseException]) -> None: + """ + L{isFailure} matches instances of L{Failure} with matching + attributes. + + :param excType: An exception type to wrap in a L{Failure} to be + matched against. + """ + matcher = isFailure(type=equal_to(excType)) + failure = Failure(excType()) + assert_that(matcher.matches(failure), equal_to(True)) + + @given(sampled_from([ValueError, ZeroDivisionError, RuntimeError])) + def test_mismatches(self, excType: Type[BaseException]) -> None: + """ + L{isFailure} does not match instances of L{Failure} with + attributes that don't match. + + :param excType: An exception type to wrap in a L{Failure} to be + matched against. + """ + matcher = isFailure(type=equal_to(excType), other=not_(anything())) + failure = Failure(excType()) + assert_that(matcher.matches(failure), equal_to(False)) + + def test_frames(self): + """ + The L{similarFrame} matcher matches elements of the C{frames} list + of a L{Failure}. + """ + try: + raise ValueError("Oh no") + except BaseException: + f = Failure() + + actualDescription = StringDescription() + matcher = isFailure( + frames=contains(similarFrame("test_frames", "test_matchers")) + ) + assert_that( + matcher.matches(f, actualDescription), + equal_to(True), + actualDescription, + ) diff --git a/src/twisted/trial/_dist/test/test_worker.py b/src/twisted/trial/_dist/test/test_worker.py index 03c6a857a54..19e520d713d 100644 --- a/src/twisted/trial/_dist/test/test_worker.py +++ b/src/twisted/trial/_dist/test/test_worker.py @@ -7,19 +7,20 @@ import os from io import BytesIO, StringIO +from typing import Type +from unittest import TestCase as PyUnitTestCase from zope.interface.verify import verifyObject -from twisted.internet.defer import fail, succeed +from hamcrest import assert_that, equal_to, has_item, has_length + +from twisted.internet.defer import fail from twisted.internet.error import ProcessDone from twisted.internet.interfaces import IAddress, ITransport -from twisted.protocols.amp import AMP from twisted.python.failure import Failure from twisted.python.filepath import FilePath -from twisted.python.reflect import fullyQualifiedName -from twisted.scripts import trial -from twisted.test.proto_helpers import StringTransport -from twisted.trial._dist import managercommands, workercommands +from twisted.test.iosim import connectedServerAndClient +from twisted.trial._dist import managercommands from twisted.trial._dist.worker import ( LocalWorker, LocalWorkerAMP, @@ -29,13 +30,9 @@ WorkerProtocol, ) from twisted.trial.reporter import TestResult -from twisted.trial.unittest import TestCase - - -class FakeAMP(AMP): - """ - A fake amp protocol. - """ +from twisted.trial.test import pyunitcases, skipping +from twisted.trial.unittest import TestCase, makeTodo +from .matchers import isFailure, matches_result, similarFrame class WorkerProtocolTests(TestCase): @@ -43,41 +40,34 @@ class WorkerProtocolTests(TestCase): Tests for L{WorkerProtocol}. """ - def setUp(self): + worker: WorkerProtocol + server: LocalWorkerAMP + + def setUp(self) -> None: """ Set up a transport, a result stream and a protocol instance. """ - self.serverTransport = StringTransport() - self.clientTransport = StringTransport() - self.server = WorkerProtocol() - self.server.makeConnection(self.serverTransport) - self.client = FakeAMP() - self.client.makeConnection(self.clientTransport) + self.worker, self.server, pump = connectedServerAndClient( + LocalWorkerAMP, WorkerProtocol, greet=False + ) + self.flush = pump.flush - def test_run(self): + def test_run(self) -> None: """ - Calling the L{workercommands.Run} command on the client returns a + Sending the L{workercommands.Run} command to the worker returns a response with C{success} sets to C{True}. """ - d = self.client.callRemote(workercommands.Run, testCase="doesntexist") - - def check(result): - self.assertTrue(result["success"]) + d = self.server.run(pyunitcases.PyUnitTest("test_pass"), TestResult()) + self.flush() + self.assertEqual({"success": True}, self.successResultOf(d)) - d.addCallback(check) - self.server.dataReceived(self.clientTransport.value()) - self.clientTransport.clear() - self.client.dataReceived(self.serverTransport.value()) - self.serverTransport.clear() - return d - - def test_start(self): + def test_start(self) -> None: """ The C{start} command changes the current path. """ curdir = os.path.realpath(os.path.curdir) self.addCleanup(os.chdir, curdir) - self.server.start("..") + self.worker.start("..") self.assertNotEqual(os.path.realpath(os.path.curdir), curdir) @@ -86,197 +76,126 @@ class LocalWorkerAMPTests(TestCase): Test case for distributed trial's manager-side local worker AMP protocol """ - def setUp(self): - self.managerTransport = StringTransport() - self.managerAMP = LocalWorkerAMP() - self.managerAMP.makeConnection(self.managerTransport) - self.result = TestResult() - self.workerTransport = StringTransport() - self.worker = AMP() - self.worker.makeConnection(self.workerTransport) - - config = trial.Options() - self.testName = "twisted.doesnexist" - config["tests"].append(self.testName) - self.testCase = trial._getSuite(config)._tests.pop() - - self.managerAMP.run(self.testCase, self.result) - self.managerTransport.clear() + def setUp(self) -> None: + self.worker, self.managerAMP, pump = connectedServerAndClient( + LocalWorkerAMP, WorkerProtocol, greet=False + ) + self.flush = pump.flush - def pumpTransports(self): - """ - Sends data from C{self.workerTransport} to C{self.managerAMP}, and then - data from C{self.managerTransport} back to C{self.worker}. - """ - self.managerAMP.dataReceived(self.workerTransport.value()) - self.workerTransport.clear() - self.worker.dataReceived(self.managerTransport.value()) + def workerRunTest( + self, testCase: PyUnitTestCase, makeResult: Type[TestResult] = TestResult + ) -> TestResult: + result = makeResult() + d = self.managerAMP.run(testCase, result) + self.flush() + self.assertEqual({"success": True}, self.successResultOf(d)) + return result - def test_runSuccess(self): + def test_runSuccess(self) -> None: """ Run a test, and succeed. """ - results = [] - - d = self.worker.callRemote(managercommands.AddSuccess, testName=self.testName) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - - self.assertTrue(results) + result = self.workerRunTest(pyunitcases.PyUnitTest("test_pass")) + assert_that(result, matches_result(successes=equal_to(1))) - def test_runExpectedFailure(self): + def test_runExpectedFailure(self) -> None: """ Run a test, and fail expectedly. """ - results = [] + expectedCase = skipping.SynchronousStrictTodo("test_todo1") + result = self.workerRunTest(expectedCase) + assert_that(result, matches_result(expectedFailures=has_length(1))) + [(actualCase, exceptionMessage, todoReason)] = result.expectedFailures + assert_that(actualCase, equal_to(expectedCase)) - d = self.worker.callRemote( - managercommands.AddExpectedFailure, - testName=self.testName, - error="error", - todo="todoReason", - ) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - - self.assertEqual(self.testCase, self.result.expectedFailures[0][0]) - self.assertTrue(results) + # Match the strings used in the test we ran. + assert_that(exceptionMessage, equal_to("expected failure")) + assert_that(todoReason, equal_to(makeTodo("todo1"))) - def test_runError(self): + def test_runError(self) -> None: """ Run a test, and encounter an error. """ - results = [] - errorClass = fullyQualifiedName(ValueError) - d = self.worker.callRemote( - managercommands.AddError, - testName=self.testName, - error="error", - errorClass=errorClass, - frames=[], + expectedCase = pyunitcases.PyUnitTest("test_error") + result = self.workerRunTest(expectedCase) + assert_that(result, matches_result(errors=has_length(1))) + [(actualCase, failure)] = result.errors + assert_that(expectedCase, equal_to(actualCase)) + assert_that( + failure, + isFailure( + type=equal_to(Exception), + value=equal_to(WorkerException("pyunit error")), + frames=has_item(similarFrame("test_error", "pyunitcases.py")), # type: ignore[arg-type] + ), ) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - - case, failure = self.result.errors[0] - self.assertEqual(self.testCase, case) - self.assertEqual(failure.type, ValueError) - self.assertEqual(failure.value, WorkerException("error")) - self.assertTrue(results) - - def test_runErrorWithFrames(self): - """ - L{LocalWorkerAMP._buildFailure} recreates the C{Failure.frames} from - the C{frames} argument passed to C{AddError}. - """ - results = [] - errorClass = fullyQualifiedName(ValueError) - d = self.worker.callRemote( - managercommands.AddError, - testName=self.testName, - error="error", - errorClass=errorClass, - frames=["file.py", "invalid code", "3"], - ) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - - case, failure = self.result.errors[0] - self.assertEqual(self.testCase, case) - self.assertEqual(failure.type, ValueError) - self.assertEqual(failure.value, WorkerException("error")) - self.assertEqual([("file.py", "invalid code", 3, [], [])], failure.frames) - self.assertTrue(results) - def test_runFailure(self): + def test_runFailure(self) -> None: """ Run a test, and fail. """ - results = [] - failClass = fullyQualifiedName(RuntimeError) - d = self.worker.callRemote( - managercommands.AddFailure, - testName=self.testName, - fail="fail", - failClass=failClass, - frames=[], + expectedCase = pyunitcases.PyUnitTest("test_fail") + result = self.workerRunTest(expectedCase) + assert_that(result, matches_result(failures=has_length(1))) + [(actualCase, failure)] = result.failures + assert_that(expectedCase, equal_to(actualCase)) + assert_that( + failure, + isFailure( + # AssertionError is the type raised by TestCase.fail + type=equal_to(AssertionError), + value=equal_to(WorkerException("pyunit failure")), + ), ) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - case, failure = self.result.failures[0] - self.assertEqual(self.testCase, case) - self.assertEqual(failure.type, RuntimeError) - self.assertEqual(failure.value, WorkerException("fail")) - self.assertTrue(results) - - def test_runSkip(self): + def test_runSkip(self) -> None: """ Run a test, but skip it. """ - results = [] - - d = self.worker.callRemote( - managercommands.AddSkip, testName=self.testName, reason="reason" - ) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - - self.assertEqual(self.testCase, self.result.skips[0][0]) - self.assertTrue(results) + expectedCase = pyunitcases.PyUnitTest("test_skip") + result = self.workerRunTest(expectedCase) + assert_that(result, matches_result(skips=has_length(1))) + [(actualCase, skip)] = result.skips + assert_that(expectedCase, equal_to(actualCase)) + assert_that(skip, equal_to("pyunit skip")) - def test_runUnexpectedSuccesses(self): + def test_runUnexpectedSuccesses(self) -> None: """ Run a test, and succeed unexpectedly. """ - results = [] + expectedCase = skipping.SynchronousStrictTodo("test_todo7") + result = self.workerRunTest(expectedCase) + assert_that(result, matches_result(unexpectedSuccesses=has_length(1))) + [(actualCase, unexpectedSuccess)] = result.unexpectedSuccesses + assert_that(expectedCase, equal_to(actualCase)) + assert_that(unexpectedSuccess, equal_to("todo7")) - d = self.worker.callRemote( - managercommands.AddUnexpectedSuccess, testName=self.testName, todo="todo" - ) - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - - self.assertEqual(self.testCase, self.result.unexpectedSuccesses[0][0]) - self.assertTrue(results) - - def test_testWrite(self): + def test_testWrite(self) -> None: """ L{LocalWorkerAMP.testWrite} writes the data received to its test stream. """ - results = [] stream = StringIO() self.managerAMP.setTestStream(stream) - - command = managercommands.TestWrite - d = self.worker.callRemote(command, out="Some output") - d.addCallback(lambda result: results.append(result["success"])) - self.pumpTransports() - + d = self.worker.callRemote(managercommands.TestWrite, out="Some output") + self.flush() + self.assertEqual({"success": True}, self.successResultOf(d)) self.assertEqual("Some output\n", stream.getvalue()) - self.assertTrue(results) - def test_stopAfterRun(self): + def test_stopAfterRun(self) -> None: """ L{LocalWorkerAMP.run} calls C{stopTest} on its test result once the C{Run} commands has succeeded. """ - result = object() stopped = [] - def fakeCallRemote(command, testCase): - return succeed(result) - - self.managerAMP.callRemote = fakeCallRemote - class StopTestResult(TestResult): - def stopTest(self, test): + def stopTest(self, test: PyUnitTestCase) -> None: stopped.append(test) - d = self.managerAMP.run(self.testCase, StopTestResult()) - self.assertEqual([self.testCase], stopped) - return d.addCallback(self.assertIdentical, result) + case = pyunitcases.PyUnitTest("test_pass") + self.workerRunTest(case, StopTestResult) + assert_that(stopped, equal_to([case])) class SpyDataLocalWorkerAMP(LocalWorkerAMP): diff --git a/src/twisted/trial/_dist/worker.py b/src/twisted/trial/_dist/worker.py index e4768f56f56..cb9a446b8de 100644 --- a/src/twisted/trial/_dist/worker.py +++ b/src/twisted/trial/_dist/worker.py @@ -185,7 +185,7 @@ def addExpectedFailure( """ Add an expected failure to the reporter. """ - _todo = Todo(todo) + _todo = Todo("" if todo is None else todo) self._result.addExpectedFailure(self._testCase, error, _todo) return {"success": True} diff --git a/src/twisted/trial/_synctest.py b/src/twisted/trial/_synctest.py index 8969e642971..dcd091f7258 100644 --- a/src/twisted/trial/_synctest.py +++ b/src/twisted/trial/_synctest.py @@ -17,11 +17,13 @@ import unittest as pyunit import warnings from dis import findlinestarts as _findlinestarts -from typing import List, NoReturn, Optional, Tuple, TypeVar, Union +from typing import Iterable, List, NoReturn, Optional, Tuple, Type, TypeVar, Union # Python 2.7 and higher has skip support built-in from unittest import SkipTest +from attrs import frozen + from twisted.internet.defer import Deferred, ensureDeferred from twisted.python import failure, log, monkey from twisted.python.deprecate import ( @@ -43,6 +45,7 @@ class FailTest(AssertionError): """ +@frozen class Todo: """ Internal object used to mark a L{TestCase} as 'todo'. Tests marked 'todo' @@ -50,19 +53,17 @@ class Todo: they do not fail the suite and the errors are reported in a separate category. If todo'd tests succeed, Trial L{TestResult}s will report an unexpected success. - """ - def __init__(self, reason, errors=None): - """ - @param reason: A string explaining why the test is marked 'todo' + @ivar reason: A string explaining why the test is marked 'todo' - @param errors: An iterable of exception types that the test is - expected to raise. If one of these errors is raised by the test, it - will be trapped. Raising any other kind of error will fail the test. - If L{None} is passed, then all errors will be trapped. - """ - self.reason = reason - self.errors = errors + @ivar errors: An iterable of exception types that the test is expected to + raise. If one of these errors is raised by the test, it will be + trapped. Raising any other kind of error will fail the test. If + L{None} then all errors will be trapped. + """ + + reason: str + errors: Optional[Iterable[Type[BaseException]]] = None def __repr__(self) -> str: return f"" @@ -81,7 +82,11 @@ def expected(self, failure): return False -def makeTodo(value): +def makeTodo( + value: Union[ + str, Tuple[Union[Type[BaseException], Iterable[Type[BaseException]]], str] + ] +) -> Todo: """ Return a L{Todo} object built from C{value}. @@ -98,11 +103,11 @@ def makeTodo(value): return Todo(reason=value) if isinstance(value, tuple): errors, reason = value - try: - errors = list(errors) - except TypeError: - errors = [errors] - return Todo(reason=reason, errors=errors) + if isinstance(errors, type): + iterableErrors: Iterable[Type[BaseException]] = [errors] + else: + iterableErrors = errors + return Todo(reason=reason, errors=iterableErrors) class _Warning: diff --git a/src/twisted/trial/reporter.py b/src/twisted/trial/reporter.py index 7e7d672a84b..04bd44db598 100644 --- a/src/twisted/trial/reporter.py +++ b/src/twisted/trial/reporter.py @@ -15,6 +15,7 @@ import unittest as pyunit import warnings from collections import OrderedDict +from typing import TYPE_CHECKING, List, Tuple from zope.interface import implementer @@ -24,13 +25,16 @@ from twisted.python.util import untilConcludes from twisted.trial import itrial, util +if TYPE_CHECKING: + from ._synctest import Todo + try: from subunit import TestProtocolClient # type: ignore[import] except ImportError: TestProtocolClient = None -def _makeTodo(value): +def _makeTodo(value: str) -> "Todo": """ Return a L{Todo} object built from C{value}. @@ -81,6 +85,11 @@ class TestResult(pyunit.TestResult): # Used when no todo provided to addExpectedFailure or addUnexpectedSuccess. _DEFAULT_TODO = "Test expected to fail" + skips: List[Tuple[itrial.ITestCase, str]] + expectedFailures: List[Tuple[itrial.ITestCase, str, "Todo"]] # type: ignore[assignment] + unexpectedSuccesses: List[Tuple[itrial.ITestCase, str]] # type: ignore[assignment] + successes: int + def __init__(self): super().__init__() self.skips = []