testplan.runners package

Subpackages

testplan.runners.pools package

Submodules

testplan.runners.base module

Executor base classes.

class testplan.runners.base.Executor(**options)[source]

Bases: Resource

Receives items, executes them and create results.

Subclasses must implement the Executor._loop logic.

CONFIG: alias of ExecutorConfig

abort_dependencies() → Generator[source]: Abort items running before aborting self.

add(item: object, uid: str) → None[source]

Adds an item for execution.

Parameters:

item – To be executed and create a result.
uid – Unique id.

added_item(uid: str) → object[source]: Returns the added item.

property added_items: OrderedDict: Returns added items.

bubble_up_discard_tasks(exec_selector: Expr, report_status: Status = Status.NONE, report_reason: str = '')[source]

property class_name: str: Returns the class name.

discard_pending_tasks(report_status: Status = Status.NONE, report_reason: str = '')[source]

get(uid: str) → object[source]: Get item result by uid.

get_current_status_for_debug() → List[str][source]

Gets information about items in Executor for debugging. Subclasses can override this method and implement a well suited method to get items current status.

Returns:: Status of items in Executor.

property is_alive: bool: Poll the loop handler thread to check it is running as expected.

pending_work() → bool[source]: Resource has pending work.

property results: OrderedDict: Items results.

starting() → None[source]: Starts the execution loop.

stopping() → None[source]: Stop the executor.

class testplan.runners.base.ExecutorConfig(**options)[source]

Bases: ResourceConfig

Configuration object for Executor resource.

Inherits all ResourceConfig options.

testplan.runners.local module

Basic local executor.

class testplan.runners.local.LocalRunner(uid='local_runner', **options)[source]

Bases: Executor

Basic local execution that inherits Executor and accepts all ExecutorConfig options.

aborting() → None[source]: Aborting logic.

discard_pending_tasks(report_status: Status = Status.INCOMPLETE, report_reason: str = '')[source]

execute(uid: str) → TestResult[source]: Execute item implementation.