python-tdd/src/core/runner.py

import pickle
import time
from django.test.runner import DiscoverRunner, ParallelTestSuite, RemoteTestResult, RemoteTestRunner


class _Py313SafeRemoteTestResult(RemoteTestResult):
    """Prevents Python 3.13 'cannot pickle traceback' crash in parallel workers.

    In Python 3.13, traceback objects are not picklable. Django's parallel
    runner serialises test errors via multiprocessing (pickle) to send them
    from worker subprocesses back to the main process. When a Selenium test
    crashes the browser, the resulting exception traceback fails to pickle,
    aborting the entire parallel run with a confusing TypeError.

    Fix: strip the traceback (replace with None) before it reaches pickle.
    Error type and message are preserved; only the traceback is lost in transit.

    Note: the fix must be on RemoteTestResult (not RemoteTestRunner) because
    addError/addFailure — where check_picklable is called — live on the result
    object, not the runner. RemoteTestRunner.run() instantiates resultclass()
    and passes it to the test suite; overriding it on the runner has no effect.
    """

    @staticmethod
    def _safe_err(err):
        exc_type, exc_value, exc_tb = err
        try:
            pickle.dumps((exc_type, exc_value, exc_tb))
        except (TypeError, AttributeError):
            return (exc_type, exc_value, None)
        return err

    def addError(self, test, err):
        super().addError(test, self._safe_err(err))

    def addFailure(self, test, err):
        super().addFailure(test, self._safe_err(err))


class _Py313SafeRemoteTestRunner(RemoteTestRunner):
    resultclass = _Py313SafeRemoteTestResult


class _SafeParallelTestSuite(ParallelTestSuite):
    runner_class = _Py313SafeRemoteTestRunner


class RobustCompressorTestRunner(DiscoverRunner):
    parallel_test_suite = _SafeParallelTestSuite

    def setup_test_environment(self, **kwargs):
        super().setup_test_environment(**kwargs)
        from compressor.storage import CompressorFileStorage
        _orig_save = CompressorFileStorage.save
        def _robust_save(self, name, content):
            for _ in range(5):
                try: return _orig_save(self, name, content)
                except PermissionError: time.sleep(0.05)
            raise
        CompressorFileStorage.save = _robust_save