Implementation of job result (test result) classes.

plainbox.impl.result – job result

This module has two basic implementation of IJobResult: MemoryJobResult and DiskJobResult.

class plainbox.impl.result.DiskJobResult(data)[source]

Bases: plainbox.impl.result._JobResultBase

A IJobResult that keeps IO logs on disk.

This type of JobResult is intended for working with most results. It does not store IO logs in memory so it is scalable to arbitrary IO log sizes. Each instance just knows where the log file is located (using the ‘io_log_filename’ attribute for that) and offers streaming API for accessing particular parts of the log.

OUTCOME_CRASH = 'crash'
OUTCOME_FAIL = 'fail'
OUTCOME_NONE = None
OUTCOME_NOT_IMPLEMENTED = 'not-implemented'
OUTCOME_NOT_SUPPORTED = 'not-supported'
OUTCOME_PASS = 'pass'
OUTCOME_SKIP = 'skip'
OUTCOME_UNDECIDED = 'undecided'
comments

Get the comments of the test operator.

The comments are sanitized to remove control characters that would cause problems when parsing the submission file.

execution_duration

The amount of time in seconds it took to run this job.

get_builder(**kwargs)

Create a new job result builder from the data in this result.

get_io_log()[source]
io_log
io_log_as_flat_text

Perform a lossy conversion from the binary I/O log to text.

Convert the I/O log to a text string, replacing non Unicode characters with U+FFFD, the REPLACEMENT CHARACTER.

Both stdout and stderr streams are merged together into a single string. I/O log record are first decoded to UTF-8 and all control characters (EXCEPT for the newline, carriage return, tab and vertical space) are removed:

>>> result = MemoryJobResult({'io_log': [
...            (0, 'stdout', b'foo\n'),
...            (1, 'stderr', b'
barn’)]})
>>> result.io_log_as_flat_text
'foo\nbar\n'

When the input bytes can’t be converted they are replaced by U+FFFD:

>>> special_char = bytes([255,])
>>> result = MemoryJobResult({'io_log': [(0, 'stdout', special_char)]})
>>> result.io_log_as_flat_text
'�'
io_log_as_text_attachment

Perform a conversion of the binary I/O log to text, if possible.

Convert the I/O log to text attachment, if possible, otherwise return an empty string.

This method is similar to _JobResultBase.io_log_as_flat_text() but only merge stdout records to recreate the original attachment file.

Returns:stdout of the given job, converted to text (assuming UTF-8 encoding) with Unicode control characters removed, if possible, or an empty string otherwise.
io_log_filename

pathname of the file containing serialized IO log records.

is_hollow

flag that indicates if the result is hollow.

Hollow results may have been created but hold no data at all. Hollow results are also tentatively deprecated, once we have some time to re-factor SessionState and specifically the job_state_map code we will remove the need to have hollow results.

Hollow results are not saved, beginning with plainbox.impl.session.suspend.SessionSuspendHelper4.

outcome

outcome of running this job.

The outcome ultimately classifies jobs (tests) as failures or successes. There are several other types of outcome that all basically mean that the job did not run for some particular reason.

outcome_color_ansi()

Get an ANSI escape sequence that represents this outcome.

outcome_color_hex()

Get the hexadecimal “#RRGGBB” color that represents this outcome.

outcome_color_rgb()
outcome_meta()

Get the OutcomeMetadata object associated with this outcome.

return_code

return code of the command associated with the job, if any.

tr_outcome()

Get the translated value of the outcome.

class plainbox.impl.result.GzipFile(filename=None, mode=None, compresslevel=9, fileobj=None, mtime=None)[source]

Bases: gzip.GzipFile

Subclass of GzipFile that works around missing read1() on python3.2.

See: http://bugs.python.org/issue10791

close()[source]
closed
detach()

Disconnect this buffer from its underlying raw stream and return it.

After the raw stream has been detached, the buffer is in an unusable state.

filename
fileno()[source]

