testplan.common.utils package

Subpackages

• testplan.common.utils.sockets package
  • Subpackages
    • testplan.common.utils.sockets.fix package
      • Submodules
        • testplan.common.utils.sockets.fix.client module
          • Client
            • Client.address
            • Client.close()
            • Client.connect()
            • Client.rawsend()
            • Client.rawsend_tsp()
            • Client.receive()
            • Client.send()
            • Client.sendlogoff()
            • Client.sendlogon()
        • testplan.common.utils.sockets.fix.parser module
          • FixParser
            • FixParser.State
              • FixParser.State.NotStarted
              • FixParser.State.ReadingBody
              • FixParser.State.ReadingCheckSum
              • FixParser.State.ReadingHeader
              • FixParser.State.ReadingLength
            • FixParser.consume()
          • tagsoverride()
        • testplan.common.utils.sockets.fix.utils module
          • utc_timestamp()
        • testplan.common.utils.sockets.fix.server module
          • ConnectionDetails
          • Server
            • Server.active_connections()
            • Server.flush()
            • Server.host
            • Server.ip
            • Server.is_connection_active()
            • Server.port
            • Server.receive()
            • Server.send()
            • Server.start()
            • Server.stop()
      • Module contents
  • Submodules
    • testplan.common.utils.sockets.client module
      • Client
        • Client.address
        • Client.close()
        • Client.connect()
        • Client.port
        • Client.receive()
        • Client.recv()
        • Client.send()
    • testplan.common.utils.sockets.codec module
      • Codec
        • Codec.parse()
        • Codec.serialize()
    • testplan.common.utils.sockets.message module
      • Message
        • Message.from_buffer()
        • Message.to_buffer()
    • testplan.common.utils.sockets.server module
      • Server
        • Server.accept_connection()
        • Server.bind()
        • Server.close()
        • Server.close_connection()
        • Server.host
        • Server.ip
        • Server.port
        • Server.receive()
        • Server.send()
        • Server.serve()
        • Server.socket
    • testplan.common.utils.sockets.tls module
      • DefaultTLSConfig
        • DefaultTLSConfig.get_context()
      • SimpleTLSConfig
        • SimpleTLSConfig.get_context()
      • TLSConfig
        • TLSConfig.get_context()
  • Module contents

Submodules

testplan.common.utils.callable module

Utilities related to python callables (functions, methods, classes etc.)

class testplan.common.utils.callable.ArgSpec(args, varargs, keywords, defaults)

Bases: tuple

args

Alias for field number 0

defaults

Alias for field number 3

keywords

Alias for field number 2

varargs

Alias for field number 1

testplan.common.utils.callable.arity(function)

Return the arity of a function

Parameters:

function (function) – function

Returns:

arity of the function

Return type:

int

testplan.common.utils.callable.getargspec(callable_)

Return an Argspec for any callable object

Parameters:

callable (Callable) – a callable object

Returns:

argspec for the callable

Return type:

ArgSpec

testplan.common.utils.callable.post(*postfunctions)

Attaches function(s) to another function for systematic execution after said function, with the same arguments

Parameters:

postfunction (callable) – function to execute after

Returns:

function decorator

Return type:

callable

testplan.common.utils.callable.pre(*prefunctions)

Attaches function(s) to another function for systematic execution before said function, with the same arguments

Parameters:

prefunction (callable) – function to execute before

Returns:

function decorator

Return type:

callable

testplan.common.utils.callable.wraps(wrapped, assigned=('__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__', '__tags__', '__tags_index__', 'summarize', 'summarize_num_passing', 'summarize_num_failing'), updated=('__dict__',))

Custom wraps function that copies some additional attr than default.

testplan.common.utils.comparison module

class testplan.common.utils.comparison.And(*callables)

Bases: MetaCallable

delimiter = 'and'

class testplan.common.utils.comparison.Callable

Bases: object

Some of our assertions can make use of callables that accept a single argument as comparator values. We also provide the helper classes below that are composable (via bitwise operators or meta callables) and reporting friendly.

class testplan.common.utils.comparison.Category

Bases: object

Internal enum. Categorises objects for comparison

ABSENT = 0

CALLABLE = 2

DICT = 5

ITERABLE = 4

REGEX = 3

VALUE = 1

class testplan.common.utils.comparison.Custom(callable_obj, description)

Bases: Callable

Utility that allows attaching descriptions to arbitrary functions.

Useful if you are making use of lambda functions and want to provide more context in the reports.

Usage:

Custom(
    callable_obj=lambda value: value.custom_method() is True,
    description='`value.custom_method()` returns True'
)

class testplan.common.utils.comparison.DictmatchAllResult(passed, index_match_levels)

Bases: object

When cast to a bool, evaluates to True when all values were matched without errors or False if one or more values mis-matched.

This object exposes two fields:

• passed: a boolean indicating if the assertion passed completely

• index_match_levels: a list containing tuples of index and match level:

  • MATCH

  • MISMATCH

  • LHS_NONE

  • RHS_NONE

The following are examples of what the fields return under various scenarios:

+-----------------------------------------+--------------------------+
|    DICTMATCH ASSERTION INPUT            |   DictmatchAllResult     |
+====================+====================+=========+================+
| Expected (LHS)     | Actual   (RHS)     | .passed | match levels   |
+--------------------+--------------------+---------+----------------+
| [{'id':0,'x':'a'}, | [{'id':0,'x':'a'}, |         | [(0,MATCH),    |
|  {'id':1,'x':'b'}, |  {'id':2,'x':'c'}, | True    |  (2,MATCH),    |
|  {'id':2,'x':'c'}] |  {'id':1,'x':'b'}] |         |  (1,MATCH)]    |
+--------------------+--------------------+---------+----------------+
| [{'id':0,'x':'a'}, | [{'id':0,'x':'a'}, |         | [(0,MATCH),    |
|  {'id':1,'x':'b'}, |  {'id':2,'x':'c'}, | False   |  (2,MATCH),    |
|  {'id':2,'x':'c'}] |  {'id':1}]         |         |  (1,MISMATCH)] |
+--------------------+--------------------+---------+----------------+
| [{'id':0,'x':'a'}, | [{'id':0,'x':'a'}, |         | [(0,MATCH),    |
|  {'id':1,'x':'b'}, |  {'id':3,'x':'d'}, | False   |  (3,LHS_NONE), |
|  {'id':2,'x':'c'}] |  {'id':1,'x':'b'}, |         |  (1,MATCH),    |
|                    |  {'id':2,'x':'c'}] |         |  (2,MATCH)]    |
+--------------------+--------------------+---------+----------------+
| [{'id':0,'x':'a'}, | [{'id':0,'x':'a'}, |         | [(0,MATCH),    |
|  {'id':1,'x':'b'}, |  {'id':1,'x':'b'}, | False   |  (1,MATCH),    |
|  {'id':2,'x':'c'}, |  {'id':3,'x':'d'}] |         |  (3,MATCH),    |
|  {'id':3,'x':'d'}] |                    |         |  (2,RHS_NONE)] |
+--------------------+--------------------+---------+----------------+

Indices are to be read as mappings from RHS values to LHS values. i.e.:

[(1, ..),(0, ..),(2, ..)]

maps: RHS:0 -> LHS:1, RHS:0 -> LHS:1, RHS:2 -> LHS:2.

LHS_NONE = 2

A value is present on the right hand side but not matched with a value on the left hand side. (e.g. an unexpected message). If any entry in index_match_levels is LHS_NONE, then passed is False.

MATCH = 0

Perfect match between identified and expected value. If all index_match_levels are MATCH, then passed is True.

MISMATCH = 1

The identified and expected values are matched with some errors. If any entry in index_match_levels is MISMATCH, then passed is False.

RHS_NONE = 3

A value is present on the left hand side but not matched with a value on the right hand side. (e.g. a missed message) If any entry in index_match_levels is RHS_NONE, then passed is False.

class testplan.common.utils.comparison.Equal(reference)

Bases: OperatorCallable

func(a, b, /)

Same as a == b.

func_repr = '=='

class testplan.common.utils.comparison.Expected(value, ignore=None, include=None)

Bases: object

An object representing an expected message, along with additional comparison flags.

Input to the “unordered_compare” function.

class testplan.common.utils.comparison.Greater(reference)

Bases: OperatorCallable

func(a, b, /)

Same as a > b.

func_repr = '>'

class testplan.common.utils.comparison.GreaterEqual(reference)

Bases: OperatorCallable

func(a, b, /)

Same as a >= b.

func_repr = '>='

class testplan.common.utils.comparison.In(container)

Bases: Callable

class testplan.common.utils.comparison.IsFalse

Bases: IsTrue

class testplan.common.utils.comparison.IsTrue

Bases: Callable

class testplan.common.utils.comparison.Less(reference)

Bases: OperatorCallable

func(a, b, /)

Same as a < b.

func_repr = '<'

class testplan.common.utils.comparison.LessEqual(reference)

Bases: OperatorCallable

func(a, b, /)

Same as a <= b.

func_repr = '<='

class testplan.common.utils.comparison.Match

Bases: object

Internal enum. Represents the result of a match.

FAIL = 'f'

IGNORED = 'i'

PASS = 'p'

static combine(lhs_match, rhs_match)

Combines to match levels into a single match level

static from_bool(passed)

Constructs a match description from a boolean value

static to_bool(match)

Converts a match value to a bool

class testplan.common.utils.comparison.MetaCallable(*callables)

