課題の背景

テスト実行中に発生するprintなどの標準出力は、デバッグや結果分析において重要な情報源です。しかし、Pythonの標準ライブラリunittestでは、これらの出力はデフォルトでコンソールに直接表示されるため、個々のテストケースごとに分けて収集するには特別な対応が必要です。

一般的なアプローチとして、sys.stdoutを一時的なio.StringIOインスタンスにリダイレクトする方法があります。これにより、各テストの出力を独立して取得できます。ただし、この手法はマルチスレッド並列実行時において問題が生じます。複数のスレッドが同じグローバルな標準出力ストリームを共有するため、異なるテスト間で出力内容が混在してしまう可能性があります。

解決策：スレッド毎の出力バッファ管理

この問題を解決するために、スレッドIDに基づいて出力を分離して保持するカスタムIOクラスを設計します。これにより、並列実行中でも各テストケースの出力を正確に取得することが可能になります。

1. スレッド固有のStringIOクラスの実装

以下のように、threading.current_thread().identをキーとして出力を保存するThreadSafeStringIOクラスを作成します。

import io
import threading
from collections import defaultdict

class ThreadSafeStringIO(io.StringIO):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._buffers = defaultdict(str)

    def write(self, text: str) -> int:
        thread_id = threading.get_ident()
        self._buffers[thread_id] += text
        return super().write(text)

    def writelines(self, lines: list) -> None:
        text = ''.join(lines)
        thread_id = threading.get_ident()
        self._buffers[thread_id] += text
        super().writelines(lines)

    def get_current_buffer(self) -> str:
        thread_id = threading.get_ident()
        return self._buffers.get(thread_id, "")

    def clear_current_buffer(self) -> None:
        thread_id = threading.get_ident()
        self._buffers[thread_id] = ""
        self.seek(0)
        self.truncate()

2. TestResultにおける出力リダイレクトのオーバーライド

カスタムTestResultクラス内で、標準出力のセットアップと復元処理を再定義します。

import sys
import unittest

class CustomTestResult(unittest.TestResult):
    def __init__(self, verbosity=1):
        super().__init__()
        self.verbosity = verbosity
        self._stdout_capture = ThreadSafeStringIO()
        self._stderr_capture = ThreadSafeStringIO()
        self._original_stdout = sys.stdout
        self._original_stderr = sys.stderr
        self.test_results = []

    def _setup_output_capture(self):
        sys.stdout = self._stdout_capture
        sys.stderr = self._stderr_capture

    def _restore_output_and_capture(self):
        thread_id = threading.get_ident()
        stdout_content = self._stdout_capture.get_current_buffer()
        stderr_content = self._stderr_capture.get_current_buffer()
        
        # スレッドバッファと内部バッファをクリア
        self._stdout_capture.clear_current_buffer()
        self._stderr_capture.clear_current_buffer()

        # 元のストリームに戻す
        sys.stdout = self._original_stdout
        sys.stderr = self._original_stderr

        return stdout_content, stderr_content

3. テストライフサイクルとの統合

テスト開始時にスレッドIDを記録し、終了時にそのIDに対応する出力を取得します。

    def startTest(self, test):
        super().startTest(test)
        test.start_time = time.time()
        test.thread_id = threading.get_ident()
        self._setup_output_capture()

    def stopTest(self, test):
        stdout_data, stderr_data = self._restore_output_and_capture()
        test.output = (stdout_data + stderr_data).strip()
        test.duration = time.time() - test.start_time
        self.test_results.append({
            'name': test._testMethodName,
            'status': getattr(test, 'status', 'success'),
            'output': test.output,
            'duration': test.duration
        })
        super().stopTest(test)

4. 実行結果の構造化

各テストの結果は辞書形式で集約され、後続のレポート生成やログ出力に利用できます。

    @property
    def summary(self):
        return {
            'total_run': len(self.test_results),
            'successes': len([r for r in self.test_results if r['status'] == 'success']),
            'failures': len([r for r in self.test_results if r['status'] == 'fail']),
            'errors': len([r for r in self.test_results if r['status'] == 'error']),
            'duration': sum(r['duration'] for r in self.test_results),
            'details': self.test_results
        }

この設計により、マルチスレッド環境下でも各テストケースの出力を確実に分離して収集でき、並列実行のパフォーマンスメリットを損なうことなく、詳細なテスト診断情報を得ることが可能になります。