e3.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 e3.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[T](Protocol):
    def meth(self) -> T:
        ...

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

class e3.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 e3.testsuite.fragment.TestFragment

Base class for testcase scheduling units.

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

index: int: Unique integer 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.

static static_push_error_result(uid: str, index: int, 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.
index – Index 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.

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 e3.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)

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.

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 e3.testsuite.fragment.ProcessTestFragment(uid: str, driver: e3.testsuite.driver.TestDriver, callback_name: str, slot: int, running_status: e3.testsuite.running_status.RunningStatus, 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

fragment_index: int

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.

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.

e3.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.