Bases: Callable

delimiter = None

class testplan.common.utils.comparison.Not(callable_obj)

Bases: Callable

class testplan.common.utils.comparison.NotEqual(reference)

Bases: OperatorCallable

func(a, b, /)

Same as a != b.

func_repr = '!='

class testplan.common.utils.comparison.NotIn(container)

Bases: In

class testplan.common.utils.comparison.OperatorCallable(reference)

Bases: Callable

Base class for simple operator based callables.

func = None

func_repr = None

class testplan.common.utils.comparison.Or(*callables)

Bases: MetaCallable

delimiter = 'or'

class testplan.common.utils.comparison.RegexAdapter

Bases: object

This is being used for internal compatibility.

classmethod check(obj)

static compare(lhs, rhs)

Compare two regular expressions - just do string equality.

classmethod match(regex, value)

classmethod serialize(obj)

class testplan.common.utils.comparison.ReportOptions(*values)

Bases: Enum

Options to control reporting behaviour for comparison results:

• ALL: report all comparisons.

• NO_IGNORED: do not report comparisons of ignored keys, include everything else.

• FAILS_ONLY: only report comparisons that have failed.

Control of reporting behaviour is provided for two main reasons. Firstly, to give control of what information is included in the final report. Secondly, as an optimization to allow comparison information to be discarded when comparing very large collections.

ALL = 1

FAILS_ONLY = 3

NO_IGNORED = 2

testplan.common.utils.comparison.basic_compare(first, second, strict=False)

Comparison used for custom match functions, can do pattern matching, function evaluation or simple equality.

Returns traceback if something goes wrong.

testplan.common.utils.comparison.check_dict_keys(data, has_keys=None, absent_keys=None)

Check if a dictionary contains given keys and/or has given keys missing.

testplan.common.utils.comparison.compare(lhs: ~typing.Dict, rhs: ~typing.Dict, ignore: ~typing.List[~typing.Hashable] = None, include: ~typing.List[~typing.Hashable] = None, report_mode=ReportOptions.ALL, value_cmp_func: ~typing.Callable[[~typing.Any, ~typing.Any], bool] = <built-in function eq>, include_only_rhs: bool = False) → Tuple[bool, List[Tuple]]

Compare two iterable key, value objects (e.g. dict or dict-like mapping) and return a status and a detailed comparison table, useful for reporting.

Ignore has precedence over include.

Parameters:

• lhs – object compared against rhs

• rhs – object compared against lhs

• ignore – list of keys to ignore in the comparison

• include – list of keys to exclusively consider in the comparison

• report_mode – Specify which comparisons should be kept and reported. Default option is to report all comparisons but this can be restricted if desired. See ReportOptions enum for more detail.

• value_cmp_func – function to compare values in a dict. Defaults to COMPARE_FUNCTIONS[‘native_equality’].

• include_only_rhs – use the keys present in rhs.

Returns:

Tuple of comparison bool (passed: True, failed: False) and a description object for the testdb report

testplan.common.utils.comparison.compare_with_callable(callable_obj, value)

testplan.common.utils.comparison.dictmatch_all_compat(match_name, comparisons, values, description, key_weightings, value_cmp_func=<built-in function eq>)

This is being used for internal compatibility.

testplan.common.utils.comparison.is_comparator(value)

Utility for finding out a value is a custom comparator or not.

testplan.common.utils.comparison.is_regex(obj)

Cannot do type check against SRE_Pattern, so we use duck typing.

testplan.common.utils.comparison.tuplefy_comparisons(comparisons, table=False)

Convert dictionary report comparisons to list and tuples composition.

testplan.common.utils.comparison.tuplefy_item(item, list_entry=False)

Convert a dictionary report item in order to consume less space in json representation.

testplan.common.utils.comparison.unordered_compare(match_name, values, comparisons, description=None, tag_weightings=None, value_cmp_func=<built-in function eq>)

Matches a list of expected values against a list of expected comparisons.

The values and comparisons may be specified in any order, and the returned value represents the best overall match between values and comparisons.

Initially all value/expected comparison combinations are evaluated and converted to an error weight.

If certain keys/tags are more imporant than others (e.g. ID FIX tags), it is possible to give them additional weighting during the comparison, by specifying a “tag_weightings” dict.

The values/comparisons permutation that results in the least error is then returned as a list of dicts that can be included in the testing report.

Note

It is possible to specify up to a maximum of 16 values or expected comparisons.

Note

len(values) and len(comparison) need not be the same.

Parameters:

• match_name (str) – name that will appear on comparison report descriptions. For example “fixmatch” will produce a comparison description such as “unordered fixmatch 2/3: expected[2] vs values[1]”

• values (generator or list of dict-like objects) – Actual values: an iterable object (e.g. list or generator) of values. Each value needs to support a dict-like interface.

• comparisons (list of Expected) – Expected values and comparison flags.

• description (str) – Message used in each reported match.

• tag_weightings (dict of str to int) – Per-key overrides that specify a different weight for different keys.

Returns:

A list of test reports that can be appended to the result object

Return type:

list of dict (keys: ‘comparison’, ‘time’, ‘description’, ‘passed’, ‘objdisplay’)

testplan.common.utils.comparison.untyped_fixtag(x, y)

Custom stringify logic for fix msg tag value, strips off insignificant trailing 0s when converting float, so that 0.0 can be compared with ‘0’

testplan.common.utils.context module

TODO.

class testplan.common.utils.context.ContextValue(driver: str, value: str)

Bases: object

A context value represents a combination of a driver name and a Jinja2 template, to be resolved on driver start.

testplan.common.utils.context.context(driver, value)

Create a context extractor from a driver name and a value expression. Value expressions must be valid Jinja2 templates, which will be resolved from the context.

testplan.common.utils.context.expand(value, contextobj, constructor=None)

Take a value and a context and return the expanded result. Apply a constructor if necessary.

testplan.common.utils.context.is_context(value)

Checks if a value is a context value :param value: Value which may have been constructed through context :type value: object

Returns:

True if it is a context value, otherwise False

Return type:

bool

testplan.common.utils.context.parse_template(template: str) → Template | Template

testplan.common.utils.context.render(template: Template | Template | str, context) → str

Renders the template with the given context, that used for expression resolution.

Parameters:

• template (Union[Template, TempitaTemplate, str]) – A template in str, Jinja2 or Tempita form that will be rendered with the context

• context – The context object which is the base context of the template expansion

Returns:

The rendered template

Return type:

str

testplan.common.utils.convert module

Conversion utilities.

testplan.common.utils.convert.delta_decode_level(hetero)

testplan.common.utils.convert.delta_encode_level(homo)

testplan.common.utils.convert.expand_values(rows: List[Tuple], level: int = 0, ignore_key: bool = False, key_path: List | None = None, match: str = '')

Recursively expands and yields all rows of items to display.

Parameters:

• rows – comparison results

• level – recursive parameter for level of nesting

• ignore_key – recursive parameter for ignoring a key

• key_path – recursive parameter to build the sequence of keys

• match – recursive parameter for inheriting match result

Returns:

rows used in building comparison result table

testplan.common.utils.convert.flatten_dict_comparison(comparison: List[Tuple]) → List[List]

Flatten the comparison object from dictionary match into a tabular format.

Parameters:

comparison – list of comparison results

Returns:

result table to be used in display

testplan.common.utils.convert.flatten_formatted_object(formatted_obj)

Flatten the formatted object which is the result of function testplan.common.utils.reporting.fmt.

Parameters:

formatted_obj – The formatted object

Returns:

List representation of flattened object

Return type:

list

testplan.common.utils.convert.make_iterables(values: Iterable) → List[List | Tuple]

Create a list of lists and tuples for each of the values.

Parameters:

values – an iterable of values

Returns:

list containing one list and tuple for each value

testplan.common.utils.convert.make_tuple(value: object, convert_none: bool = False) → Tuple | object

Converts a value into a tuple.

Parameters:

• value – value to make the tuple out of

• convert_none – whether to convert None

Returns:

the value or the value converted to a tuple

testplan.common.utils.convert.nested_groups(iterable: Iterable, key_funcs: Sequence[Callable]) → List[Tuple | Tuple[List[Tuple | Tuple[RecursiveListTuple]]]]

Creates nested groups from the given iterable using key_funcs

Parameters:

• iterable – iterable of items

• key_funcs – key functions to sort by, applied in a waterfall

Returns:

recursively nested groups of items sorted by key functions

testplan.common.utils.convert.sort_and_group(iterable: Iterable, key: Callable) → List[Tuple]

Sorts an iterable and groups the items by the given key function.

Parameters:

• iterable – iterable of items

• key – key function to sort by

Returns:

groups of items sorted by key

testplan.common.utils.difflib module

This file is base on the difflib from python standard library (version: 2.7.9) it provides diff (context/unified) functions with more options like GNU diff, including: –ignore-space-change, –ignore-whitespace, –ignore-blank-lines Due to the different algorithm, its output might be a slightly difference compared with that of gnu diff or windiff, but it won’t lead to confusing.

Module difflib – helpers for computing deltas between objects.

Function get_close_matches(word, possibilities, n=3, cutoff=0.6):

Use SequenceMatcher to return list of the best “good enough” matches.

Function context_diff(a, b):

For two lists of strings, return a delta in context diff format.

Function diff(a, b):

Return a delta: the difference between a and b (lists of strings).

Function unified_diff(a, b):

