Twisted parallel test runner

gistfile1.py

class BufferWritesDevice(object):

  def __init__(self):
    self._data = []

  def write(self, string):
    self._data.append(string)

  def read(self):
    return ''.join(self._data)

  def flush(self, *args, **kwargs):
    pass

  def isatty(self):
    return False

....

def _tests_func(self, tests, worker_index):
  if not isinstance(tests, (list, set)):
    tests = [ tests ]

  args = [ '-e' ]
  args.extend(tests)

  config = Options()
  config.parseOptions(args)

  stream = BufferWritesDevice()
  runner = self._make_runner(config=config, stream=stream)
  suite = _getSuite(config)
  result = setup_test_db(worker_index, None, runner.run, suite)
  result = TestResult().from_trial_result(result)
  return result

...

def _make_runner(self, config, stream):
  # Based on twisted.scripts.trial._makeRunner
  mode = None
  if config['debug']:
      mode = TrialRunner.DEBUG
  if config['dry-run']:
      mode = TrialRunner.DRY_RUN
  return TrialRunner(config['reporter'],
                            mode=mode,
                            stream=stream,
                            profile=config['profile'],
                            logfile=config['logfile'],
                            tracebackFormat=config['tbformat'],
                            realTimeErrors=config['rterrors'],
                            uncleanWarnings=config['unclean-warnings'],
                            workingDirectory=config['temp-directory'],
                            forceGarbageCollection=config['force-gc'])

...