openstack/deb-python-testtools
Code Issues Proposed changes

Hide TestRecord and StreamToTestRecord

Browse Source 
While they are probably OK to be exposed, let's keep this patch minimal
& not news-worthy
This commit is contained in:
Jonathan Lange
2015-11-12 10:20:51 +00:00
parent bc8b162510
commit 7c9906eb8b
1 changed files with 20 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
testtools/testresult/real.py
View File

@@ -634,7 +634,7 @@ class StreamTagger(CopyStreamResult):
super(StreamTagger, self).status(*args, **kwargs) super(StreamTagger, self).status(*args, **kwargs)
class TestRecord(PClass): class _TestRecord(PClass):
"""Representation of a test.""" """Representation of a test."""
"""The test id.""" """The test id."""
@@ -679,8 +679,8 @@ class TestRecord(PClass):
* tags: The tags for the test. A set of unicode strings. * tags: The tags for the test. A set of unicode strings.
* details: A dict of file attachments - ``testtools.content.Content`` * details: A dict of file attachments - ``testtools.content.Content``
objects. objects.
* status: One of the StreamResult status codes (including inprogress) or * status: One of the StreamResult status codes (including inprogress)
'unknown' (used if only file events for a test were received...) or 'unknown' (used if only file events for a test were received...)
* timestamps: A pair of timestamps - the first one received with this * timestamps: A pair of timestamps - the first one received with this
test id, and the one in the event that triggered the notification. test id, and the one in the event that triggered the notification.
Hung tests have a None for the second end event. Timestamps are not Hung tests have a None for the second end event. Timestamps are not
@@ -771,36 +771,36 @@ _status_map = pmap({
}) })
class StreamToTestRecord(StreamResult): class _StreamToTestRecord(StreamResult):
"""A specialised StreamResult that emits a callback as tests complete. """A specialised StreamResult that emits a callback as tests complete.
Top level file attachments are simply discarded. Hung tests are detected Top level file attachments are simply discarded. Hung tests are detected
by stopTestRun and notified there and then. by stopTestRun and notified there and then.
The callback is passed a ``TestRecord`` object. The callback is passed a ``_TestRecord`` object.
Only the most recent tags observed in the stream are reported. Only the most recent tags observed in the stream are reported.
""" """
def __init__(self, on_test): def __init__(self, on_test):
"""Create a StreamToTestRecord calling on_test on test completions. """Create a _StreamToTestRecord calling on_test on test completions.
:param on_test: A callback that accepts one parameter: :param on_test: A callback that accepts one parameter:
a ``TestRecord`` object describing a test. a ``_TestRecord`` object describing a test.
""" """
super(StreamToTestRecord, self).__init__() super(_StreamToTestRecord, self).__init__()
self.on_test = on_test self.on_test = on_test
if parse_mime_type is None: if parse_mime_type is None:
raise ImportError("mimeparse module missing.") raise ImportError("mimeparse module missing.")
def startTestRun(self): def startTestRun(self):
super(StreamToTestRecord, self).startTestRun() super(_StreamToTestRecord, self).startTestRun()
self._inprogress = {} self._inprogress = {}
def status(self, test_id=None, test_status=None, test_tags=None, def status(self, test_id=None, test_status=None, test_tags=None,
runnable=True, file_name=None, file_bytes=None, eof=False, runnable=True, file_name=None, file_bytes=None, eof=False,
mime_type=None, route_code=None, timestamp=None): mime_type=None, route_code=None, timestamp=None):
super(StreamToTestRecord, self).status( super(_StreamToTestRecord, self).status(
test_id, test_status, test_id, test_status,
test_tags=test_tags, runnable=runnable, file_name=file_name, test_tags=test_tags, runnable=runnable, file_name=file_name,
file_bytes=file_bytes, eof=eof, mime_type=mime_type, file_bytes=file_bytes, eof=eof, mime_type=mime_type,
@@ -836,7 +836,7 @@ class StreamToTestRecord(StreamResult):
return case return case
def stopTestRun(self): def stopTestRun(self):
super(StreamToTestRecord, self).stopTestRun() super(_StreamToTestRecord, self).stopTestRun()
while self._inprogress: while self._inprogress:
case = self._inprogress.popitem()[1] case = self._inprogress.popitem()[1]
self.on_test(case.got_timestamp(None)) self.on_test(case.got_timestamp(None))
@@ -846,7 +846,7 @@ class StreamToTestRecord(StreamResult):
return return
key = (test_id, route_code) key = (test_id, route_code)
if key not in self._inprogress: if key not in self._inprogress:
self._inprogress[key] = TestRecord.create(test_id, timestamp) self._inprogress[key] = _TestRecord.create(test_id, timestamp)
return key return key
@@ -879,16 +879,16 @@ class StreamToDict(StreamResult):
# class, and have this inherit from that. # class, and have this inherit from that.
def __init__(self, on_test): def __init__(self, on_test):
"""Create a StreamToTestRecord calling on_test on test completions. """Create a _StreamToTestRecord calling on_test on test completions.
:param on_test: A callback that accepts one parameter: :param on_test: A callback that accepts one parameter:
a ``TestRecord`` object describing a test. a dictionary describing a test.
""" """
super(StreamToDict, self).__init__() super(StreamToDict, self).__init__()
self._hook = StreamToTestRecord(self._handle_test) self._hook = _StreamToTestRecord(self._handle_test)
# XXX: Not clear whether its part of the supported interface for # XXX: Not clear whether its part of the supported interface for
# self.on_test to be the passed-in on_test. If not, we could reduce # self.on_test to be the passed-in on_test. If not, we could reduce
# the boilerplate by subclassing StreamToTestRecord. # the boilerplate by subclassing _StreamToTestRecord.
self.on_test = on_test self.on_test = on_test
def _handle_test(self, test_record): def _handle_test(self, test_record):
@@ -913,7 +913,7 @@ def test_dict_to_case(test_dict):
:param test_dict: A test dict as generated by StreamToDict. :param test_dict: A test dict as generated by StreamToDict.
:return: A PlaceHolder test object. :return: A PlaceHolder test object.
""" """
return TestRecord( return _TestRecord(
id=test_dict['id'], id=test_dict['id'],
tags=test_dict['tags'], tags=test_dict['tags'],
details=test_dict['details'], details=test_dict['details'],
@@ -932,7 +932,7 @@ class StreamSummary(StreamResult):
def __init__(self): def __init__(self):
super(StreamSummary, self).__init__() super(StreamSummary, self).__init__()
self._hook = StreamToTestRecord(self._gather_test) self._hook = _StreamToTestRecord(self._gather_test)
self._handle_status = { self._handle_status = {
'success': self._success, 'success': self._success,
'skip': self._skip, 'skip': self._skip,
@@ -1688,8 +1688,8 @@ class StreamToExtendedDecorator(StreamResult):
# ExtendedToOriginalDecorator takes care of thunking details back to # ExtendedToOriginalDecorator takes care of thunking details back to
# exceptions/reasons etc. # exceptions/reasons etc.
self.decorated = ExtendedToOriginalDecorator(decorated) self.decorated = ExtendedToOriginalDecorator(decorated)
# StreamToTestRecord buffers and gives us individual tests. # _StreamToTestRecord buffers and gives us individual tests.
self.hook = StreamToTestRecord(self._handle_tests) self.hook = _StreamToTestRecord(self._handle_tests)
def status(self, test_id=None, test_status=None, *args, **kwargs): def status(self, test_id=None, test_status=None, *args, **kwargs):
if test_status == 'exists': if test_status == 'exists':