testplan.common.entity package

Submodules

testplan.common.entity.base module

Module containing base classes that represent object entities that can accept configuration, start/stop/run/abort, create results and have some state.

class testplan.common.entity.base.Entity(**options: Any)

Bases: Loggable

Base class for Entity and Resource objects providing common functionality like runpath creation, abort policy and common attributes.

Parameters:

• runpath (str or NoneType callable that returns str) – Path to be used for temp/output files by entity.

• path_cleanup (bool) – Remove previous runpath created dirs/files.

• status_wait_timeout (int) – Timeout for wait status events.

• abort_wait_timeout (int) – Timeout for entity abort.

• active_loop_sleep (float) – Sleep time on busy waiting loops.

CONFIG

alias of EntityConfig

STATUS

alias of EntityStatus

abort() → None

Default abort policy. First abort all dependencies and then itself.

abort_dependencies() → Generator[Entity, None, None]

Returns an empty generator.

property aborted: bool

Returns if entity was aborted.

aborting() → None

Aborting logic for self.

property active: bool

Entity not aborting/aborted.

property cfg: Config

Configuration object.

context_input(exclude: list | None = None) → Dict[str, Any]

All attr of self in a dict for context resolution

define_runpath() → None

Define runpath directory based on parent object and configuration.

classmethod filter_locals(local_vars: Dict[str, Any]) → Dict[str, Any]

Filter out init params of None value, they will take default value defined in its ConfigOption object; also filter out special vars that are not init params from local_vars.

Parameters:

local_vars

property logger: TestplanLogger

logger object

make_runpath_dirs() → None

Creates runpath related directories.

property parent: Entity | None

Returns parent Entity.

pause() → None

Pauses entity execution.

pausing() → None

property report: ReportLink

A handle to access the report via recursive parent

resume() → None

Resumes entity execution.

resuming() → None

property runpath: str

Path to be used for temp/output files by entity.

property scratch: str

Path to be used for temp files by entity.

property status: EntityStatus

Status object.

property timer: Timer

uid() → str

Unique identifier of self.

wait(target_status: str | None, timeout: int | None = None) → None

Wait until objects status becomes target status.

Parameters:

• target_status (str) – expected status

• timeout (int or NoneType) – timeout in seconds

class testplan.common.entity.base.EntityConfig(**options: Any)

Bases: Config

Configuration object for Entity object.

All classes that inherit Entity can define a configuration that inherits this one’s schema.

classmethod build_schema() → Schema

Build a validation schema using the config options defined in this class and its parent classes.

denormalize() → Config

Create new config object that inherits all explicit attributes from its parents as well.

get_local(name: str, default: Any = None) → Any

Returns a local config setting (not from container)

classmethod get_options() → Dict[Any, Any]

Config options for base Entity class.

ignore_extra_keys = False

property parent: Config | None

Returns the parent configuration.

set_local(name: str, value: Any) → None

set without any check

class testplan.common.entity.base.EntityStatus

Bases: object

Represents current status of an Entity object.

TODO: Utilise metadata to store information.

NONE = None

PAUSED = 'PAUSED'

PAUSING = 'PAUSING'

RESUMING = 'RESUMING'

change(new: str | None) → None

Transition to new status.

Parameters:

new (NoneType or str) – status to be set

clear_metadata() → None

Re-initializes metadata as empty.

property metadata: OrderedDict[str, Any]

TODO

reset() → None

Reset status as None.

property tag: str | None

Current status value.

transitions() → Dict[str | None, set]

Returns all legal transitions of the status of the Entity.

update_metadata(**metadata: Any) → None

Updates metadata.

Parameters:

metadata (OrderedDict) – additional metadata

class testplan.common.entity.base.Environment(parent: Entity | None = None)

Bases: object

A collection of resources that can be started/stopped.

Parameters:

parent (Entity) – Reference to parent object.

add(item: Resource, uid: str | None = None) → str

