tessl/pypi-testtools
Extensions to the Python standard library unit testing framework
twisted-support.mddocs/
Twisted Integration
Complete integration with Twisted framework for testing asynchronous code using Deferreds, including specialized matchers and test execution classes.
Capabilities
Deferred Matchers
Specialized matchers for testing Twisted Deferred objects and asynchronous operations.
def succeeded():
"""
Match successfully resolved Twisted Deferred objects.
Matches Deferred objects that have fired successfully
with a result value (not an error).
Returns:
Matcher: Deferred success matcher
Example:
self.assertThat(deferred, succeeded())
"""
def failed():
"""
Match failed Twisted Deferred objects.
Matches Deferred objects that have fired with an
error/exception rather than a successful result.
Returns:
Matcher: Deferred failure matcher
Example:
self.assertThat(deferred, failed())
"""
def has_no_result():
"""
Match Deferred objects with no result yet.
Matches Deferred objects that have not yet fired
(neither success nor failure).
Returns:
Matcher: Deferred pending matcher
Example:
self.assertThat(pending_deferred, has_no_result())
"""Asynchronous Test Execution
Test runner classes for executing tests that return Deferred objects.
class AsynchronousDeferredRunTest(testtools.RunTest):
"""
Run Deferred-returning tests asynchronously.
Executes test methods that return Deferred objects,
properly handling the asynchronous completion and
result propagation.
"""
def __init__(self, case, handlers=None, reactor=None, timeout=None):
"""
Create asynchronous Deferred test runner.
Args:
case: TestCase instance
handlers: Optional exception handlers
reactor: Twisted reactor instance
timeout (float): Optional test timeout in seconds
"""
def run(self, result):
"""
Execute the test asynchronously.
Runs the test method and waits for Deferred completion,
handling both success and failure cases appropriately.
Args:
result: TestResult to record results
Returns:
TestResult: The test result after completion
"""
class AsynchronousDeferredRunTestForBrokenTwisted(AsynchronousDeferredRunTest):
"""
Workaround for broken Twisted versions.
Alternative asynchronous test runner that works around
known issues in certain Twisted versions or configurations.
"""
def __init__(self, case, handlers=None, reactor=None, timeout=None):
"""
Create workaround asynchronous test runner.
Args:
case: TestCase instance
handlers: Optional exception handlers
reactor: Twisted reactor instance
timeout (float): Optional test timeout
"""
class SynchronousDeferredRunTest(testtools.RunTest):
"""
Run Deferred-returning tests synchronously.
Executes Deferred-returning tests in a synchronous manner
by spinning the reactor until completion.
"""
def __init__(self, case, handlers=None, reactor=None, timeout=None):
"""
Create synchronous Deferred test runner.
Args:
case: TestCase instance
handlers: Optional exception handlers
reactor: Twisted reactor instance
timeout (float): Optional test timeout
"""
def run(self, result):
"""
Execute the test synchronously.
Runs the test and spins the reactor until the
Deferred completes, then returns the result.
Args:
result: TestResult to record results
Returns:
TestResult: The test result after completion
"""Log Capture and Management
Classes for capturing and managing Twisted log output during tests.
class CaptureTwistedLogs(testtools.RunTest):
"""
Capture Twisted log output during test execution.
Intercepts Twisted log messages generated during test
execution and attaches them to test results for debugging.
"""
def __init__(self, case, handlers=None):
"""
Create log capturing test runner.
Args:
case: TestCase instance
handlers: Optional exception handlers
"""
def run(self, result):
"""
Execute test with log capture.
Runs the test while capturing all Twisted log output
and attaching it to the test result for analysis.
Args:
result: TestResult to record results and logs
Returns:
TestResult: Test result with captured logs
"""Helper Functions
Utility functions for common Twisted testing patterns.
def assert_fails_with(deferred, *expected_failures):
"""
Assert that Deferred fails with specific error type.
Verifies that a Deferred object fails and that the
failure matches one of the expected error types.
Args:
deferred: Deferred object to check
*expected_failures: Expected exception types
Returns:
Deferred: Deferred that fires when assertion completes
Example:
d = failing_operation()
return assert_fails_with(d, ValueError, TypeError)
"""
def flush_logged_errors(*error_types):
"""
Flush logged errors from Twisted logging system.
Removes accumulated error log entries of specified types
from Twisted's global error log, useful for test cleanup.
Args:
*error_types: Error types to flush from logs
Returns:
list: List of flushed error entries
Example:
# Clean up expected errors after test
flush_logged_errors(ConnectionLost, TimeoutError)
"""Usage Examples
Basic Deferred Testing
import testtools
from testtools.twistedsupport import *
from twisted.internet import defer, reactor
from twisted.internet.defer import inlineCallbacks
class AsyncTest(testtools.TestCase):
def test_successful_deferred(self):
"""Test Deferred that completes successfully."""
d = defer.succeed("test result")
self.assertThat(d, succeeded())
def test_failed_deferred(self):
"""Test Deferred that fails."""
d = defer.fail(ValueError("test error"))
self.assertThat(d, failed())
def test_pending_deferred(self):
"""Test Deferred that hasn't fired yet."""
d = defer.Deferred()
self.assertThat(d, has_no_result())
# Later, fire the deferred
d.callback("result")
self.assertThat(d, succeeded())Asynchronous Test Methods
import testtools
from testtools.twistedsupport import AsynchronousDeferredRunTest
from twisted.internet import defer, reactor
from twisted.internet.defer import inlineCallbacks
import time
class AsyncOperationTest(testtools.TestCase):
@testtools.run_test_with(AsynchronousDeferredRunTest)
def test_async_operation(self):
"""Test method that returns a Deferred."""
def slow_operation():
d = defer.Deferred()
reactor.callLater(0.1, d.callback, "completed")
return d
# Return Deferred from test method
d = slow_operation()
d.addCallback(lambda result: self.assertEqual(result, "completed"))
return d
@testtools.run_test_with(AsynchronousDeferredRunTest)
@inlineCallbacks
def test_with_inline_callbacks(self):
"""Test using inlineCallbacks decorator."""
def async_fetch(url):
d = defer.Deferred()
reactor.callLater(0.05, d.callback, f"data from {url}")
return d
# Use yield with inlineCallbacks
result1 = yield async_fetch("http://example.com/data1")
result2 = yield async_fetch("http://example.com/data2")
self.assertIn("data1", result1)
self.assertIn("data2", result2)Error Handling and Failures
import testtools
from testtools.twistedsupport import *
from twisted.internet import defer
from twisted.internet.error import ConnectionLost, TimeoutError
class ErrorHandlingTest(testtools.TestCase):
@testtools.run_test_with(AsynchronousDeferredRunTest)
def test_expected_failure(self):
"""Test that expects a specific failure."""
def failing_operation():
return defer.fail(ConnectionLost("Connection dropped"))
d = failing_operation()
return assert_fails_with(d, ConnectionLost)
@testtools.run_test_with(AsynchronousDeferredRunTest)
def test_multiple_possible_failures(self):
"""Test that can fail with different error types."""
def unstable_operation():
import random
if random.random() < 0.5:
return defer.fail(TimeoutError("Request timed out"))
else:
return defer.fail(ConnectionLost("Connection lost"))
d = unstable_operation()
return assert_fails_with(d, TimeoutError, ConnectionLost)
def test_error_cleanup(self):
"""Test with error cleanup."""
# Generate some errors that get logged
defer.fail(ValueError("test error 1"))
defer.fail(TypeError("test error 2"))
# Clean up the logged errors
flushed = flush_logged_errors(ValueError, TypeError)
self.assertEqual(len(flushed), 2)Log Capture Testing
import testtools
from testtools.twistedsupport import CaptureTwistedLogs
from twisted.python import log
from twisted.internet import defer
class LogCaptureTest(testtools.TestCase):
@testtools.run_test_with(CaptureTwistedLogs)
def test_with_log_capture(self):
"""Test that captures Twisted logs."""
def operation_with_logging():
log.msg("Starting operation")
log.msg("Processing data")
d = defer.succeed("result")
d.addCallback(lambda r: log.msg(f"Operation completed: {r}") or r)
return d
d = operation_with_logging()
d.addCallback(lambda result: self.assertEqual(result, "result"))
return d
# Logs are automatically captured and attached to test resultComplex Asynchronous Testing
import testtools
from testtools.twistedsupport import *
from twisted.internet import defer, reactor
from twisted.internet.defer import inlineCallbacks, gatherResults
class ComplexAsyncTest(testtools.TestCase):
@testtools.run_test_with(AsynchronousDeferredRunTest)
@inlineCallbacks
def test_multiple_async_operations(self):
"""Test coordinating multiple async operations."""
def fetch_user(user_id):
d = defer.Deferred()
reactor.callLater(0.1, d.callback, {"id": user_id, "name": f"User{user_id}"})
return d
def fetch_permissions(user_id):
d = defer.Deferred()
reactor.callLater(0.05, d.callback, ["read", "write"])
return d
# Fetch user data and permissions concurrently
user_d = fetch_user(123)
perms_d = fetch_permissions(123)
user, permissions = yield gatherResults([user_d, perms_d])
self.assertEqual(user["id"], 123)
self.assertIn("read", permissions)
@testtools.run_test_with(AsynchronousDeferredRunTest)
def test_chained_operations(self):
"""Test chained asynchronous operations."""
def step1():
return defer.succeed("step1_result")
def step2(prev_result):
self.assertEqual(prev_result, "step1_result")
return defer.succeed("step2_result")
def step3(prev_result):
self.assertEqual(prev_result, "step2_result")
return defer.succeed("final_result")
# Chain the operations
d = step1()
d.addCallback(step2)
d.addCallback(step3)
d.addCallback(lambda result: self.assertEqual(result, "final_result"))
return d
@testtools.run_test_with(AsynchronousDeferredRunTest)
def test_timeout_handling(self):
"""Test operations with timeout."""
def slow_operation():
d = defer.Deferred()
# This operation takes too long
reactor.callLater(2.0, d.callback, "too slow")
return d
# Set up timeout
d = slow_operation()
timeout_d = defer.Deferred()
reactor.callLater(0.5, timeout_d.errback,
TimeoutError("Operation timed out"))
# Race between operation and timeout
racing_d = defer.DeferredList([d, timeout_d],
fireOnOneCallback=True,
fireOnOneErrback=True,
consumeErrors=True)
return assert_fails_with(racing_d, TimeoutError)Integration with TestCase Features
import testtools
from testtools.twistedsupport import *
from testtools.matchers import *
from twisted.internet import defer
class IntegratedAsyncTest(testtools.TestCase):
@testtools.run_test_with(AsynchronousDeferredRunTest)
def test_with_matchers_and_content(self):
"""Test combining async testing with matchers and content."""
def fetch_data():
# Simulate fetching data
data = {
"users": [{"name": "Alice"}, {"name": "Bob"}],
"total": 2,
"status": "success"
}
return defer.succeed(data)
def verify_data(data):
# Attach data to test results for debugging
self.addDetail('fetched_data',
testtools.content.json_content(data))
# Use matchers for sophisticated assertions
self.assertThat(data, MatchesDict({
"users": HasLength(2),
"total": Equals(2),
"status": Equals("success")
}))
# Verify user structure
self.assertThat(data["users"], AllMatch(
MatchesDict({"name": IsInstance(str)})
))
return data
d = fetch_data()
d.addCallback(verify_data)
return d
@testtools.run_test_with(AsynchronousDeferredRunTest)
@testtools.skipIf(not twisted_available, "Twisted not available")
def test_conditional_async(self):
"""Test with conditional execution."""
def maybe_async_operation():
if should_run_async():
return defer.succeed("async_result")
else:
return "sync_result"
result = yield maybe_async_operation()
self.assertIn("result", result)Install with Tessl CLI
npx tessl i tessl/pypi-testtools