testplan.runners.pools.tasks package

Submodules

testplan.runners.pools.tasks.base module

Tasks and task results base module.

class testplan.runners.pools.tasks.base.Task(target: str | Test | None = None, module: str | None = None, path: str | None = None, args: tuple | None = None, kwargs: dict | None = None, uid: str | None = None, rerun: int = 0, weight: int = 0, part: Tuple[int, int] | None = None)[source]

Bases: SelectiveSerializable

Container of a target or path to a target that can be materialized into a runnable item. The arguments of the Task need to be serializable.

Parameters:

  • target – A runnable or a string path to a runnable or a callable to a runnable or a string path to a callable to a runnable.

  • module – Module name that contains the task target definition.

  • path – Path to module, default is current working directory.

  • args – Args of target for task materialization.

  • kwargs – Kwargs of target for task materialization.

  • uid – Task uid.

  • rerun – Rerun the task up to user specified times until it passes, by default 0 (no rerun). To enable task rerun feature, set to positive value no greater than 3.

  • weight – Affects task scheduling - the larger the weight, the sooner task will be assigned to a worker. Default weight is 0, tasks with the same weight will be scheduled in the order they are added.

  • part – part param that will be propagate to MultiTest

MAX_RERUN_LIMIT = 3
abort()[source]

For compatibility reason when task is added into an executor.

property aborted

Returns if task was aborted.

property args: Tuple

Task target args.

property executors

Executors to which the task had been assigned.

property kwargs: Dict

Task target kwargs.

materialize(target=None) → Test[source]

Create the actual task target executable/runnable object.

property module: str

Task target module.

property reassign_cnt: int

how many times the task is reassigned for rerun.

rebase_path(local, remote)[source]

adapt task’s path for remote execution if necessary

property rerun: int

how many times the task is allowed to rerun.

property serializable_attrs: Tuple[str, ...]

Attributes to be included in serialization.

uid() → str[source]

Task string uid.

property weight: int
exception testplan.runners.pools.tasks.base.TaskMaterializationError[source]

Bases: Exception

Error materializing task target to be executed.

class testplan.runners.pools.tasks.base.TaskResult(task: Task | None = None, result: TestResult | None = None, status: bool = False, reason: str | None = None, follow: Task | None = None)[source]

Bases: SelectiveSerializable

Contains result of the executed task target and status/errors/reason information that happened during task execution.

May contain follow up tasks.

property follow: Task | None

Follow up tasks that need to be scheduled next.

property reason: str | None

Reason for failed status.

property result: TestResult | None

Actual task target result.

property serializable_attrs: Tuple

Attributes to be included in serialization.

property status: bool

Result status. Should be True on correct successful execution.

property task: Task | None

Original task.

uid() → str[source]

Task result uid

class testplan.runners.pools.tasks.base.TaskTargetInformation(target_params: Sequence[Sequence | dict], multitest_parts: int | str | NoneType, task_kwargs: Dict[str, Any])[source]

Bases: object

multitest_parts: int | str | None
target_params: Sequence[Sequence | dict]
task_kwargs: Dict[str, Any]
testplan.runners.pools.tasks.base.is_task_target(func)[source]

Check if a callable object is a task target.

testplan.runners.pools.tasks.base.set_task_target(func: Callable, info: TaskTargetInformation)[source]

Mark a callable object as a task target which can be packaged in a Task object.

testplan.runners.pools.tasks.base.task_target(parameters: Callable | Sequence[Sequence | dict] = None, multitest_parts: int | Literal['auto'] | None = None, **kwargs)[source]

Decorator to make task target discoverable by plan.schedule_all.

Parameters:

  • parameters (list or tuple that contains list or tuple or dict) – A collection of parameters to be used to create task objects. list or tuple entry will be passed to target as positional arguments and dict entry will be passed to target as keyword arguments.

  • multitest_parts (int or “auto”) – The number of multitest parts that will be generated from this task target, only applies if the task returns multitest type

  • kwargs (dict) – additional args to Task class, e.g rerun, weight etc.

Module contents

Serializable tasks and task results module.