Python Subunit

1.4.5 · active · verified Sun Apr 12

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

Install

Imports

Quickstart

This quickstart demonstrates how to use `python-subunit` to both generate and consume a Subunit stream. It defines a standard `unittest.TestCase`, captures its execution output using `subunit.TestProtocolClient` into a byte stream, and then uses `subunit.ProtocolTestCase` with a custom `StreamResult` to parse and report on that stream. This illustrates the core functionality of converting `unittest` results to Subunit and back.

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)

view raw JSON →