For two lists of strings, return a delta in unified diff format.

Class SequenceMatcher:

A flexible class for comparing pairs of sequences of any type.

Class Differ:

For producing human-readable deltas from sequences of lines of text.

class testplan.common.utils.difflib.Differ(linejunk=None, charjunk=None, ignore_space_change=False, ignore_whitespaces=False, ignore_blank_lines=False)

Bases: object

Differ is a class for comparing sequences of lines of text, and producing human-readable differences or deltas. Differ uses SequenceMatcher both to compare sequences of lines, and to compare sequences of characters within similar (near-matching) lines.

Use get_opcodes() and get_merged_opcodes() to get a list of 5-tuples describing how to turn a into b. The former can give detailed transformation steps, especially the ‘replace’ operation for similar lines will be recorded, it is userful for ndiff (not implemented in this file), the latter gives concise transformation steps by merging adjacent op items if they can be merged. get_grouped_opcodes() is used for context_diff() and unified_diff().

Example: Comparing two texts.

>>> a = ['aaa\n', 'bbb\n', 'c\n', 'cc\n', 'ccc\n', '\n', 'ddd\n',
... 'eee\n', 'ggg\n']
>>> b = ['aaaa\n', 'bbbb\n', 'c\n', 'cc\n', 'ccc\n', 'dddd\n', 'hhh\n',
... 'fff\n', '\n', 'ggg\n']
>>> d = Differ()
>>> for op in d.get_opcodes(a, b): print(op)
...
('replace', 0, 1, 0, 1)
('replace', 1, 2, 1, 2)
('equal', 2, 5, 2, 5)
('delete', 5, 6, 5, 5)
('replace', 6, 7, 5, 6)
('replace', 7, 8, 6, 9)
('equal', 8, 9, 9, 10)
>>> for op in d.get_merged_opcodes(a, b): print(op)
...
('replace', 0, 2, 0, 2)
('equal', 2, 5, 2, 5)
('replace', 5, 8, 5, 9)
('equal', 8, 9, 9, 10)
>>> for op in d.get_grouped_opcodes(a, b, 1): print(op)
...
[('replace', 0, 2, 0, 2), ('equal', 2, 3, 2, 3)]
[('equal', 4, 5, 4, 5), ('replace', 5, 8, 5, 9), ('equal', 8, 9, 9, 10)]

Note that when instantiating a Differ object we may pass functions to filter out line and character ‘junk’. See Differ.__init__ for details.

get_grouped_opcodes(a, b, n=3)

Isolate change clusters by eliminating ranges with no changes.

Return a generator of groups with up to n lines of context. Each group is in the same format as returned by get_opcodes().

>>> from pprint import pprint
>>> a = map(str, range(1,40))
>>> b = a[:]
>>> b[8:8] = ['i']     # Make an insertion
>>> b[20] += 'x'       # Make a replacement
>>> b[23:28] = []      # Make a deletion
>>> b[30] += 'y'       # Make another replacement
>>> pprint(list(SequenceMatcher(None,a,b).get_grouped_opcodes()))
[[('equal', 5, 8, 5, 8),
  ('insert', 8, 8, 8, 9),
  ('equal', 8, 11, 9, 12)],
 [('equal', 16, 19, 17, 20),
  ('replace', 19, 20, 20, 21),
  ('equal', 20, 22, 21, 23),
  ('delete', 22, 27, 23, 23),
  ('equal', 27, 30, 23, 26)],
 [('equal', 31, 34, 27, 30),
  ('replace', 34, 35, 30, 31),
  ('equal', 35, 38, 31, 34)]]

get_merged_opcodes(a, b)

Similar like get_opcodes(), but the adjacent items might be merge as one, for example:

(‘equal’, 2, 9, 0, 7) (‘equal’, 9, 11, 7, 9) (‘replace’, 11, 12, 9, 10) (‘replace’, 12, 13, 10, 11) (‘replace’, 13, 14, 11, 12) will be merged as: (‘equal’, 2, 11, 0, 9) (‘replace’, 11, 14, 9, 12)

Another example:

(‘delete’, 11, 12, 10, 10) (‘replace’, 12, 13, 10, 11) will be merged as: (‘replace’, 11, 13, 10, 11)

get_opcodes(a, b)

Compare two sequences of lines; generate the resulting delta.

Each sequence must contain individual single-line strings ending with newlines. Such sequences can be obtained from the readlines() method of file-like objects. The delta generated also consists of newline- terminated strings, ready to be printed as-is via the writeline() method of a file-like object.

Example:

>>> for op in Differ().get_opcodes('one\ntwo\nthree\n',
...                                'ore\nthree\nemu\n'):
...    print(op)
...
('replace', 0, 2, 0, 1)
('equal', 2, 3, 1, 2)
('insert', 3, 3, 2, 3)

testplan.common.utils.difflib.IS_CHARACTER_JUNK(ch, ws=' \t')

Return 1 for ignorable character: iff ch is a space or tab.

Examples:

>>> IS_CHARACTER_JUNK(' ')
True
>>> IS_CHARACTER_JUNK('\t')
True
>>> IS_CHARACTER_JUNK('\n')
False
>>> IS_CHARACTER_JUNK('x')
False

testplan.common.utils.difflib.IS_LINE_JUNK(line, pat=<built-in method match of re.Pattern object>)

Return 1 for ignorable line: iff line is blank or contains a single ‘#’.

Examples:

>>> IS_LINE_JUNK('\n')
True
>>> IS_LINE_JUNK('  #   \n')
True
>>> IS_LINE_JUNK('hello\n')
False

class testplan.common.utils.difflib.Match(a, b, size)

Bases: tuple

a

Alias for field number 0

b

Alias for field number 1

size

Alias for field number 2

class testplan.common.utils.difflib.SequenceMatcher(isjunk=None, a='', b='', autojunk=False)

Bases: object

SequenceMatcher is a flexible class for comparing pairs of sequences of any type, so long as the sequence elements are hashable. The basic algorithm predates, and is a little fancier than, an algorithm published in the late 1980’s by Ratcliff and Obershelp under the hyperbolic name “gestalt pattern matching”. The basic idea is to find the longest contiguous matching subsequence that contains no “junk” elements (R-O doesn’t address junk). The same idea is then applied recursively to the pieces of the sequences to the left and to the right of the matching subsequence. This does not yield minimal edit sequences, but does tend to yield matches that “look right” to people.

SequenceMatcher tries to compute a “human-friendly diff” between two sequences. Unlike e.g. UNIX(tm) diff, the fundamental notion is the longest contiguous & junk-free matching subsequence. That’s what catches peoples’ eyes. The Windows(tm) windiff has another interesting notion, pairing up elements that appear uniquely in each sequence. That, and the method here, appear to yield more intuitive difference reports than does diff. This method appears to be the least vulnerable to synching up on blocks of “junk lines”, though (like blank lines in ordinary text files, or maybe “<P>” lines in HTML files). That may be because this is the only method of the 3 that has a concept of “junk” <wink>.

Example, comparing two strings, and considering blanks to be “junk”:

>>> s = SequenceMatcher(lambda x: x == " ",
...                     "private Thread currentThread;",
...                     "private volatile Thread currentThread;")
>>>

.ratio() returns a float in [0, 1], measuring the “similarity” of the sequences. As a rule of thumb, a .ratio() value over 0.6 means the sequences are close matches:

>>> print(round(s.ratio(), 3))
0.866
>>>

If you’re only interested in where the sequences match, .get_matching_blocks() is handy:

>>> for block in s.get_matching_blocks():
...     print("a[%d] and b[%d] match for %d elements" % block)
a[0] and b[0] match for 8 elements
a[8] and b[17] match for 21 elements
a[29] and b[38] match for 0 elements

Note that the last tuple returned by .get_matching_blocks() is always a dummy, (len(a), len(b), 0), and this is the only case in which the last tuple element (number of elements matched) is 0.

If you want to know how to change the first sequence into the second, use .get_opcodes():

>>> for opcode in s.get_opcodes():
...     print("%6s a[%d:%d] b[%d:%d]" % opcode)
 equal a[0:8] b[0:8]
insert a[8:8] b[8:17]
 equal a[8:29] b[17:38]

See the Differ class for a fancy human-friendly file differencer, which uses SequenceMatcher both to compare sequences of lines, and to compare sequences of characters within similar (near-matching) lines.

See also function get_close_matches() in this module, which shows how simple code building on SequenceMatcher can be used to do useful work.

Timing: Basic R-O is cubic time worst case and quadratic time expected case. SequenceMatcher is quadratic time for the worst case and has expected-case behavior dependent in a complicated way on how many elements the sequences have in common; best case time is linear.

Methods:

__init__(isjunk=None, a=’’, b=’’)

Construct a SequenceMatcher.

set_seqs(a, b)

Set the two sequences to be compared.

set_seq1(a)

Set the first sequence to be compared.

set_seq2(b)

Set the second sequence to be compared.

find_longest_match(alo, ahi, blo, bhi)

Find longest matching block in a[alo:ahi] and b[blo:bhi].

get_matching_blocks()

Return list of triples describing matching subsequences.

get_opcodes()

Return list of 5-tuples describing how to turn a into b.

ratio()

Return a measure of the sequences’ similarity (float in [0,1]).

quick_ratio()

Return an upper bound on .ratio() relatively quickly.

real_quick_ratio()

Return an upper bound on ratio() very quickly.

find_longest_match(alo, ahi, blo, bhi)