Adds a Resource to the Environment.

Parameters:

• item (Resource) – Resource to be added.

• uid (str or NoneType) – Unique identifier.

Returns:

Unique identifier assigned to item added.

Return type:

str

all_status(target: str | None) → bool

Checks whether all resources have target status.

Parameters:

target (str) – expected status

first() → str

Returns the UID of the first resource of the environment.

get(key: str, default: Resource | None = None) → Resource | None

items() → Iterator[Tuple[str, Resource]]

parent: Entity | None

remove(uid: str) → None

Removes resource with the given uid from the environment.

Parameters:

uid – Unique identifier.

start() → None

Starts all resources sequentially and log errors.

start_exceptions: OrderedDict[Resource, str]

start_in_pool(pool: ThreadPool, timeout: float | None = None) → None

Start all resources concurrently in thread pool and log exceptions.

Parameters:

pool – thread pool

stop(is_reversed: bool = False) → None

Stop all resources, optionally in reverse order, and log exceptions.

Parameters:

is_reversed – flag whether to stop resources in reverse order

stop_exceptions: OrderedDict[Resource, str]

stop_in_pool(pool: ThreadPool, timeout: float | None = None) → None

Stop all resources concurrently in thread pool and log exceptions.

Parameters:

pool – thread pool

sync_stop_resource(resource: Resource) → None

Stop a resource and log exceptions.

class testplan.common.entity.base.FailedAction(error_msg: str)

Bases: object

Simple falsy container that can be used for returning results of certain failed async actions.

The error_msg can later on be used for enriching the error messages.

class testplan.common.entity.base.ReportLink(name: str)

Bases: object

A recursive linkage that will be available to all Entity object.

children: List[ReportLink]

class testplan.common.entity.base.Resource(**options: Any)

Bases: Entity

An object that can be started/stopped and expose its context object of key/value pair information.

A Resource is usually part of an Environment object of a Runnable object.

Parameters:

• async_start (bool) – Resource can start asynchronously.

• auto_start (bool) – Enables the Environment to start the Resource automatically.

Also inherits all Entity options.

CONFIG

alias of ResourceConfig

STATUS

alias of ResourceStatus

abort() → None

Default abort policy. First abort all dependencies and then itself.

abort_dependencies() → Generator[Entity, None, None]

Returns an empty generator.

property aborted: bool

Returns if entity was aborted.

aborting() → None

Aborting logic for self.

property active: bool

Entity not aborting/aborted.

property async_start: bool

If True, the resource’s parent will take the responsibility to check that the resource has already STARTED or STOPPED.

property auto_start: bool

If False, the resource will not be automatically started by its parent (generally, a Environment object) while the parent is starting.

property cfg: Config

Configuration object.

property context: Environment | None

Key/value pair information of a Resource.

context_input(exclude: list | None = None) → Dict[str, Any]

All attr of self in a dict for context resolution

define_runpath() → None

Define runpath directory based on parent object and configuration.

failover() → Resource | None

API to create the failover resource, to be implemented in derived class

fetch_error_log() → List[str]

Override this method in Resource subclasses to automatically add any useful logs into the report, in case of startup/shutdown exception.

Returns:

text from log files

classmethod filter_locals(local_vars: Dict[str, Any]) → Dict[str, Any]

Filter out init params of None value, they will take default value defined in its ConfigOption object; also filter out special vars that are not init params from local_vars.

Parameters:

local_vars

force_started() → None

Change the status to STARTED (e.g. exception raised).

force_stop() → None

Change the status to STOPPED (e.g. exception raised).

property is_alive: bool

Called to periodically poll the resource health. Default implementation assumes the resource is always healthy.

property logger: TestplanLogger

logger object

make_runpath_dirs() → None

Creates runpath related directories.

property parent: Entity | None

Returns parent Entity.

pause() → None

Pauses entity execution.

pausing() → None

Pause the resource.

pending_work() → bool

