avocado_i2n.cartgraph.node module

Utility for the main test suite substructures like test nodes.

SUMMARY

Copyright: Intra2net AG

INTERFACE

class avocado_i2n.cartgraph.node.PrefixTreeNode(variant: str = None, parent: str = None)

Bases: object

A node of a prefix tree.

check_child(variant: str) → bool

Check child prefix tree node.

unset_child(variant: str) → None

Unset child prefix tree node.

traverse() → Generator[None, None, None]

Traverse the current node.

class avocado_i2n.cartgraph.node.PrefixTree

Bases: object

A trie structure used for faster prefix lookup.

insert(test_node: TestNode) → None

Insert a test node name in the prefix tree.

get(name: str) → list[TestNode]

Get all the names of the prefix tree.

class avocado_i2n.cartgraph.node.EdgeRegister

Bases: object

A register for the Cartesian graph edges allowing counter and worker stats extraction.

register(node: TestNode, worker: TestWorker) → None

Register a worker visit for the given (possibly bridged) test node.

Parameters:

• node – possibly registered test node to register visits for

• worker – worker that visited the test node

class avocado_i2n.cartgraph.node.TestNode(prefix: str, recipe: Reparsable)

Bases: Runnable

A wrapper for all test relevant parts.

These include parameters, parser, used objects and dependencies to/from other test nodes (setup/cleanup).

class ReadOnlyDict

Bases: dict[Any, Any]

Custom implementation of a read-only attribute of dictionary type.

pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem(*args: tuple[type, ...], **kwargs: dict[str, type]) → None

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

clear() → None.  Remove all items from D.

update([E, ]**F) → None.  Update D from dict/iterable E and F.

If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

setdefault(*args: tuple[type, ...], **kwargs: dict[str, type]) → None

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

prefix_pattern = re.compile('^(\\d+)([abcd]?)(.+)')

digit: 0 for object root, >0 for everything else letter: “a” (autosetup), “b” (byproduct), “c” (cleanup), “d” (duplicate)

property params: Params

Parameters (cache) property.

property shared_started_workers: set[TestWorker]

Workers that have previously started traversing this node (incl. leaves and others).

property shared_finished_workers: set[TestWorker]

Workers that have previously finished traversing this node (incl. leaves and others).

property shared_involved_workers: set[TestWorker]

Workers that picked up the node and possibly have continued to either its setup or cleanup.

property shared_results: list[dict[str, str]]

Test results shared across all bridged nodes.

property shared_filtered_results: list[dict[str, str]]

Test results shared across all bridged nodes.

property shared_result_worker_ids: set[str]

ID-s of workers that produced the shared results.

property bridged_nodes: tuple[TestNode]

Read-only list of bridged nodes.

property cloned_nodes: tuple[TestNode]

Read-only list of cloned nodes.

property setup_nodes: dict[TestNode, set[TestObject]]

Read-only dict of setup nodes.

property cleanup_nodes: dict[TestNode, set[TestObject]]

Read-only dict of cleanup nodes.

property setless_form: Params

Test set invariant form of the test node name.

property bridged_form: Params

Test worker invariant form of the test node name.

property long_prefix: Params

Sufficiently unique prefix to identify a diagram test node.

property id: Params

Use unique ID to identify a test node.

property id_test: Params

Use a unique test ID to identify a test node.

is_occupied(worker: TestWorker = None) → bool

Check if the test node is sufficiently occupied with respect to a given worker in various scopes.

Parameters:

worker – test worker with respect to which to consider various scopes

is_flat() → bool

Check if the test node is flat and does not yet have objects and dependencies to evaluate.

is_shared_root() → bool

Check if the test node is the root of all test nodes for all test objects.

is_object_root() → bool

Check if the test node is the root of all test nodes for some test object.

is_unrolled(worker: TestWorker = None) → bool

Check if the test is unrolled as composite node with dependencies.

Parameters:

worker – worker a flat node is unrolled for

Raises:

RuntimeError if the current node is not flat (cannot be unrolled)

is_setup_ready(worker: TestWorker) → bool

Check if all dependencies of the test were run or there were none.

Parameters:

worker – relative setup readiness with respect to a worker ID

is_cleanup_ready(worker: TestWorker) → bool

Check if all dependent tests were run or there were none.

Parameters:

worker (str) – relative setup readiness with respect to a worker ID

is_started(worker: TestWorker = None, threshold: int = 1) → bool

Check if the test is currently traversed by at least N (-1 for all) workers of all or some scopes.

Parameters:

• worker – evaluate with respect to an optional worker ID scope or globally if none given

