testsuite.fragment

Classes

`FragmentCallback`	Base class for protocol classes.
`FragmentData`	Data for a job unit in the testsuite.
`TestFragment`	Base class for testcase scheduling units.
`ThreadTestFragment`	Run a test fragment in a thread.
`ProcessTestFragment`	Run a test fragment in a separate process.

Functions

run_fragment(→ None)

Run a test fragment.

Module Contents

class testsuite.fragment.FragmentCallback

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

__call__(previous_values: Dict[str, Any], slot: int) → None

class testsuite.fragment.FragmentData

Data for a job unit in the testsuite.

Each FragmentData instance is recorded in the testsuite global DAG to control the order of execution of all fragments with the requested level of parallelism.

Note that the job scheduler turns FragmentData instances into TestFragment ones during the execution (see Testsuite.job_factory callback).

uid: str

driver: e3.testsuite.driver.TestDriver

name: str

callback: FragmentCallback

callback_by_name: bool: Whether callback is just the name method of driver.

matches(driver_cls: Type[e3.testsuite.driver.TestDriver], name: str) → bool

Return whether this fragment matches the given name/test driver.

If name is left to None, just check the driver type.

clear_driver_data() → None

Remove references to TestDriver instances and related data.

Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data.

class testsuite.fragment.TestFragment

Base class for testcase scheduling units.

uid: str: Unique string identifier for this test fragment.

driver: e3.testsuite.driver.TestDriver: Test driver that is responsible for this test fragment.

running_status: e3.testsuite.running_status.RunningStatus: RunningStatus instance to signal when job starts/completes.

result_queue: e3.testsuite.driver.ResultQueue: List of test results that this fragments plans to integrate to the testsuite report.

event_notifier: e3.testsuite.event_notifications.EventNotifier: Event notifier for test start/end events.

started_test: bool: Set when the fragment starts: whether this fragment was the first to start for its owning test driver.

ended_test: bool: Set when the fragment completes: whether this fragment was the last to complete for its owning test driver.

maybe_notify_started() → None

Send a test start notification when appropriate.

If this fragment was the first one to start for its test driver, send the corresponding notification.

maybe_notify_ended() → None

Send a test end notification when appropriate.

If this fragment was the last one to complete for its test driver, send the corresponding notification.

static static_push_error_result(uid: str, driver: e3.testsuite.driver.TestDriver) → None

Generate a test result to log the exception and traceback.

This helper method is meant to be used when the execution of the test fragments aborts because of an uncaught exception. We must report a test error, and we provide exception information for post-mortem investigation.

Parameters:

uid – UID for the test fragment.
driver – TestDriver for the test fragment.

push_error_result(exc: Exception) → None: Shortcut for static_push_error_result on the current fragment.

must_run() → bool

Return if the fragment must be run.

It must be skipped when the testsuite has aborted because of too many consecutive failures: we push a SKIP result in that case.

abstractmethod clear_driver_data() → None

Remove references to TestDriver instances and related data.

Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data.

class testsuite.fragment.ThreadTestFragment(uid: str, driver: e3.testsuite.driver.TestDriver, callback: FragmentCallback, previous_values: Dict[str, Any], notify_end: Callable[[str], None], running_status: e3.testsuite.running_status.RunningStatus, event_notifier: e3.testsuite.event_notifications.EventNotifier)

Bases: e3.job.Job, TestFragment

Run a test fragment in a thread.

driver: Test driver that is responsible for this test fragment.

previous_values

running_status: RunningStatus instance to signal when job starts/completes.

result_queue: e3.testsuite.driver.ResultQueue: List of test results that this fragments plans to integrate to the testsuite report.

event_notifier: Event notifier for test start/end events.

clear_driver_data() → None

Remove references to TestDriver instances and related data.

Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data.

run() → None: Run the test fragment.

class testsuite.fragment.ProcessTestFragment(uid: str, driver: e3.testsuite.driver.TestDriver, callback_name: str, slot: int, running_status: e3.testsuite.running_status.RunningStatus, event_notifier: e3.testsuite.event_notifications.EventNotifier, env: e3.env.Env)

Bases: e3.testsuite.multiprocess_scheduler.Worker, TestFragment

Run a test fragment in a separate process.

class Input

Subprocess input data.

fragment_uid: str

driver_cls: Type[e3.testsuite.driver.TestDriver]

test_env: Dict[str, Any]

callback_name: str

slot: int

class Output

Subprocess output data.

result_queue: e3.testsuite.driver.ResultQueue

exchange_file: str: Name of the file to exchange data with the subprocess.

running_status: RunningStatus instance to signal when job starts/completes.

event_notifier: Event notifier for test start/end events.

result_queue = []: List of test results that this fragments plans to integrate to the testsuite report.

clear_driver_data() → None

Remove references to TestDriver instances and related data.

Doing this is necessary after each fragment is complete to keep memory consumption under control for big testsuites: test driver instances may contain a lot of data.

start() → e3.os.process.Run

Create and return the subprocess to do the work.

All subclasses must override this.

collect_result() → None

extract_result_queue() → None

Read the result queue from the exchange file.

Try to extract the result queue from the exchange file and put results in the driver’s result queue. If anything goes sour, create an error result in the same result queue.

testsuite.fragment.run_fragment(argv: List[str] | None = None) → None

Run a test fragment.

This function is meant to be the entry point of a standalone script, to run a fragment in a subprocess, separate from the main testsuite process.