Resource has pending work.

post_start() → None

Steps to be executed right after resource is started.

post_stop() → None

Steps to be executed right after resource is stopped.

pre_start() → None

Steps to be executed right before resource starts.

pre_stop() → None

Steps to be executed right before resource stops.

register_failover(klass: Type[Entity], params: dict) → None

Register a failover class to instantiate if resource start fails.

Parameters:

• klass – failover class

• params – parameters for failover class __init__ method

property report: ReportLink

A handle to access the report via recursive parent

restart() → None

Stop and start the resource.

resume() → None

Resumes entity execution.

resuming() → None

Resume the resource.

property runpath: str

Path to be used for temp/output files by entity.

property scratch: str

Path to be used for temp files by entity.

start() → None

Triggers the start logic of a Resource by executing :py:meth: Resource.starting <testplan.common.entity.base.Resource.starting> method.

starting() → None

Start logic for Resource that also sets the status to STARTED.

property status: EntityStatus

Status object.

stop() → None

Triggers the stop logic of a Resource by executing :py:meth: Resource.stopping <testplan.common.entity.base.Resource.stopping> method.

stopping() → None

Stop logic for Resource that also sets the status to STOPPED.

property timer: Timer

uid() → str

Unique identifier of self.

wait(target_status: str | None, timeout: int | None = None) → None

Wait until objects status becomes target status.

Parameters:

• target_status (str) – expected status

• timeout (int or NoneType) – timeout in seconds

class testplan.common.entity.base.ResourceConfig(**options: Any)

Bases: EntityConfig

Configuration object for Resource entity.

classmethod build_schema() → Schema

Build a validation schema using the config options defined in this class and its parent classes.

denormalize() → Config

Create new config object that inherits all explicit attributes from its parents as well.

get_local(name: str, default: Any = None) → Any

Returns a local config setting (not from container)

classmethod get_options() → Dict[Any, Any]

Resource specific config options.

ignore_extra_keys = False

property parent: Config | None

Returns the parent configuration.

set_local(name: str, value: Any) → None

set without any check

class testplan.common.entity.base.ResourceStatus

Bases: EntityStatus

Status of a Resource entity.

NONE = None

PAUSED = 'PAUSED'

PAUSING = 'PAUSING'

RESUMING = 'RESUMING'

STARTED = 'STARTED'

STARTING = 'STARTING'

STOPPED = 'STOPPED'

STOPPING = 'STOPPING'

change(new: str | None) → None

Transition to new status.

Parameters:

new (NoneType or str) – status to be set

clear_metadata() → None

Re-initializes metadata as empty.

property metadata: OrderedDict[str, Any]

TODO

reset() → None

Reset status as None.

property tag: str | None

Current status value.

transitions() → Dict[str | None, set]

Defines the status transitions of a Resource entity.

update_metadata(**metadata: Any) → None

Updates metadata.

Parameters:

metadata (OrderedDict) – additional metadata

class testplan.common.entity.base.ResourceTimings

Bases: object

RESOURCE_SETUP = 'setup'

RESOURCE_TEARDOWN = 'teardown'

class testplan.common.entity.base.Runnable(**options: Any)

Bases: Entity

An object that defines steps, a run method to execute the steps and provides results with the RunnableResult object.

It contains an Environment object of Resource objects that can be started/stopped and utilized by the steps defined.

Parameters:

• interactive_port (int or NoneType) – Enable interactive execution mode on a port.

• interactive_block (bool) – Block on run() on interactive mode.

Also inherits all Entity options.

CONFIG

alias of RunnableConfig

ENVIRONMENT

alias of Environment

RESULT

alias of RunnableResult

STATUS

alias of RunnableStatus

abort() → None

Default abort policy. First abort all dependencies and then itself.

abort_dependencies() → Generator[Resource, None, None]

Yield all dependencies to be aborted before self abort.

property aborted: bool

Returns if entity was aborted.