• threshold – how eagerly the node is considered started in terms of number of required workers to use as a threshold (1 for most eagerly, -1 for most fully)

Returns:

whether the test was run by at least N workers of all or some scopes (N=threshold)

is_finished(worker: TestWorker = None, threshold: int = 1) → bool

Check if the test was ever traversed by at least N (-1 for all) workers of all or some scopes.

Parameters:

• worker – evaluate with respect to an optional worker ID scope or globally if none given

• threshold – how eagerly the node is considered started in terms of number of required workers to use as a threshold (1 for most eagerly, -1 for most fully)

Returns:

whether the test was run by at least one worker of all or some scopes

Threshold of 1 is the most eager manner so that any already available setup nodes are considered finished. If we instead wait for this setup to be cleaned up or synced, this would count most of the setup as finished in the very end of the traversal.

Threshold of -1 is for fully traversed node by all workers unless restricted within some scope of setup reuse.

should_parse(worker: TestWorker = None) → bool

Parse if node has been dropped in all its setup nodes by at least one worker.

Parameters:

worker – evaluate with respect to an optional worker ID scope or globally if none given

Returns:

whether the test node should be parsed

should_rerun(worker: TestWorker = None) → bool

Check if the test node should be rerun based on some retry criteria.

Parameters:

worker – evaluate with respect to an optional worker ID scope or globally if none given

Returns:

whether the test node should be retried

The retry parameters are max_tries and rerun_status or stop_status. The first is the maximum number of tries, and the second two indicate when to continue or stop retrying in terms of encountered test status and can be a list of statuses.

default_run_decision(worker: TestWorker) → bool

Set default decision policy on whether a test node should be run or skipped.

Parameters:

worker – worker which makes the run decision

Returns:

whether the worker should run the test node

default_clean_decision(worker: TestWorker) → bool

Set default decision policy on whether a test node should be cleaned or skipped.

Parameters:

worker – worker which makes the clean decision

Returns:

whether the worker should clean the test node

classmethod prefix_priority(prefix1: str, prefix2: str) → int

Class method for secondary prioritization using test prefixes.

Parameters:

• prefix1 – first prefix to use for the priority comparison

• prefix2 – second prefix to use for the priority comparison

Returns:

negative integer if prefix1 < prefix2, positive if prefix1 > prefix2, 0 otherwise (lower is better in our standard sorting)

This function also does recursive calls of sub-prefixes.

pick_parent(worker: TestWorker) → TestNode

Pick the next available parent based on some priority.

Parameters:

worker – worker for which the parent is selected

Returns:

the next parent node

Raises:

RuntimeError

The current order will prioritize less traversed test paths.

pick_child(worker: TestWorker) → TestNode

Pick the next available child based on some priority.

Parameters:

worker – worker for which the child is selected

Returns:

the next child node

Raises:

RuntimeError

The current order will prioritize less traversed test paths.

drop_parent(test_node: TestNode, worker: TestWorker) → None

Add a parent node to the set of visited nodes for this test.

Parameters:

• test_node – visited node

• worker – worker visiting the node

Raises:

ValueError if visited node is not directly dependent

drop_child(test_node: TestNode, worker: TestWorker) → None

Add a child node to the set of visited nodes for this test.

Parameters:

• test_node – visited node

• worker – worker visiting the node

Raises:

ValueError if visited node is not directly dependent

descend_from_node(test_node: TestNode, test_object: TestObject) → None

Turn the current node into a child of a parent node for a given object.

Parameters:

• test_node – parent node the current node is a child of

• test_object – test object via which the dependency is determined

bridge_with_node(test_node: TestNode) → None

Bridge current node with equivalent node for a different worker.

Parameters:

test_node – equivalent node for a different worker

Raises:

ValueError if bridged node is not equivalent

clone_as_source(test_nodes: list[TestNode]) → None

Convert the node to a clone source for a list of its clones.

Parameters:

test_nodes – clones to register as a clone source to

pull_locations() → None

Update all setup locations for the current node.

update_restrs(object_restrs: dict[str, str]) → None

Update any restrictions with further filters.

Parameters:

object_restrs – multi-line object restrictions to append

regenerate_params(verbose: bool = False) → None

Regenerate all parameters from the current reparsable config.

Parameters:

verbose (bool) – whether to show generated parameter dictionaries

regenerate_vt_parameters() → None

Regenerate the parameters provided to the VT runner.

scan_states() → bool

Scan for present object states to reuse the test from previous runs.

Returns:

whether all required states are available

sync_states(params: Params) → None

Sync or drop present object states to clean or later skip tests from previous runs.

validate() → None

Validate the test node for sane attribute-parameter correspondence.