testsuite.running_status

Classes

RunningStatus

Module Contents

class testsuite.running_status.RunningStatus(filename: str, update_interval: float = 1.0, max_consecutive_failures: int = 0)

dag: e3.collection.dag.DAG | None = None

filename

lock

running: Dict[str, int]

Set of test fragments ids currently running, indexed by UID.

The dict values come from the builtin id() function called on the test fragments.

completed: Dict[str, int]

Set of test fragments ids that completed their job, indexed by UID.

The dict values come from the builtin id() function called on the test fragments.

status_counters: Dict[e3.testsuite.result.TestStatus, int]

Snapshot of the testsuite’s report index status counters.

We preserve a copy to avoid inconsistent state due to race conditions: these counters are updated in the collect_result method while the other sets are updated in TestFragment.run, which is executed in workers (i.e. other threads).

update_interval = 1.0

no_update_before = 0.0

max_consecutive_failures = 0

consecutive_failures = 0: Number of consecutive failure/error results we just processed. Used to abort the testsuite when there are too many issues.

aborted_too_many_failures = False: Whether the testsuite aborted because of too many consecutive test failures (see the –max-consecutive-failures command-line option).

set_dag(dag: e3.collection.dag.DAG) → None: Set the DAG that contains TestFragment instances.

start(fragment: e3.testsuite.fragment.TestFragment) → None: Put a fragment in the “running” set.

complete(fragment: e3.testsuite.fragment.TestFragment) → None: Move a fragment from the “running” set to the “completed” set.

process_result(result: e3.testsuite.result.TestResultSummary) → None

Integrate a test result in the testsuite status.

This increments status counters and triggers testsuite abortion if there were too many consecutive failures.

dump() → None: Write a report for this status as human-readable text to “fp”.