# Path: stestr/repository/abstract.py
# class AbstractRepositoryFactory:
# class AbstractRepository:
# class AbstractTestRun:
# class RepositoryNotFound(Exception):
#     def initialise(self, url):
#     def open(self, url):
#     def count(self):
#     def get_failing(self):
#     def get_run_ids(self):
#     def remove_run_id(self, run_id):
#     def get_inserter(self, partial=False, run_id=None, metadata=None):
#     def _get_inserter(self, partial=False, run_id=None, metadata=None):
#     def get_latest_run(self):
#     def get_test_run(self, run_id):
#     def get_test_times(self, test_ids):
#     def _get_test_times(self, test_ids):
#     def latest_id(self):
#     def get_test_ids(self, run_id):
#         def gather(test_dict):
#     def find_metadata(self, metadata):
#     def get_id(self):
#     def get_subunit_stream(self):
#     def get_test(self):
#     def get_metadata(self):
#     def __init__(self, url):
# 
# Path: stestr/utils.py
# class CallWhenProcFinishes:
#     def __init__(self, process, callback):
#     def stdin(self):
#     def stdout(self):
#     def stderr(self):
#     def returncode(self):
#     def wait(self):
# def _iter_internal_streams(input_streams, stream_type):
# def iter_streams(input_streams, stream_type):
# def cleanup_test_name(name, strip_tags=True, strip_scenarios=False):


import errno
import os
import sys
import tempfile
import subunit.v2
import testtools
from io import BytesIO
from operator import methodcaller
from future.moves.dbm import dumb as my_dbm
from subunit import TestProtocolClient
from testtools.compat import _b
from stestr.repository import abstract as repository
from stestr import utils
        from stestr.repository import memory

    def __init__(self, repository, partial=False, run_id=None, metadata=None):
        # XXX: Perhaps should factor into a decorator and use an unaltered
        # TestProtocolClient.
        self._repository = repository
        self._run_id = run_id
        self._metadata = metadata
        if not self._run_id:
            fd, name = tempfile.mkstemp(dir=self._repository.base)
            self.fname = name
            stream = os.fdopen(fd, 'wb')
        else:
            self.fname = os.path.join(self._repository.base, self._run_id)
            stream = open(self.fname, 'ab')
        self.partial = partial
        # The time take by each test, flushed at the end.
        self._times = {}
        self._test_start = None
        self._time = None
        subunit_client = testtools.StreamToExtendedDecorator(
            TestProtocolClient(stream))
        self.hook = testtools.CopyStreamResult([
            subunit_client,
            testtools.StreamToDict(self._handle_test)])
        self._stream = stream

    def _handle_test(self, test_dict):
        start, stop = test_dict['timestamps']
        if test_dict['status'] == 'exists' or None in (start, stop):
            return

# Path: stestr/repository/abstract.py
# class AbstractRepositoryFactory:
# class AbstractRepository:
# class AbstractTestRun:
# class RepositoryNotFound(Exception):
#     def initialise(self, url):
#     def open(self, url):
#     def count(self):
#     def get_failing(self):
#     def get_run_ids(self):
#     def remove_run_id(self, run_id):
#     def get_inserter(self, partial=False, run_id=None, metadata=None):
#     def _get_inserter(self, partial=False, run_id=None, metadata=None):
#     def get_latest_run(self):
#     def get_test_run(self, run_id):
#     def get_test_times(self, test_ids):
#     def _get_test_times(self, test_ids):
#     def latest_id(self):
#     def get_test_ids(self, run_id):
#         def gather(test_dict):
#     def find_metadata(self, metadata):
#     def get_id(self):
#     def get_subunit_stream(self):
#     def get_test(self):
#     def get_metadata(self):
#     def __init__(self, url):
# 
# Path: stestr/utils.py
# class CallWhenProcFinishes:
#     def __init__(self, process, callback):
#     def stdin(self):
#     def stdout(self):
#     def stderr(self):
#     def returncode(self):
#     def wait(self):
# def _iter_internal_streams(input_streams, stream_type):
# def iter_streams(input_streams, stream_type):
# def cleanup_test_name(name, strip_tags=True, strip_scenarios=False):


# Path: stestr/repository/file.py
import errno
import os
import sys
import tempfile
import subunit.v2
import testtools
from io import BytesIO
from operator import methodcaller
from future.moves.dbm import dumb as my_dbm
from subunit import TestProtocolClient
from testtools.compat import _b
from stestr.repository import abstract as repository
from stestr import utils
        from stestr.repository import memory


    def __init__(self, repository, partial=False, run_id=None, metadata=None):
        # XXX: Perhaps should factor into a decorator and use an unaltered
        # TestProtocolClient.
        self._repository = repository
        self._run_id = run_id
        self._metadata = metadata
        if not self._run_id:
            fd, name = tempfile.mkstemp(dir=self._repository.base)
            self.fname = name
            stream = os.fdopen(fd, 'wb')
        else:
            self.fname = os.path.join(self._repository.base, self._run_id)
            stream = open(self.fname, 'ab')
        self.partial = partial
        # The time take by each test, flushed at the end.
        self._times = {}
        self._test_start = None
        self._time = None
        subunit_client = testtools.StreamToExtendedDecorator(
            TestProtocolClient(stream))
        self.hook = testtools.CopyStreamResult([
            subunit_client,
            testtools.StreamToDict(self._handle_test)])
        self._stream = stream

    def _handle_test(self, test_dict):
        start, stop = test_dict['timestamps']
        if test_dict['status'] == 'exists' or None in (start, stop):
            return

        test_id = utils.cleanup_test_name(test_dict['id'])