Invoke the underlying file object’s fileno() method.

This will raise AttributeError if the underlying file object doesn’t support fileno().

flush(zlib_mode=2)[source]
isatty()

Return whether this is an ‘interactive’ stream.

Return False if it can’t be determined.

mtime

Last modification time read from stream, or None

myfileobj = None
peek(n)[source]
read(size=-1)[source]
read1(n)[source]
readable()[source]
readinto()
readinto1()
readline(size=-1)[source]
readlines()

Return a list of lines from the stream.

hint can be specified to control the number of lines read: no more lines will be read if the total size (in bytes/characters) of all lines so far exceeds hint.

rewind()[source]

Return the uncompressed stream file position indicator to the beginning of the file

seek(offset, whence=0)[source]
seekable()[source]
tell()

Return current stream position.

truncate()

Truncate file to size bytes.

File pointer is left unchanged. Size defaults to the current IO position as reported by tell(). Returns the new size.

writable()[source]
write(data)[source]
writelines()
class plainbox.impl.result.IOLogRecord(delay, stream_name, data)

Bases: tuple

count(value) → integer -- return number of occurrences of value
data

Alias for field number 2

delay

Alias for field number 0

index(value[, start[, stop]]) → integer -- return first index of value.

Raises ValueError if the value is not present.

stream_name

Alias for field number 1

class plainbox.impl.result.IOLogRecordReader(stream)[source]

Bases: object

Class for streaming :class`IOLogRecord` instances from a text stream.

close()[source]
read_record()[source]

Read the next record from the stream.

Returns:None if the stream is empty
Returns:next IOLogRecord as found in the stream.
class plainbox.impl.result.IOLogRecordWriter(stream)[source]

Bases: object

Class for writing IOLogRecord instances to a text stream.

close()[source]
write_record(record)[source]

Write an IOLogRecord to the stream.

class plainbox.impl.result.JobResultBuilder(*args, **kwargs)[source]

Bases: plainbox.impl.pod.POD

A builder for job result objects.

add_comment(comment)[source]

Add a new comment.

The comment is safely combined with any prior comments.

as_dict() → dict

Return the data in this POD as a dictionary.

Note

UNSET values are not added to the dictionary.

as_tuple() → tuple

Return the data in this POD as a tuple.

Order of elements in the tuple corresponds to the order of field declarations.

comments

comments from the test operator

Side effects of assign filters:
  • unset or type-checked (value must be of type str)
execution_duration

time of test execution

Side effects of assign filters:
  • unset or type-checked (value must be of type float)
field_list = [<Field name:'outcome'>, <Field name:'execution_duration'>, <Field name:'comments'>, <Field name:'return_code'>, <Field name:'io_log'>, <Field name:'io_log_filename'>]
get_result()[source]

Use the current state of the builder to create a new result.

Returns:A new MemoryJobResult or DiskJobResult with all the data
Raises:ValueError – If both io_log and io_log_filename were used.
io_log

history of the I/O log of the (optional) test process

Side effects of assign filters:
  • unset or type-checked (value must be of type list)
  • unset or type-checked sequence (items must be of type tuple)
io_log_filename

path to a structured I/O log file of the (optional) test process

Side effects of assign filters:
  • unset or type-checked (value must be of type str)
namedtuple_cls

alias of JobResultBuilder

outcome

outcome of a test

Side effects of assign filters:
  • unset or type-checked (value must be of type str)
return_code

return code from the (optional) test process

Side effects of assign filters:
  • unset or type-checked (value must be of type int)
class plainbox.impl.result.MemoryJobResult(data)[source]

Bases: plainbox.impl.result._JobResultBase

A IJobResult that keeps IO logs in memory.

This type of JobResult is indented for writing unit tests where the hassle of going through the filesystem would make them needlessly complicated.

