tessl/pypi-twisted
An asynchronous networking framework written in Python
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
testing.mddocs/
Testing Framework
Comprehensive unit testing framework with asynchronous test support and specialized test runners. Twisted Trial extends Python's unittest with Deferred support, making it ideal for testing asynchronous code.
Capabilities
Test Cases
Base test case classes with support for asynchronous operations and Twisted-specific testing patterns.
class trial.unittest.TestCase:
"""
Asynchronous-capable test case for testing Twisted code.
Extends unittest.TestCase with Deferred support.
Attributes:
- timeout: Default test timeout in seconds
"""
timeout = 120.0
def setUp(self):
"""
Set up test fixture. Can return Deferred for async setup.
Returns:
None or Deferred: Completion indicator
"""
def tearDown(self):
"""
Clean up test fixture. Can return Deferred for async cleanup.
Returns:
None or Deferred: Completion indicator
"""
def addCleanup(self, function, *args, **kwargs):
"""
Add cleanup function to run after test.
Args:
function: Cleanup function
*args, **kwargs: Arguments for function
"""
def doCleanups(self):
"""
Run all registered cleanup functions.
Returns:
None or Deferred: Completion indicator
"""
def assertFailure(self, deferred, *expectedFailures):
"""
Assert that a Deferred will fail with specific exception types.
Args:
deferred (Deferred): Deferred to test
*expectedFailures: Expected exception types
Returns:
Deferred[Failure]: Fires with failure when deferred fails
"""
def successResultOf(self, deferred):
"""
Get successful result of already-fired Deferred.
Args:
deferred (Deferred): Deferred that should have succeeded
Returns:
Result value of the Deferred
"""
def failureResultOf(self, deferred, *expectedFailures):
"""
Get failure result of already-failed Deferred.
Args:
deferred (Deferred): Deferred that should have failed
*expectedFailures: Expected exception types
Returns:
Failure: The failure
"""
def assertNoResult(self, deferred):
"""
Assert that Deferred has not yet fired.
Args:
deferred (Deferred): Deferred to check
"""
class trial.unittest.SynchronousTestCase:
"""
Synchronous test case that cannot handle Deferreds.
Use for tests that don't involve any asynchronous operations.
Faster than TestCase but cannot return Deferreds from test methods.
"""
def setUp(self):
"""Set up test fixture synchronously."""
def tearDown(self):
"""Clean up test fixture synchronously."""
# Test decorators and markers
class trial.unittest.Todo:
"""
Mark a test as expected to fail (todo item).
"""
def __init__(self, reason, errors=None):
"""
Args:
reason (str): Reason for expected failure
errors: Expected error types
"""
def trial.unittest.skip(reason):
"""
Skip a test with given reason.
Args:
reason (str): Skip reason
Returns:
Decorator function
"""
def trial.unittest.skipIf(condition, reason):
"""
Skip test if condition is true.
Args:
condition (bool): Skip condition
reason (str): Skip reason
Returns:
Decorator function
"""
class trial.unittest.SkipTest(Exception):
"""
Exception raised to skip a test.
"""
class trial.unittest.FailTest(Exception):
"""
Exception raised to fail a test.
"""Test Case Usage Example:
from twisted.trial import unittest
from twisted.internet import defer, reactor, task
class AsyncTestCase(unittest.TestCase):
def setUp(self):
# Async setup
self.service = MyService()
return self.service.start()
def tearDown(self):
# Async cleanup
return self.service.stop()
def test_async_operation(self):
"""Test that returns a Deferred."""
d = self.service.processData("test")
d.addCallback(self.assertEqual, "expected_result")
return d
def test_failure_case(self):
"""Test expected failure."""
d = self.service.badOperation()
return self.assertFailure(d, ValueError, TypeError)
@defer.inlineCallbacks
def test_inline_callbacks(self):
"""Test using inlineCallbacks for cleaner async code."""
result1 = yield self.service.step1()
result2 = yield self.service.step2(result1)
self.assertEqual(result2, "expected")
def test_synchronous_result(self):
"""Test already-completed Deferred."""
d = defer.succeed("immediate")
result = self.successResultOf(d)
self.assertEqual(result, "immediate")
class SyncTestCase(unittest.SynchronousTestCase):
def test_pure_sync(self):
"""Test that doesn't involve any async operations."""
result = self.service.sync_method()
self.assertEqual(result, "expected")Test Suites and Discovery
Test suite management and test discovery mechanisms.
class trial.unittest.TestSuite:
"""
Container for multiple test cases.
"""
def __init__(self, tests=()):
"""
Args:
tests: Sequence of test cases
"""
def addTest(self, test):
"""
Add a test to the suite.
Args:
test: Test case or suite to add
"""
def addTests(self, tests):
"""
Add multiple tests to the suite.
Args:
tests: Sequence of tests to add
"""
def countTestCases(self):
"""
Count total number of test cases.
Returns:
int: Number of test cases
"""
class runner.TestLoader:
"""
Test discovery and loading.
"""
def loadTestsFromTestCase(self, testCaseClass):
"""
Load tests from a test case class.
Args:
testCaseClass: Test case class
Returns:
TestSuite: Suite containing tests
"""
def loadTestsFromModule(self, module):
"""
Load tests from a module.
Args:
module: Python module containing tests
Returns:
TestSuite: Suite containing tests
"""
def loadTestsFromName(self, name):
"""
Load tests from dotted name.
Args:
name (str): Dotted test name
Returns:
TestSuite: Suite containing tests
"""
def unittest.decorate(testClass, classDecorator):
"""
Apply decorator to all test methods in a class.
Args:
testClass: Test case class
classDecorator: Decorator function
Returns:
Decorated test class
"""Test Runners
Test execution engines with various output formats and reporting options.
class runner.TrialRunner:
"""
Main test runner for executing trial tests.
Attributes:
- stream: Output stream for results
- tbformat: Traceback format ('default', 'brief', 'verbose')
- args: Command line arguments
"""
stream = None
tbformat = 'default'
def __init__(self, reporterFactory, mode=None, logfile='-', stream=None, profile=False, tracebackFormat='default', realTimeErrors=False, uncleanWarnings=False, workingDirectory=None, tests=None):
"""
Args:
reporterFactory: Factory for creating result reporters
mode: Test mode ('debug', etc.)
logfile: Path for log output
stream: Output stream
profile: Whether to enable profiling
tracebackFormat: Format for tracebacks
realTimeErrors: Show errors in real time
uncleanWarnings: Show cleanup warnings
workingDirectory: Working directory for tests
tests: Test suite to run
"""
def run(self, test):
"""
Run tests with this runner.
Args:
test: Test case or suite to run
Returns:
TestResult: Test execution results
"""
class runner.LoggedSuite:
"""
Test suite that logs test execution.
"""
def __init__(self, tests=(), logfile=None):
"""
Args:
tests: Test cases to include
logfile: File for logging output
"""Test Reporters
Result reporting and output formatting for test runs.
class reporter.TestResult:
"""
Test result collector and reporter base class.
Attributes:
- successes: Number of successful tests
- errors: List of test errors
- failures: List of test failures
- skips: List of skipped tests
- expectedFailures: List of expected failures
- unexpectedSuccesses: List of unexpected successes
"""
successes = 0
errors = None
failures = None
skips = None
expectedFailures = None
unexpectedSuccesses = None
def startTest(self, test):
"""
Called when test starts.
Args:
test: Test case being started
"""
def stopTest(self, test):
"""
Called when test ends.
Args:
test: Test case that ended
"""
def addError(self, test, error):
"""
Add test error.
Args:
test: Test case
error: Error tuple (type, value, traceback)
"""
def addFailure(self, test, failure):
"""
Add test failure.
Args:
test: Test case
failure: Failure tuple (type, value, traceback)
"""
def addSuccess(self, test):
"""
Add successful test.
Args:
test: Test case
"""
def addSkip(self, test, reason):
"""
Add skipped test.
Args:
test: Test case
reason (str): Skip reason
"""
class reporter.TreeReporter:
"""
Tree-style test reporter showing hierarchical test structure.
"""
def __init__(self, stream, tbformat='default', realtime=False, publisher=None):
"""
Args:
stream: Output stream
tbformat (str): Traceback format
realtime (bool): Show results in real time
publisher: Log publisher
"""
class reporter.VerboseTextReporter:
"""
Verbose text reporter with detailed output.
"""
def __init__(self, stream, tbformat='default', realtime=False, publisher=None):
"""
Args:
stream: Output stream
tbformat (str): Traceback format
realtime (bool): Show results in real time
publisher: Log publisher
"""
class reporter.MinimalReporter:
"""
Minimal test reporter with concise output.
"""
def __init__(self, stream, tbformat='default', realtime=False, publisher=None):
"""
Args:
stream: Output stream
tbformat (str): Traceback format
realtime (bool): Show results in real time
publisher: Log publisher
"""
class reporter.TimingTextReporter:
"""
Text reporter that includes timing information.
"""
def __init__(self, stream, tbformat='default', realtime=False, publisher=None):
"""
Args:
stream: Output stream
tbformat (str): Traceback format
realtime (bool): Show results in real time
publisher: Log publisher
"""
class reporter.SubunitReporter:
"""
Reporter that outputs results in Subunit format.
"""
def __init__(self, stream, tbformat='default', realtime=False, publisher=None):
"""
Args:
stream: Output stream
tbformat (str): Traceback format
realtime (bool): Show results in real time
publisher: Log publisher
"""Testing Utilities
Helper functions and utilities for writing and running tests.
class util.TestCase:
"""Additional testing utilities."""
def patch(self, obj, attribute, value):
"""
Patch an object attribute for the duration of the test.
Args:
obj: Object to patch
attribute (str): Attribute name
value: New attribute value
"""
def mktemp(self):
"""
Create a temporary file path unique to this test.
Returns:
str: Temporary file path
"""
def util.suppress(action=None, *suppressedWarnings):
"""
Context manager/decorator to suppress warnings during tests.
Args:
action: Warning action ('ignore', 'error', etc.)
*suppressedWarnings: Warning categories to suppress
"""
def util.acquireAttribute(objects, attr, default=None):
"""
Get attribute from first object that has it.
Args:
objects: Sequence of objects to check
attr (str): Attribute name
default: Default value if not found
Returns:
Attribute value or default
"""
# Test constants
util.DEFAULT_TIMEOUT_DURATION = 120.0 # Default test timeoutTesting Utilities Example:
from twisted.trial import unittest, util
class UtilityTestCase(unittest.TestCase):
def setUp(self):
self.temp_file = self.mktemp() # Unique temp file path
self.original_value = MyClass.class_attribute
def test_with_patch(self):
# Temporarily patch an attribute
self.patch(MyClass, 'class_attribute', 'test_value')
self.assertEqual(MyClass.class_attribute, 'test_value')
# Attribute restored automatically after test
@util.suppress('ignore', DeprecationWarning)
def test_with_suppressed_warnings(self):
# This test won't show deprecation warnings
deprecated_function()
def tearDown(self):
# Clean up temp file if created
if os.path.exists(self.temp_file):
os.unlink(self.temp_file)Mock Objects and Test Doubles
Utilities for creating mock objects and test doubles in Twisted tests.
class proto_helpers.StringTransport:
"""
Transport that stores written data in memory for testing.
Attributes:
- value: Data written to transport (bytes)
- disconnected: Whether transport is disconnected
"""
value = b''
disconnected = False
def write(self, data):
"""
Write data to in-memory buffer.
Args:
data (bytes): Data to write
"""
def writeSequence(self, data):
"""
Write sequence of data chunks.
Args:
data: Sequence of bytes objects
"""
def loseConnection(self):
"""Mark transport as disconnected."""
def clear(self):
"""Clear stored data."""
def value(self):
"""
Get all written data.
Returns:
bytes: All data written to transport
"""
class proto_helpers.StringTransportWithDisconnection:
"""
StringTransport that notifies protocol of disconnection.
"""
def loseConnection(self):
"""Disconnect and notify protocol."""
class proto_helpers.AccumulatingProtocol:
"""
Protocol that accumulates all received data.
Attributes:
- data: All received data (bytes)
- transport: Connected transport
"""
data = b''
transport = None
def dataReceived(self, data):
"""
Accumulate received data.
Args:
data (bytes): Received data
"""Protocol Testing Example:
from twisted.trial import unittest
from twisted.test import proto_helpers
class ProtocolTestCase(unittest.TestCase):
def setUp(self):
self.protocol = MyProtocol()
self.transport = proto_helpers.StringTransport()
self.protocol.makeConnection(self.transport)
def test_data_handling(self):
# Send data to protocol
self.protocol.dataReceived(b"test data")
# Check what protocol wrote back
sent_data = self.transport.value()
self.assertEqual(sent_data, b"expected response")
def test_disconnection(self):
# Test protocol disconnection
self.transport.loseConnection()
self.assertTrue(self.transport.disconnected)
# Protocol should handle disconnection gracefully
self.assertFalse(self.protocol.connected)Required Imports
Complete import statements for testing with Twisted Trial:
# Core testing framework
from twisted.trial import unittest, runner, reporter
from twisted.trial.unittest import TestCase, SynchronousTestCase
from twisted.python.failure import Failure
from twisted.internet import defer
# Test decorators and utilities
from twisted.trial.unittest import skip, skipIf, Todo, SkipTest
from twisted.test import proto_helpers # For StringTransport
# Common patterns
from twisted.trial.unittest import TestCase as TrialTestCase
from twisted.internet.defer import succeed, fail, inlineCallbacks, returnValue
# Running tests
# Command line: trial mypackage.test.test_module
# In code: trial.unittest.main()Install with Tessl CLI
npx tessl i tessl/pypi-twisted