aborting() → None

Aborting logic for self.

property active: bool

Entity not aborting/aborted.

add_main_batch_steps() → None

Runnable steps to be executed while environment is running.

add_post_main_steps() → None

Runnable steps to run before environment stopped.

add_post_resource_steps() → None

Runnable steps to run after environment stopped.

add_pre_main_steps() → None

Runnable steps to run after environment started.

add_pre_resource_steps() → None

Runnable steps to run before environment started.

add_resource(resource: Resource, uid: str | None = None) → str

Adds a resource in the runnable environment.

Parameters:

• resource (Subclass of Resource) – Resource to be added.

• uid (str or NoneType) – Optional input resource uid.

Returns:

Resource uid assigned.

Return type:

str

add_start_resource_steps() → None

Runnable steps to start environment

add_stop_resource_steps() → None

Runnable steps to stop environment

property cfg: Config

Configuration object.

context_input(exclude: list | None = None) → Dict[str, Any]

All attr of self in a dict for context resolution

define_runpath() → None

Define runpath directory based on parent object and configuration.

dry_run() → None

A testing process that creates result for each step.

classmethod filter_locals(local_vars: Dict[str, Any]) → Dict[str, Any]

Filter out init params of None value, they will take default value defined in its ConfigOption object; also filter out special vars that are not init params from local_vars.

Parameters:

local_vars

property i: TestRunnerIHandler | None

property interactive: TestRunnerIHandler | None

property logger: TestplanLogger

logger object

make_runpath_dirs() → None

Creates runpath related directories.

property parent: Entity | None

Returns parent Entity.

pause() → None

Pauses entity execution.

pausing() → None

Pauses the resource.

post_step_call(step: Callable) → None

Callable to be invoked after each step.

pre_step_call(step: Callable) → None

Callable to be invoked before each step.

property report: ReportLink

A handle to access the report via recursive parent

property resources: Environment

Returns the Environment of Resources.

result: RunnableResult

resume() → None

Resumes entity execution.

resuming() → None

Resumes the resource.

run() → RunnableResult | None

Executes the defined steps and populates the result object.

run_result() → bool

Returns if a run was successful.

property runpath: str

Path to be used for temp/output files by entity.

property scratch: str

Path to be used for temp files by entity.

setup() → None

Setup step to be executed first.

should_run() → bool

Determines if current object should run.

skip_step(step: Callable) → bool

Callable to determine if step should be skipped.

property status: EntityStatus

Status object.

teardown() → None

Teardown step to be executed last.

property timer: Timer

uid() → str

Unique identifier of self.

wait(target_status: str | None, timeout: int | None = None) → None

Wait until objects status becomes target status.

Parameters:

• target_status (str) – expected status

• timeout (int or NoneType) – timeout in seconds

class testplan.common.entity.base.RunnableConfig(**options: Any)

Bases: EntityConfig

Configuration object for Runnable entity.

classmethod build_schema() → Schema

Build a validation schema using the config options defined in this class and its parent classes.

denormalize() → Config

Create new config object that inherits all explicit attributes from its parents as well.

get_local(name: str, default: Any = None) → Any

Returns a local config setting (not from container)

classmethod get_options() → Dict[Any, Any]

Runnable specific config options.

ignore_extra_keys = False

property parent: Config | None

Returns the parent configuration.

set_local(name: str, value: Any) → None

set without any check

class testplan.common.entity.base.RunnableManager(**options: Any)

Bases: Entity

Executes a Runnable entity in a separate thread and handles the abort signals.

Parameters:

• parse_cmdline (bool) – Parse command line arguments.

• runnable (TestRunner) – Test runner.

• resources (list of Resources) – Initial resources.

• abort_signals (list of signals) – Signals to catch and trigger abort.

Also inherits all Entity options.

CONFIG

alias of RunnableManagerConfig

STATUS

alias of EntityStatus

abort() → None