OUTCOME_CRASH = 'crash'
OUTCOME_FAIL = 'fail'
OUTCOME_NONE = None
OUTCOME_NOT_IMPLEMENTED = 'not-implemented'
OUTCOME_NOT_SUPPORTED = 'not-supported'
OUTCOME_PASS = 'pass'
OUTCOME_SKIP = 'skip'
OUTCOME_UNDECIDED = 'undecided'
comments

Get the comments of the test operator.

The comments are sanitized to remove control characters that would cause problems when parsing the submission file.

execution_duration

The amount of time in seconds it took to run this job.

get_builder(**kwargs)

Create a new job result builder from the data in this result.

get_io_log()[source]
io_log
io_log_as_flat_text

Perform a lossy conversion from the binary I/O log to text.

Convert the I/O log to a text string, replacing non Unicode characters with U+FFFD, the REPLACEMENT CHARACTER.

Both stdout and stderr streams are merged together into a single string. I/O log record are first decoded to UTF-8 and all control characters (EXCEPT for the newline, carriage return, tab and vertical space) are removed:

>>> result = MemoryJobResult({'io_log': [
...            (0, 'stdout', b'foo\n'),
...            (1, 'stderr', b'
barn’)]})
>>> result.io_log_as_flat_text
'foo\nbar\n'

When the input bytes can’t be converted they are replaced by U+FFFD:

>>> special_char = bytes([255,])
>>> result = MemoryJobResult({'io_log': [(0, 'stdout', special_char)]})
>>> result.io_log_as_flat_text
'�'
io_log_as_text_attachment

Perform a conversion of the binary I/O log to text, if possible.

Convert the I/O log to text attachment, if possible, otherwise return an empty string.

This method is similar to _JobResultBase.io_log_as_flat_text() but only merge stdout records to recreate the original attachment file.

Returns:stdout of the given job, converted to text (assuming UTF-8 encoding) with Unicode control characters removed, if possible, or an empty string otherwise.
is_hollow

flag that indicates if the result is hollow.

Hollow results may have been created but hold no data at all. Hollow results are also tentatively deprecated, once we have some time to re-factor SessionState and specifically the job_state_map code we will remove the need to have hollow results.

Hollow results are not saved, beginning with plainbox.impl.session.suspend.SessionSuspendHelper4.

outcome

outcome of running this job.

The outcome ultimately classifies jobs (tests) as failures or successes. There are several other types of outcome that all basically mean that the job did not run for some particular reason.

outcome_color_ansi()

Get an ANSI escape sequence that represents this outcome.

outcome_color_hex()

Get the hexadecimal “#RRGGBB” color that represents this outcome.

outcome_color_rgb()
outcome_meta()

Get the OutcomeMetadata object associated with this outcome.

return_code

return code of the command associated with the job, if any.

tr_outcome()

Get the translated value of the outcome.

class plainbox.impl.result.OutcomeMetadata(value, unicode_sigil, tr_outcome, tr_label, color_ansi, color_hex, hexr_xml_mapping, hexr_xml_allowed, hexr_xml_order)

Bases: tuple

color_ansi

Alias for field number 4

color_hex

Alias for field number 5

count(value) → integer -- return number of occurrences of value
hexr_xml_allowed

Alias for field number 7

hexr_xml_mapping

Alias for field number 6

hexr_xml_order

Alias for field number 8

index(value[, start[, stop]]) → integer -- return first index of value.

Raises ValueError if the value is not present.

tr_label

Alias for field number 3

tr_outcome

Alias for field number 2

unicode_sigil

Alias for field number 1

value

Alias for field number 0

plainbox.impl.result.outcome_color_ansi(outcome)[source]

Get an ANSI escape sequence that represents this outcome.

plainbox.impl.result.outcome_color_hex(outcome)[source]

Get the hexadecimal “#RRGGBB” color that represents this outcome.

plainbox.impl.result.outcome_meta(outcome)[source]

Get the OutcomeMetadata object associated with this outcome.

plainbox.impl.result.tr_outcome(outcome)[source]

Get the translated value of OUTCOME_ constant.

comments powered by Disqus