Find longest matching block in a[alo:ahi] and b[blo:bhi].

If isjunk is not defined:

Return (i,j,k) such that a[i:i+k] is equal to b[j:j+k], where

alo <= i <= i+k <= ahi blo <= j <= j+k <= bhi

and for all (i’,j’,k’) meeting those conditions,

k >= k’ j <= j’ and if j == j’, i <= i’

In other words, of all maximal matching blocks, return one that starts earliest in b, and of all those maximal matching blocks that start earliest in b, return the one that starts earliest in a.

>>> s = SequenceMatcher(None, " abcd", "abcd abcd")
>>> s.find_longest_match(0, 5, 0, 9)
Match(a=0, b=4, size=5)

If isjunk is defined, first the longest matching block is determined as above, but with the additional restriction that no junk element appears in the block. Then that block is extended as far as possible by matching (only) junk elements on both sides. So the resulting block never matches on junk except as identical junk happens to be adjacent to an “interesting” match.

Here’s the same example as before, but considering blanks to be junk. That prevents “ abcd” from matching the “ abcd” at the tail end of the second sequence directly. Instead only the “abcd” can match, and matches the leftmost “abcd” in the second sequence:

>>> s = SequenceMatcher(lambda x: x==" ", " abcd", "abcd abcd")
>>> s.find_longest_match(0, 5, 0, 9)
Match(a=1, b=0, size=4)

If no blocks match, return (alo, blo, 0).

>>> s = SequenceMatcher(None, "ab", "c")
>>> s.find_longest_match(0, 2, 0, 1)
Match(a=0, b=0, size=0)

get_grouped_opcodes(n=3)

Isolate change clusters by eliminating ranges with no changes.

Return a generator of groups with up to n lines of context. Each group is in the same format as returned by get_opcodes().

>>> from pprint import pprint
>>> a = map(str, range(1,40))
>>> b = a[:]
>>> b[8:8] = ['i']     # Make an insertion
>>> b[20] += 'x'       # Make a replacement
>>> b[23:28] = []      # Make a deletion
>>> b[30] += 'y'       # Make another replacement
>>> pprint(list(SequenceMatcher(None, a, b).get_grouped_opcodes()))
[[('equal', 5, 8, 5, 8),
  ('insert', 8, 8, 8, 9),
  ('equal', 8, 11, 9, 12)],
 [('equal', 16, 19, 17, 20),
  ('replace', 19, 20, 20, 21),
  ('equal', 20, 22, 21, 23),
  ('delete', 22, 27, 23, 23),
  ('equal', 27, 30, 23, 26)],
 [('equal', 31, 34, 27, 30),
  ('replace', 34, 35, 30, 31),
  ('equal', 35, 38, 31, 34)]]

get_matching_blocks()

Return list of triples describing matching subsequences.

Each triple is of the form (i, j, n), and means that a[i:i+n] == b[j:j+n]. The triples are monotonically increasing in i and in j. New in Python 2.5, it’s also guaranteed that if (i, j, n) and (i’, j’, n’) are adjacent triples in the list, and the second is not the last triple in the list, then i+n != i’ or j+n != j’. IOW, adjacent triples never describe adjacent equal blocks.

The last triple is a dummy, (len(a), len(b), 0), and is the only triple with n==0.

>>> s = SequenceMatcher(None, "abxcd", "abcd")
>>> s.get_matching_blocks()
[Match(a=0, b=0, size=2),
 Match(a=3, b=2, size=2),
 Match(a=5, b=4, size=0)]

get_opcodes()

Return list of 5-tuples describing how to turn a into b.

Each tuple is of the form (tag, i1, i2, j1, j2). The first tuple has i1 == j1 == 0, and remaining tuples have i1 == the i2 from the tuple preceding it, and likewise for j1 == the previous j2.

The tags are strings, with these meanings:

‘replace’: a[i1:i2] should be replaced by b[j1:j2] ‘delete’: a[i1:i2] should be deleted.

Note that j1==j2 in this case.

‘insert’: b[j1:j2] should be inserted at a[i1:i1].

Note that i1==i2 in this case.

‘equal’: a[i1:i2] == b[j1:j2]

>>> a = "qabxcd"
>>> b = "abycdf"
>>> s = SequenceMatcher(None, a, b)
>>> for tag, i1, i2, j1, j2 in s.get_opcodes():
...    print("%7s a[%d:%d] (%s) b[%d:%d] (%s)" %
...          (tag, i1, i2, a[i1:i2], j1, j2, b[j1:j2]))
 delete a[0:1] (q) b[0:0] ()
  equal a[1:3] (ab) b[0:2] (ab)
replace a[3:4] (x) b[2:3] (y)
  equal a[4:6] (cd) b[3:5] (cd)
 insert a[6:6] () b[5:6] (f)

quick_ratio()

Return an upper bound on ratio() relatively quickly.

This isn’t defined beyond that it is an upper bound on .ratio(), and is faster to compute.

ratio()

Return a measure of the sequences’ similarity (float in [0,1]).

Where T is the total number of elements in both sequences, and M is the number of matches, this is 2.0*M / T. Note that this is 1 if the sequences are identical, and 0 if they have nothing in common.

.ratio() is expensive to compute if you haven’t already computed .get_matching_blocks() or .get_opcodes(), in which case you may want to try .quick_ratio() or .real_quick_ratio() first to get an upper bound.

>>> s = SequenceMatcher(None, "abcd", "bcde")
>>> s.ratio()
0.75
>>> s.quick_ratio()
0.75
>>> s.real_quick_ratio()
1.0

real_quick_ratio()

Return an upper bound on ratio() very quickly.

This isn’t defined beyond that it is an upper bound on .ratio(), and is faster to compute than either .ratio() or .quick_ratio().

set_seq1(a)

Set the first sequence to be compared.

The second sequence to be compared is not changed.

>>> s = SequenceMatcher(None, "abcd", "bcde")
>>> s.ratio()
0.75
>>> s.set_seq1("bcde")
>>> s.ratio()
1.0
>>>

SequenceMatcher computes and caches detailed information about the second sequence, so if you want to compare one sequence S against many sequences, use .set_seq2(S) once and call .set_seq1(x) repeatedly for each of the other sequences.

See also set_seqs() and set_seq2().

set_seq2(b)

Set the second sequence to be compared.

The first sequence to be compared is not changed.

>>> s = SequenceMatcher(None, "abcd", "bcde")
>>> s.ratio()
0.75
>>> s.set_seq2("abcd")
>>> s.ratio()
1.0
>>>

SequenceMatcher computes and caches detailed information about the second sequence, so if you want to compare one sequence S against many sequences, use .set_seq2(S) once and call .set_seq1(x) repeatedly for each of the other sequences.

See also set_seqs() and set_seq1().

set_seqs(a, b)

Set the two sequences to be compared.

>>> s = SequenceMatcher()
>>> s.set_seqs("abcd", "bcde")
>>> s.ratio()
0.75

testplan.common.utils.difflib.context_diff(a, b, n=3, ignore_space_change=False, ignore_whitespaces=False, ignore_blank_lines=False)

Compare two sequences of lines; generate the delta as a context diff.

Context diffs are a compact way of showing line changes and a few lines of context. The number of context lines is set by ‘n’ which defaults to three.

By default, the diff control lines (those with *** or —) are created with a trailing newline. This is helpful so that inputs created from file.readlines() result in diffs that are suitable for file.writelines() since both the inputs and outputs have trailing newlines.

The context diff format normally has a header for filenames and modification times. Here, “a.text” and “b.text” are displayed instead of file names, and current UTC time instead of the modification time.

• ignore_space_change: refer to gnu diff option -b

• ignore_whitespaces: refer to gnu diff option -w

• ignore_blank_lines: refer to gnu diff option -B

Example:

>>> difference = context_diff('one\ntwo\nthree\n'.splitlines(True),
                              'ore\nthree\nemu\n'.splitlines(True))
>>> print(''.join(difference))
--- a.text    2018-08-28 11:40:17 UTC
+++ b.text    2018-08-28 11:40:17 UTC
***************
*** 1,3 ****
! one
! two
  three
--- 1,3 ----
! ore
  three
+ emu

testplan.common.utils.difflib.diff(a, b, ignore_space_change=False, ignore_whitespaces=False, ignore_blank_lines=False, unified=False, context=False)

Compare two blocks of text or two sequences of lines and generate delta as a normal/unified/context diff. Lines that only contain whitespaces have lower priority to get matched.

If a or b is a string, then is will be split as list of strings which are separated by ‘n’, ‘rn’ or ‘r’, while keep the line terminator.

By default, the function generates the delta as a normal diff, but you can specify a unified diff or context diff. And you can directly set the parameter unified or context to a positive integer, that means the number of context lines, which defaults to three.

• ignore_space_change: refer to gnu diff option -b

• ignore_whitespaces: refer to gnu diff option -w

• ignore_blank_lines: refer to gnu diff option -B

• unified: if True, output in unified format, default False

• context: if True, output in context format, default False

testplan.common.utils.difflib.get_close_matches(word, possibilities, n=3, cutoff=0.6)

Use SequenceMatcher to return list of the best “good enough” matches.

word is a sequence for which close matches are desired (typically a string).

possibilities is a list of sequences against which to match word (typically a list of strings).

Optional arg n (default 3) is the maximum number of close matches to return. n must be > 0.

