Python Subunit
Subunit is a streaming protocol for test results, designed to be easily generated and parsed. The `python-subunit` library provides extensions to Python's `unittest` framework, enabling the generation and consumption of Subunit streams. It facilitates test aggregation, archiving, isolation, and grid testing across different languages and machines. The library supports both Version 1 (human-readable) and Version 2 (binary) of the protocol, with a focus on Version 2 for improved robustness and multiplexing. It is currently at version 1.4.5 and maintained with a moderate release cadence.
Warnings
- breaking Subunit has two major protocol versions: v1 (human-readable) and v2 (binary). While v2 is more robust and intended to supersede v1, `python-subunit`'s bundled tools *only* accept and emit v2. Interoperating with older third-party libraries that use v1 requires explicit conversion filters (`subunit-1to2` and `subunit-2to1`). This can cause compatibility issues if not handled.
- gotcha When extending `unittest.TestResult` objects with `python-subunit`'s extensions (e.g., for tags, extra details, timestamps), `TestResult` objects that do *not* implement these extension methods will either lose fidelity or discard the extended data without raising an error. This can lead to silent data loss if the consuming `TestResult` is not fully compatible with the `subunit` extensions.
Install
-
pip install python-subunit
Imports
- TestProtocolClient
from subunit import TestProtocolClient
- ProtocolTestCase
from subunit import ProtocolTestCase
- StreamResult
from subunit.test_results import StreamResult
Quickstart
import unittest
import io
from subunit import TestProtocolClient, ProtocolTestCase
from subunit.test_results import StreamResult
# 1. Define a simple unittest.TestCase
class MyTests(unittest.TestCase):
def test_success(self):
self.assertTrue(True)
def test_failure(self):
self.fail("This test explicitly failed")
# 2. Capture a test run as a Subunit stream
stream_buffer = io.BytesIO()
# TestProtocolClient is a TestResult, so a TextTestRunner can use it
result_client = TestProtocolClient(stream_buffer)
runner = unittest.TextTestRunner(result=result_client)
print("--- Running tests and capturing Subunit stream ---")
suite = unittest.TestSuite()
suite.addTest(MyTests('test_success'))
suite.addTest(MyTests('test_failure'))
runner.run(suite)
subunit_stream_bytes = stream_buffer.getvalue()
print("\n--- Captured Subunit Stream (raw bytes) ---")
# For demonstration, decode and print the start of the stream if it's text-like
try:
print(subunit_stream_bytes.decode('utf-8')[:200] + '...' if len(subunit_stream_bytes) > 200 else subunit_stream_bytes.decode('utf-8'))
except UnicodeDecodeError:
print(subunit_stream_bytes[:200], '... (binary stream)')
# 3. Parse the Subunit stream back into unittest results
class MyStreamProcessor(StreamResult):
def status(self, test_id=None, test_status=None, **kwargs):
super().status(test_id=test_id, test_status=test_status, **kwargs)
print(f"Processing status for {test_id}: {test_status}")
def addSuccess(self, test):
print(f"Test passed: {test.id()}")
def addFailure(self, test, err):
print(f"Test failed: {test.id()} - {err[0].__name__}: {err[1]}")
print("\n--- Parsing Subunit stream back into results ---")
parse_result = MyStreamProcessor()
# ProtocolTestCase can read the stream and feed events to a TestResult
ProtocolTestCase.run_with_stream(subunit_stream_bytes, parse_result)