Default abort policy. First abort all dependencies and then itself.

abort_dependencies() → Generator[Runnable, None, None]

Dependencies to be aborted first.

property aborted: bool

Returns if entity was aborted.

aborting() → None

Suppressing not implemented debug log by parent class.

property active: bool

Expose the runnable active attribute.

property cfg: Config

Expose the runnable configuration object.

context_input(exclude: list | None = None) → Dict[str, Any]

All attr of self in a dict for context resolution

define_runpath() → None

Define runpath directory based on parent object and configuration.

enrich_options(options: Dict[str, Any]) → Dict[str, Any]

Enrich the options using parsed command line arguments. Override this method to add extra argument processing logic. The result dictionary is used to initialize the configuration.

Parameters:

options

classmethod filter_locals(local_vars: Dict[str, Any]) → Dict[str, Any]

Filter out init params of None value, they will take default value defined in its ConfigOption object; also filter out special vars that are not init params from local_vars.

Parameters:

local_vars

property logger: TestplanLogger

logger object

make_runpath_dirs() → None

Creates runpath related directories.

property parent: Entity | None

Returns parent Entity.

pause() → None

Pauses entity execution.

pausing() → None

Pause the runnable execution.

property report: ReportLink

A handle to access the report via recursive parent

resume() → None

Resumes entity execution.

resuming() → None

Resume the runnable execution.

run() → RunnableResult | Any

Executes target runnable defined in configuration in a separate thread.

Returns:

Runnable result object.

Return type:

py:class:

RunnableResult <testplan.common.entity.base.RunnableResult>

property runnable: Runnable

Runnable instance.

property runpath: str

Expose the runnable runpath.

property scratch: str

Path to be used for temp files by entity.

property status: EntityStatus

Expose the runnable status.

property timer: Timer

uid() → str

Unique identifier of self.

wait(target_status: str | None, timeout: int | None = None) → None

Wait until objects status becomes target status.

Parameters:

• target_status (str) – expected status

• timeout (int or NoneType) – timeout in seconds

class testplan.common.entity.base.RunnableManagerConfig(**options: Any)

Bases: EntityConfig

Configuration object for RunnableManager entity.

classmethod build_schema() → Schema

Build a validation schema using the config options defined in this class and its parent classes.

denormalize() → Config

Create new config object that inherits all explicit attributes from its parents as well.

get_local(name: str, default: Any = None) → Any

Returns a local config setting (not from container)

classmethod get_options() → Dict[Any, Any]

RunnableManager specific config options.

ignore_extra_keys = False

property parent: Config | None

Returns the parent configuration.

set_local(name: str, value: Any) → None

set without any check

class testplan.common.entity.base.RunnableResult

Bases: object

Result object of a Runnable entity.

run: bool | Exception

step_results: OrderedDict[str, Any]

class testplan.common.entity.base.RunnableStatus

Bases: EntityStatus

Status of a Runnable entity.

EXECUTING = 'EXECUTING'

FINISHED = 'FINISHED'

NONE = None

PAUSED = 'PAUSED'

PAUSING = 'PAUSING'

RESUMING = 'RESUMING'

RUNNING = 'RUNNING'

change(new: str | None) → None

Transition to new status.

Parameters:

new (NoneType or str) – status to be set

clear_metadata() → None

Re-initializes metadata as empty.

property metadata: OrderedDict[str, Any]

TODO

reset() → None

Reset status as None.

property tag: str | None

Current status value.

transitions() → Dict[str | None, set]

Defines the status transitions of a Runnable entity.

update_metadata(**metadata: Any) → None

Updates metadata.

Parameters:

metadata (OrderedDict) – additional metadata

exception testplan.common.entity.base.StatusTransitionException

Bases: Exception

To be raised on illegal state transition attempt.

add_note(note, /)

Add a note to the exception

args

with_traceback(tb, /)

Set self.__traceback__ to tb and return self.

Module contents

Common entity implementations.