Optional arg cutoff (default 0.6) is a float in [0, 1]. Possibilities that don’t score at least that similar to word are ignored.

The best (no more than n) matches among the possibilities are returned in a list, sorted by similarity score, most similar first.

>>> get_close_matches("appel", ["ape", "apple", "peach", "puppy"])
['apple', 'ape']
>>> import keyword as _keyword
>>> get_close_matches("wheel", _keyword.kwlist)
['while']
>>> get_close_matches("apple", _keyword.kwlist)
[]
>>> get_close_matches("accept", _keyword.kwlist)
['except']

testplan.common.utils.difflib.unified_diff(a, b, n=3, ignore_space_change=False, ignore_whitespaces=False, ignore_blank_lines=False)

Compare two sequences of lines; generate the delta as a unified diff.

Unified diffs are a compact way of showing line changes and a few lines of context. The number of context lines is set by ‘n’ which defaults to three.

By default, the diff control lines (those with —, +++, or @@) are created with a trailing newline. This is helpful so that inputs created from file.readlines() result in diffs that are suitable for file.writelines() since both the inputs and outputs have trailing newlines.

The unidiff format normally has a header for filenames and modification times. Here, “a.text” and “b.text” are displayed instead of file names, and current UTC time instead of the modification time.

• ignore_space_change: refer to gnu diff option -b

• ignore_whitespaces: refer to gnu diff option -w

• ignore_blank_lines: refer to gnu diff option -B

Example:

>>> difference = unified_diff('one\ntwo\nthree\n'.splitlines(True),
                              'ore\nthree\nemu\n'.splitlines(True))
>>> print(''.join(difference))
--- a.text    2018-08-28 11:36:46 UTC
+++ b.text    2018-08-28 11:36:46 UTC
@@ -1,3 +1,3 @@
-one
-two
+ore
 three
+emu

testplan.common.utils.exceptions module

Utilities for exception handling.

testplan.common.utils.exceptions.should_raise(exception, item, args=None, kwargs=None, pattern=None)

“Utility that validates callable should raise specific exception.

Parameters:

• exception (Exception) – Exception should be raised.

• item (callable) – Callable that should raise.

• args (tuple) – Callable args.

• kwargs (dict) – Callable kwargs.

• pattern (Compiled regex) – Compiled regex pattern that needs to match the exception.

testplan.common.utils.exceptions.suppress_exception(logger=<TestplanLogger testplan.common.utils.exceptions (WARNING)>)

Suppress & log exceptions for the given function.

This is mostly used during exporters steps, as we would like to return the original retcode from testplan run, without raising any non-test-related errors.

testplan.common.utils.helper module

This module provides helper functions that will add common information of Testplan execution to test report.

They could be used directly in testcases or provided to pre/pose_start/stop hooks.

Also provided is a predefined testsuite that can be included in user’s Multitest directly.

class testplan.common.utils.helper.DriverLogCollector(name: str = 'DriverLogCollector', description: str = 'logs', ignore: List[str] = None, file_pattern: List[str] = None, recursive: bool = True, failure_only: bool = True)

Bases: object

Customizable file collector class used for collecting driver logs.

Parameters:

• name – Name of the object shown in the report.

• description – Text description for the assertion.

• ignore – List of patterns of file name to ignore when attaching a directory.

• file_pattern – List of patterns of file name to include when attaching a directory. (Defaults: “stdout*”, “stderr*”)

• recursive – Recursively traverse sub-directories and attach all files, default is to only attach files in top directory.

• failure_only – Only collect files on failure.

class testplan.common.utils.helper.TestplanExecutionInfo

Bases: object

Utility testsuite to log generic information of Testplan execution.

environment(env, result)

Environment

get_testcases()

Return the unbound method objects marked as a testcase from a testsuite class.

hardware(env, result)

Host hardware

logging(env, result)

Testplan log

name = None

path(env, result)

Execution path

strict_order = False

testplan.common.utils.helper.attach_driver_logs_if_failed(env: Environment, result: Result) → None

Attaches stdout and stderr files to the report for each driver.

Parameters:

• env – environment

• result – testcase result

testplan.common.utils.helper.attach_log(result: Result) → None

Attaches top-level testplan.log file to the report.

Parameters:

result – testcase result

testplan.common.utils.helper.clean_runpath_if_passed(env: Environment, result: Result) → None

Deletes multitest-level runpath if the multitest passed.

Parameters:

• env – environment

• result – result object

testplan.common.utils.helper.get_hardware_info() → Dict

Return a variety of host hardware information.

Returns:

dictionary of hardware information

testplan.common.utils.helper.log_cmd(result: Result) → None

Saves command line arguments to the report.

Parameters:

result – testcase result

testplan.common.utils.helper.log_environment(result: Result) → None

Saves host environment variable to the report.

Parameters:

result – testcase result

testplan.common.utils.helper.log_hardware(result: Result) → None

Saves host hardware information to the report.

Parameters:

result – testcase result

testplan.common.utils.helper.log_pwd(result: Result) → None

Saves current path to the report.

Parameters:

result – testcase result

testplan.common.utils.interface module

Validates methods signature.

exception testplan.common.utils.interface.MethodSignatureMismatch

Bases: Exception

MethodSignatureMismatch Exception

exception testplan.common.utils.interface.NoSuchMethodInClass

Bases: Exception

NoSuchMethodInClass Exception

testplan.common.utils.interface.check_signature(func: callable, args_list: list | str) → bool

Checks if the given function’s signature matches the given list of args

Parameters:

• func – function whose signature to check

• args_list – list of arg names to match as signature

Returns:

True if the signature is matching

Raises:

MethodSignatureMismatch – if the given function’s signature differs from the provided

testplan.common.utils.logger module

This module provides an interface from testplan to the standard python logging module. It defines some extra logging levels and handles setting up separate handlers for logging to stdout and to a file under the runpath.

The logging facility is used for:

• Overall Testplan status

• Driver status updates & logs

• Worker & pool messages

• Test progress information (e.g. Pass / Fail status)

• Exporter statuses

class testplan.common.utils.logger.Loggable

Bases: object

Base class that allows an object to log via self.logger. The advantage of objects having their own logger over using a single global logger is that the logger name can be used to identify which class the logs originate from. Loggers are hierarchical so logs made from a child will bubble up to the parent - all logs will be ultimately handled by the root testplan logger and its associated handlers.

property logger: TestplanLogger

logger object

class testplan.common.utils.logger.TestplanLogger(*args, **kwargs)

Bases: Logger

Custom Logger class for Testplan. Adds extra logging level and corresponding method for USER_INFO.

LEVELS = {'CRITICAL': 50, 'DEBUG': 10, 'ERROR': 40, 'INFO': 20, 'USER_INFO': 25, 'WARNING': 30}

log_test_status(name: str, status: Status, indent: int = 0, level: int = 25)

Shortcut to log a pass/fail status for a test.

user_info(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘USER_INFO’

testplan.common.utils.logger.configure_file_logger(level, runpath)

Configure the file logger.

Parameters:

• level (int) – Logging level - should be one of the values in TestplanLogger.LEVELS.

• runpath (str) – top-level runpath - the log file will be created in here.

testplan.common.utils.match module

Module of utility types and functions that perform matching.

class testplan.common.utils.match.LogMatcher(log_path: PathLike | str, binary: bool = False)

Bases: Loggable

Single line matcher for text files (usually log files). Once matched, it remembers the line number of the match and subsequent matches are scanned from the current line number. This can be useful when matched lines are not unique for the entire log file. Support simple cases of log rotation

expect(regex: str | bytes | Pattern, timeout: float = 5.0)

Context manager as a composite of seek_eof() and match(). On entering seeking to log stream EOF, on exiting doing log matching, as expected pattern should be (indirectly) produced by context manager body.

Parameters:

• regex – Regex string or compiled regular expression.

• timeout – Timeout in seconds as a float for regex matching.

get_between(mark1=None, mark2=None)

Returns the content of the file from the start marker to the end marker. It is possible to omit either marker to receive everything from start to end of file.

Note

Since markers point to the byte position immediately after match, this function will not return what was matched for mark1, but will return the contents of what was matched for mark2.

Parameters:

• mark1 (str) – mark name of start position (None for beginning of file)

• mark2 (str) – mark name of end position (None for end of file)

Returns:

The content between mark1 and mark2.

Return type:

str

mark(name: str)

Marks the current file position with the specified name. The mark name can later be used to set the file position

Parameters:

name – Name of the mark.

match(regex: str | bytes | Pattern, timeout: float = 5.0, raise_on_timeout: bool = True) → Match | None

Matches each line in the log file from the current line number to the end of the file. If a match is found the line number is stored and the match is returned. By default an exception is raised if no match is found.

Parameters:

• regex – Regex string or compiled regular expression (re.compile)

• timeout – Timeout in seconds to wait for matching process, 0 means matching till EOF and not waiting for new lines, any value greater than 0 means doing matching up to such seconds, defaults to 5 seconds

• raise_on_timeout – To raise TimeoutException or not

Returns:

The regex match or None if no match is found

match_all(regex: str | bytes | Pattern, timeout: float = 5.0, raise_on_timeout: bool = True) → List[Match]

Similar to match, but returns all occurrences of regex. By default an exception is raised if no match is found.

Parameters:

• regex – Regex string or compiled regular expression (re.compile)

• timeout – Timeout in seconds to find out all matches in file, defaults to 5 seconds.

• raise_on_timeout – To raise TimeoutException or not

Returns:

A list of regex matches

match_between(regex: str | bytes | Pattern, mark1: str, mark2: str)

Matches file against passed in regex. Matching is performed from file position denoted by mark1 and ends before file position denoted by mark2. If a match is not found then None is returned.

Parameters:

• regex – regex string or compiled regular expression (re.compile)

• mark1 – mark name of start position (None for beginning of file)

• mark2 – mark name of end position

Returns:

The regex match or None if no match is found

not_match(regex: str | bytes | Pattern, timeout: float = 5.0)

Opposite of match() which raises an exception if a match is found. Matching is performed from the current file position. If match is not found within timeout period then no exception is raised.

Parameters:

• regex – Regex string or compiled regular expression (re.compile)

• timeout – Timeout in seconds to wait for matching process, 0 means should not wait and return whatever matched on initial scan, defaults to 5 seconds

not_match_between(regex: str | bytes | Pattern, mark1: str, mark2: str)

Opposite of match_between() which returns None if a match is not found. Matching is performed from file position denoted by mark1 and ends before file position denoted by mark2. If a match is found then False is returned otherwise True.

Parameters:

• regex – regex string or compiled regular expression (re.compile)

• mark1 – mark name of start position (None for beginning of file)

• mark2 – mark name of end position

seek(mark: str | None = None)

Sets current file position to the specified mark. The mark has to exist. If the mark is None sets current file position to beginning of file.

Parameters:

mark – Name of the mark.

seek_eof()

Sets current file position to the current end of file.

seek_sof()

Sets current file position to the start of file.

class testplan.common.utils.match.ScopedLogfileMatch(log_matcher: LogMatcher, regex: str | bytes | Pattern, timeout: float = 5.0)

Bases: object

testplan.common.utils.match.match_regexps_in_file(logpath: PathLike, log_extracts: List[Pattern]) → Tuple[bool, Dict[str, str], List[Pattern]]

Return a boolean, dict pair indicating whether all log extracts matches, as well as any named groups they might have matched.

Parameters:

• logpath – Log file path.

• log_extracts – Regex list.

Returns:

Match result.

testplan.common.utils.networking module

Utility functions related to networking

testplan.common.utils.networking.get_hostname_access_url(port, url_path)

testplan.common.utils.networking.port_to_int(port)

Convert a port string to an integer.

testplan.common.utils.parser module

Arguments parsing utilities.

class testplan.common.utils.parser.ArgMixin

Bases: object

Utility mixin that can be used with Enums for cmdline arg parsing.

Supports:

• ‘kebab-case’ <-> ‘CONSTANT_NAME’ conversion.

• Custom parser logic that displays all available options on failure.

• Pretty help text rendering for each option. (need to be used with argparse.RawTextHelpFormatter)

classmethod enum_to_str(enm)

EnumClass.MY_ENUM -> my-enum

classmethod get_descriptions()

Override this method to return a dictionary with Enums as keys and description strings as values.

This will later on be rendered via –help command.

classmethod get_help_text(default)

Render help text in the ‘arg-value’: ‘description’ format.

classmethod get_parser_context(default=None, **kwargs)

Shortcut method for populating Argparse.parser.add_argument params.

classmethod parse(arg)

Get the enum for given cmdline arg in string form, display all available options when an invalid value is parsed.

classmethod str_to_enum(val)

my-enum -> EnumClass.MY_ENUM

testplan.common.utils.path module

Dirs/file path utilities.

class testplan.common.utils.path.StdFiles(directory)

Bases: object

stderr and stdout file creation and management

close() → None

Close fds

open_err(mode: str = 'r') → TextIOWrapper

Open the stderr file with the defined access

open_out(mode: str = 'r') → TextIOWrapper

Open the stdout file with the defined access

testplan.common.utils.path.archive(path, timestamp)

Append a timestamp to an existing file’s name.

Parameters:

• path (str) – Path to a file that should be archived

• timestamp (str) – timestamp

Returns:

path to the archived file

Return type:

str

testplan.common.utils.path.change_directory(directory)

A context manager that changes working directory and returns to original on exit.

Parameters:

directory (str) – Directory to change into.

testplan.common.utils.path.default_runpath(entity)

Returns default runpath for an Entity object.

testplan.common.utils.path.fix_home_prefix(path)

Try to replace a real path (/a/path/user) with a symlink path (/symlink/path/user), with clue from userhome and current working directory.

testplan.common.utils.path.hash_file(filepath)

Hashes the contents of a file. SHA1 algorithm is used.

Parameters:

filepath (str) – Path to file to hash.

Returns:

Hashed value as a string

Return type:

str

testplan.common.utils.path.instantiate(template, values, destination)

Instantiate a templated file with a set of values and place it in a target destination.

Parameters:

• template (str) – the path to the templated file

• values (dict) – the context dict to be used when instantiating the templated file

• destination (str) – the path to the destination directory/file to write the instantiated templated file to

Returns:

None

Return type:

NoneType

testplan.common.utils.path.is_subdir(child, parent)

Check whether “parent” is a sub-directory of “child”.

Parameters:

• child (str) – Child path.

• parent (str) – Parent directory to check against.

Returns:

True if child is a sub-directory of the parent.

Return type:

bool

testplan.common.utils.path.makedirs(path)

A trivial wrapper for os.makedirs that doesn’t raise when the directory already exists.

Parameters:

path (str) – Path to be created.

testplan.common.utils.path.makeemptydirs(path)

Make an empty directory at a location.

Parameters:

path (str) – Path to be created.

testplan.common.utils.path.module_abspath(module)

Returns file path of a python module.

testplan.common.utils.path.pwd()

Working directory path.

testplan.common.utils.path.rebase_path(path, old_base, new_base)

Rebase path from old_base to new_base and convert to Linux form.

testplan.common.utils.path.removeemptydir(path)

Remove a directory if it does exist and is empty.

Parameters:

path (str) – Path to be created.

testplan.common.utils.path.traverse_dir(directory, topdown=True, ignore=None, only=None, recursive=True, include_subdir=True)

Recursively traverse all files and sub directories in a directory and get a list of relative paths.

Parameters:

• directory (str) – Path to a directory that will be traversed.

• topdown (bool) – Browse the directory in a top-down or bottom-up approach.

• ignore (list) – List of patterns to ignore by glob style filtering.

• only (list) – List of patterns to include by glob style filtering.

• recursive (bool) – Traverse directories recursively, set to False to only list items in top directory.

• include_subdir (bool) – Include all sub directories and files if True, or exclude directories in the result.

Returns:

A list of relative file paths

Return type:

list of str

testplan.common.utils.path.unique_name(name, names)

Takes a file or directory name and set of other file or directory names

Returns a new unique name if it exists already in the specifed set, or the original if it does not exist in the set.

The names set is unaffected. The new name is generated by appending a suffix before the extention (if any).

For example 'stdout.log' becomes 'stdout-1.log', then 'stdout-2.log', 'stdout-3.log', etc.

Parameters:

• name (str) – A file or directory name (not path) that may or may not be unique

• names (set of str) – A set of names (not paths), not modified by this function

Returns:

Either the same, or a new unique name

Return type:

str

testplan.common.utils.process module

System process utilities module.

class testplan.common.utils.process.LogDetailsOption(*values)

Bases: Enum

LOG_ALWAYS = 1

LOG_ON_ERROR = 2

NEVER_LOG = 3

testplan.common.utils.process.cleanup_child_procs(procs: List[Process], timeout: float, log: Callable[[BaseException], None]) → None

Kill child processes to eliminate possibly orphaned processes.

Parameters:

• procs – List of child processes to kill.

• timeout – Timeout in seconds, defaults to 5 seconds.

• log – Callback function to log exceptions.

testplan.common.utils.process.enforce_timeout(process, timeout=1, callback=None, output=None)

testplan.common.utils.process.execute_cmd(cmd, label=None, check=True, stdout=None, stderr=None, logger=None, env=None, detailed_log: LogDetailsOption = LogDetailsOption.LOG_ON_ERROR)

Execute a subprocess command.

Parameters:

• cmd – Command to execute - list of parameters.

• label – Optional label for debugging

• check – When True, check that the return code of the command is 0 to ensure success - raises a RuntimeError otherwise. Defaults to True - should be explicitly disabled for commands that may legitimately return non-zero return codes.

• stdout – Optional file-like object to redirect stdout to.

• stderr – Optional file-like object to redirect stderr to.

• logger – Optional logger object as logging destination.

• env – Optional dict object as environment variables.

• detailed_log – Enum to determine when stdout and stderr outputs should be logged. LOG_ALWAYS - Outputs are logged on success and failure. LOG_ON_ERROR - Outputs are logged on failure. NEVER_LOG - Outputs are never logged.

Returns:

Return code of the command.

testplan.common.utils.process.kill_proc_and_child_procs(proc: Popen, child_procs: List[Process], log: Callable[[BaseException], None] = None, timeout: int = 5) → None

Kill a process and its child processes.

This function attempts to kill a given process and its child processes. It first kills the main process, waits for it to terminate, and then attempts to kill any remaining child processes.

Parameters:

• proc (subprocess.Popen) – The main process to kill.

• child_procs (List[psutil.Process]) – A list of child processes to kill.

• log (Callable[[BaseException], None], optional) – A callable to log exceptions, defaults to None.

• timeout (int, optional) – The timeout in seconds to wait for processes to terminate, defaults to 5.

Returns:

The return code of the main process if it was terminated, otherwise None.

Return type:

Union[int, None]

testplan.common.utils.process.kill_process(proc: Popen, timeout: int = 5, signal_: Signals | None = None, output: IO | None = None, on_failed_termination: Callable[[int, int], None] | None = None) → int | None

If alive, kills the process. First call terminate() or pass signal_ if specified to terminate for up to time specified in timeout parameter. If process hangs then call kill().

Parameters:

• proc – process to kill

• timeout – timeout in seconds, defaults to 5 seconds

• output – Optional file like object for writing logs.

:param on_failed_terminationcallable or None

A callback function that is executed when process fails to terminate after the timeout. When supplied, this callback will be executed after SIGTERM fails and before SIGKILL. It receives two arguments: pid as int and timeout as int and can be leveraged to collect additional diagnostic info about the process.

Returns:

Exit code of process

testplan.common.utils.process.kill_process_psutil(proc: Process, timeout: int = 5, signal_: Signals | None = None, output: IO | None = None, on_failed_termination: Callable[[int, int], None] | None = None) → List[Process]

If alive, kills the process (an instance of psutil.Process). Try killing the child process at first and then killing itself. First call terminate() or pass signal_ if specified to terminate for up to time specified in timeout parameter. If process hangs then call kill().

Parameters:

• proc – process to kill

• timeout – timeout in seconds, defaults to 5 seconds

• output – Optional file like object for writing logs.

:param on_failed_terminationcallable or None

A callback function that is executed when process fails to terminate after the timeout. When supplied, this callback will be executed after SIGTERM fails and before SIGKILL. It receives two arguments: pid as int and timeout as int and can be leveraged to collect additional diagnostic info about the process.

Returns:

List of processes which are still alive

testplan.common.utils.process.process_is_alive(proc_id: int) → bool

testplan.common.utils.process.subprocess_popen(args, bufsize=0, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=False, startupinfo=None, creationflags=0)

Wrapper for Subprocess.Popen, which defaults close_fds=True on Linux. It’s the behaviour we nearly always want, and which has become default in 3.2+.

On Windows, closed_fds=False.

testplan.common.utils.process.wait_process_clean(proc_id: int, timeout: int = 5, output: IO = None)

testplan.common.utils.registry module

Provides Registry mapping utility.

class testplan.common.utils.registry.Registry

Bases: Loggable

A utility that provides a decorator (@registry.bind) for mapping objects to another (decorated) class.

Supports absolute or category based defaults via @registry.bind_default decorator as well.

Example:

>>> registry = Registry()

>>> class ClassA:
    ... pass

>>>  # instances of ClassA are now bound to ClassB for this registry
>>> @registry.bind(ClassA)
>>> class ClassB:
    ... pass

>>> obj_a = ClassA()

>>> registry[obj_a] is ClassB
... True

bind(*classes)

Decorator for binding one or more classes to another.

Parameters:

classes – One or more classes that will be bound to the decorated class.

bind_default(category=None)

Decorator for binding a class as category based or absolute default.

Parameters:

category – (optional) If provided, the decorated class will be the default for the given category, otherwise it will be the absolute default.

property default

get_category(obj)

Override this to define logic for generating the category key from the object instance.

get_lookup_key(obj)

This method is used for generating the key when do a lookup from the registry. Object class is used by default.

get_record_key(obj)

This method is used for generating the key when we bind an object (possibly a class) via the registry.

testplan.common.utils.remote module

Remote execution utilities.

testplan.common.utils.remote.copy_cmd(source, target, exclude=None, port=None, deref_links=False)

Returns remote copy command.

testplan.common.utils.remote.filepath_exist_cmd(path)

Checks if filepath exists.

testplan.common.utils.remote.link_cmd(path, link)

Returns link creation command.

testplan.common.utils.remote.mkdir_cmd(path)

Return mkdir command

testplan.common.utils.remote.rm_cmd(path)

Return rm command

testplan.common.utils.remote.ssh_bin()

testplan.common.utils.remote.ssh_cmd(ssh_cfg, command)

Prefix command with ssh binary and option.

Parameters:

• ssh_cfg – dict with “host” and “port” (optional) keys

• command – command to execute on remote host

Returns:

full cmd list

testplan.common.utils.remote.worker_is_remote(remote_host)

Check remote_host is not the same as current.

testplan.common.utils.reporting module

This module contains utilities that are mostly used to make assertion comparison data report friendly.

class testplan.common.utils.reporting.AbsentType

Bases: object

A singleton to represent the lack of a value in a comparison. None is not used to avoid the situation where a key may be present and it’s value is None.

descr = 'ABSENT'

testplan.common.utils.reporting.callable_name(callable_obj)

Extract the name of a callable object

Parameters:

callable_obj (Any object as long as it is callable) – Callable object

Returns:

Either the function name or the name of the class of the callable

Return type:

str

testplan.common.utils.reporting.fmt(obj)

Recursively formats an object as plain old data.

Parameters:

obj – The object to format

Returns:

The plain old data representation of “obj” that can be serialised to JSON

Return type:

object or a (object, object) pair

testplan.common.utils.strings module

String manipulation utilities.

class testplan.common.utils.strings.Color

Bases: object

Utility class with shortcuts for colored console output.

static colored(msg, color)

static green(msg)

static passed(msg=None, check=None)

static red(msg)

static yellow(msg)

testplan.common.utils.strings.format_description(description)

Get rid of empty lines and unindent multiline text until the leftmost indented line has no whitespaces on the left.

In:

(assume dots represent whitespace)

….Hello world ……Foo bar <EMPTY-LINE> ….1 2 3 4

Out:

Hello World ..Foo bar 1 2 3 4

testplan.common.utils.strings.get_docstring(obj)

Get object docstring without leading whitespace. :param obj: Object to be extracted docstring. :type obj: object :return: Docstring of the object. :rtype: str or NoneType

testplan.common.utils.strings.indent(lines_str, indent_size=2)

Indent a multi-line string with a common indent.

Parameters:

• lines_str (str) – Multi-line string.

• indent_size (int) – Number of spaces to indent by - defaults to 2.

Returns:

New string with extra indent.

Return type:

str

testplan.common.utils.strings.map_to_str(value)

Convert bytes to str byte-by-byte

testplan.common.utils.strings.slugify(value)

Normalizes string, converts to lowercase, removes non-alpha characters, and converts spaces to hyphens.

Parameters:

value (str) – string value to slugify

Returns:

slugified string value, suitable as a directory or filename

Return type:

str

testplan.common.utils.strings.to_bytes(value, encoding='utf-8', errors='strict')

Coerce a string to bytes type.

Parameters:

• value (str or bytes) – A string to be converted

• encoding (str) – Encoding method

• errors (str) – Error handling scheme

Returns:

Converted byte string.

Return type:

bytes

testplan.common.utils.strings.to_str(value, encoding='utf-8', errors='strict')

Coerce a string to str type.

Parameters:

• value (str or bytes) – A string to be converted

• encoding (str) – Encoding method

• errors (str) – Error handling scheme

Returns:

Converted unicode string.

Return type:

str

testplan.common.utils.strings.uuid4()

Generate a globally unique id.

Returns:

A string complied with uuid.uuid4 format

Return type:

str

testplan.common.utils.strings.wrap(text, width=150)

Wraps text within given width limit, keeping initial indentation of each line (and generated lines). Useful for wrapping exception messages.

Parameters:

• text – Text to be wrapped.

• width – Maximum character limit for each line.

Returns:

Wrapped text

testplan.common.utils.table module

Utilities for working with tables.

class testplan.common.utils.table.TableEntry(source, placeholder=None)

Bases: object

Represents a table. Internally represented either as a list of list or a list of dict. — Note tuples are counted as lists here, callers should be able to handle them.

as_list_of_dict()

Returns the table as list of dict

Returns:

the table

Return type:

list of dict for the table

as_list_of_list()

Returns the table as list of list

Returns:

the table

Return type:

list of list for the table

property columns

Get column names.

testplan.common.utils.testing module

This module contains utilites for testing Testplan itself.

class testplan.common.utils.testing.FixMessage(*args, **kwargs)

Bases: OrderedDict

Basic FIX message for testing. A FIX message may be either typed or untyped. In other respects is acts like a plain dict.

copy()

Override copy() to return another FixMessage, preserving the typed_values attribute.

class testplan.common.utils.testing.XMLComparison(tag, children=None, **kwargs)

Bases: object

Testing utility for generated XML file contents.

Recursively compares children as well, supports simple string or regex matching.

Usage:

my_file.xml

<root foo="bar">
    <parent id="0"/>
    <parent id="1">
        <child hello="world" time="12:00"/>
    </parent>
</root>

comparison = XMLComparison(
    tag='root',
    foo='bar',
    children=[
        XMLComparison(tag='parent', id='0'),
        XMLComparison(
            tag='parent', id='1'
            children=[
                XMLComparison(
                    tag='child', hello='world',
                    time=re.compile('\d{2}:\d{2}')
                )
            ]
        ),

    ]
)

with open('my_file.xml') as xml_file:
    comparison.compare(xml_file.read())

compare(xml_str, encoding='utf-8')

Compare with xml string input.

testplan.common.utils.testing.argv_overridden(*override_ctx)

Override sys.argv for the given context. This is not a thread safe operation, may cause issues if you run tests in parallel threads.

testplan.common.utils.testing.captured_logging(logger, level=20)

Utility for capturing a logger object’s output at a specific level, with a default level of INFO. Useful for command line output testing.

testplan.common.utils.testing.check_entry(expected, actual)

Utility function for comparing serialized entries.

testplan.common.utils.testing.check_iterable(expected, actual, curr_path='ROOT', _orig_exp=None, _orig_act=None)

Utility for checking an iterable, supports custom func assertions along with normal value matches.

testplan.common.utils.testing.check_report(expected, actual, skip=None)

Utility function for comparing report objects.

Skip uid attribute, entries will be checked recursively via check_entry.

testplan.common.utils.testing.check_report_context(report, ctx)

Utility function for checking filtered/ordered test results, we are not interested in report contents, just the existence of reports with matching names, with the correct order.

testplan.common.utils.testing.context_wrapper(ctx_manager, *ctx_args, **ctx_kwargs)

Higher order function that returns a decorator that runs the wrapped func within the context of the given ctx_manager initialized by ctx_args and ctx_kwargs

testplan.common.utils.testing.disable_log_propagation(logger)

Disables log propagation for the given logger.

testplan.common.utils.testing.log_level_changed(logger, level)

Change log level for the given logger.

testplan.common.utils.testing.log_propagation_disabled(logger)

Disables log propagation for the given logger.

WARNING: Use this logic sparingly as it will hide actual exceptions from getting displayed in the console.

A use case would be testing out exception logging to a custom target without showing these messages in the console itself, leaving us with a clean console output.

testplan.common.utils.testing.override_argv(*override_ctx)

Override sys.argv for the wrapped function.

testplan.common.utils.testing.suppress_warnings(func)

Suppress warnings within a function

testplan.common.utils.testing.to_stdout(*items)

Utility function that can be used for testing logging output along with captured_logging.

testplan.common.utils.testing.warnings_suppressed()

Suppress warnings within a block

testplan.common.utils.thread module

Threading utilities.

class testplan.common.utils.thread.Barrier(n)

Bases: object

Implements a re-usable, two-phase barrier. Allows a fixed number of threads to wait for each other to reach a certain point.

For python >= 3.2 you can just use threading.Barrier instead, this class is provided for compatibility with Python 2.

Parameters:

n (int) – Number of threads to wait for at the barrier.

wait()

Wait for all threads to reach the barrier before returning.

testplan.common.utils.thread.execute_as_thread(target, args=None, kwargs=None, daemon=False, join=True, break_join=None, join_sleep=0.01, timeout=None)

Execute target callable in a separate thread.

Parameters:

• target (callable) – Target callable.

• args (tuple) – Callable args.

• kwargs (kwargs) – Callable kwargs.

• daemon (bool) – Set daemon thread.

• join (bool) – Join thread before return.

• break_join (callable) – Condition for join early break.

• join_sleep (int) – Join break condition check sleep time.

• timeout (TimeoutException) – Timeout duration.

testplan.common.utils.thread.interruptible_join(thread, timeout=None)

Joining a thread without ignoring signal interrupts.

Parameters:

• thread (threading.Thread) – Thread object to wait to terminate.

• timeout (Optional[numbers.Number]) – If specified, TimeoutException will be raised if the thread does not terminate within the specified timeout.

testplan.common.utils.timing module

Time related utilities.

class testplan.common.utils.timing.Interval(start, end)

Bases: _Interval

Class that represents a block of time.

property elapsed

Return duration in seconds.

class testplan.common.utils.timing.KThread(*args, **kwargs)

Bases: Thread

A subclass of threading.Thread, with a kill() method.

globaltrace(frame, event, arg)

kill()

localtrace(frame, event, arg)

start()

Start the thread.

exception testplan.common.utils.timing.TimeoutException

Bases: Exception

Timeout exception error.

class testplan.common.utils.timing.TimeoutExceptionInfo(start_time=None)

Bases: object

Holds timeout exception information.

msg()

Return a message to be used by TimeoutException containing timing information.

class testplan.common.utils.timing.Timer

Bases: dict

Dict wrapper with a method for recording durations.

end(key)

Record the end timestamp for the given key.

first(key)

last(key)

merge(timer)

record(key)

Records duration for the given key.

>>> timer = Timer()
>>> with timer.record('my-key'):
>>>  ... custom code ...
>>>  ... custom code ...
>>> timer.last(key='my-key').elapsed
21.5

start(key)

Record the start timestamp for the given key.

class testplan.common.utils.timing.TimerCtxManager(timer, key)

Bases: object

Context manager for storing durations. Uses tz aware utc timestamps.

testplan.common.utils.timing.exponential_interval(initial: float = 0.1, multiplier: float = 2, maximum: float | None = None, minimum: float | None = None) → Generator[float, None, None]

Generator that returns exponentially increasing/decreasing values, can be used for generating values for time.sleep for periodic checks.

Parameters:

• initial (number) – Initial value for the sequence.

• multiplier (number) – Multiplier for generating new values in the sequence. Each new value will be generated by multiplication of the multiplier and the last generated value of the sequence.

• minimum (number) – Optional minimum value for generated numbers.

• maximum (number) – Optional maximum value for generated numbers.

Returns:

Sequence of values

Return type:

generator of number

testplan.common.utils.timing.format_duration(duration: int) → str

Format seconds in hours / minutes / seconds in readable format.

>>> format_duration(3730)
1 hours 2 minutes 10 seconds

Parameters:

duration (number) – Total duration in seconds

Returns:

Duration in readable format.

Return type:

str

testplan.common.utils.timing.get_sleeper(interval: float | Tuple[float, float], timeout: float = 10, raise_timeout_with_msg: str | Callable[[], str] | None = None, timeout_info: bool = False) → Generator[bool, None, None]

Generator that implements sleep steps for replacing while True: do task; time.sleep() code blocks. Depending on the interval argument, it can sleeps with constant interval or start with min_interval and then doubles the interval in each iteration up to max_interval.

It yields True until timeout is reached where it then yields False or raises a TimeoutException based on input arguments.

Parameters:

• interval (float or tuple of float as (min_interval, max_interval)) – Sleep time between each yield in seconds.

• timeout (float) – Timeout in seconds

• raise_timeout_with_msg (NoneType or str or callable) – Message or Function to be used for raising an optional TimeoutException.

• timeout_info (bool) – Include timeout exception timing information in exception message raised.

testplan.common.utils.timing.iana_tz() → str

IANA TZ identifier

testplan.common.utils.timing.now() → datetime

Timezone aware local time.

testplan.common.utils.timing.parse_duration(duration: str) → int

Parse given duration string and return duration value in seconds.

Parameters:

duration (str) – Duration value in format <hours>H <minutes>M <seconds>S

Returns:

Duration in seconds

Return type:

int

testplan.common.utils.timing.retry_until_timeout(exception: Type[Exception], item: Callable[[...], Any], timeout: int, args: List[Any] = None, kwargs: Mapping[str, Any] = None, interval: float = 0.05, raise_on_timeout: bool = True) → Any

Retry calling an item until timeout duration while ignoring exceptions.

Parameters:

• exception (type) – Exception class to catch.

• item (callable) – Function to call.

• args (Optional[Iterable[Any]]) – Positional args to pass to item

• kwargs (Optional[Dict[str, Any]]) – Keyword args to pass to item

• interval (int) – time to wait between successive call attempts, in seconds.

• raise_on_timeout – Whether to raise a TimeoutException on timeout, defaults to True.

Returns:

Result of item.

Return type:

Any

testplan.common.utils.timing.timeout(seconds: int, err_msg: str = 'Timeout after {} seconds.') → Callable[[Callable], Callable]

Decorator for a normal function to limit its execution time.

Parameters:

• seconds (int) – Time limit for task execution.

• err_msg (str) – Error message on timeout.

Returns:

Decorated function.

Return type:

callable

testplan.common.utils.timing.utcnow() → datetime

Timezone aware UTC now.

testplan.common.utils.timing.wait(predicate: Callable[[], bool], timeout: int, interval: float = 0.05, raise_on_timeout: bool = True) → bool

Wait until a predicate evaluates to True.

Parameters:

• predicate (callable) – Input predicate.

• timeout (int) – Timeout duration.

• interval (float) – Sleep interval for predicate check.

• raise_on_timeout (bool) – Raise exception if hits timeout, defaults to True.

Returns:

Predicate result.

Return type:

bool

testplan.common.utils.timing.wait_until_predicate(predicate: Callable[[], bool], timeout: int, interval: float = 1.0)

Inverting wait() method behavior to raise if predicate() is True instead of raising on timeout.

Parameters:

• predicate (callable) – any callable object

• timeout (float) – timeout in seconds

• interval (float) – interval at which to check the predicate in seconds

Raises:

RuntimeError if the predicate is True.

testplan.common.utils.timing.with_timer(func: Callable) → Callable

Decorator to add function execution as event to timer

testplan.common.utils.validation module

This module contains helper validation functions to be used with configuration schemas.

testplan.common.utils.validation.has_method(method_name)

Validator that checks if a given class has method with the given name

testplan.common.utils.validation.is_subclass(parent_kls)

Validator for subclass check.

When we have a class as a value in a validation schema, schema.py will implicitly try to do an isinstance check on the dict to be validated.

Using this function will allow us to do issubclass checks.

testplan.common.utils.validation.is_valid_email(email)

Validator that checks if an email is valid

testplan.common.utils.validation.is_valid_url(url)

Validator that checks if a url is valid

Module contents

Common utility modules.