faust.livecheck.runners

LiveCheck - Test runner.

class faust.livecheck.runners.TestRunner(case: _Case, test: TestExecution, started: float)

Execute and keep track of test instance.

state: State = 'INIT'

report: Optional[TestReport] = None

error: Optional[BaseException] = None

case: _Case

test: TestExecution

started: float

ended: Optional[float]

runtime: Optional[float]

logs: List[Tuple[str, Tuple]]

signal_latency: Dict[str, float]

async execute() → None

Execute this test.

Return type:

None

async skip(reason: str) → NoReturn

Skip this test execution.

Return type:

_SpecialForm

async on_skipped(exc: LiveCheckTestSkipped) → None

Call when a test execution was skipped.

Return type:

None

async on_start() → None

Call when a test starts executing.

Return type:

None

async on_signal_wait(signal: BaseSignal, timeout: float) → None

Call when the test is waiting for a signal.

Return type:

None

async on_signal_received(signal: BaseSignal, time_start: float, time_end: float) → None

Call when a signal related to this test is received.

Return type:

None

async on_failed(exc: BaseException) → None

Call when an invariant in the test has failed.

Return type:

None

async on_error(exc: BaseException) → None

Call when test execution raises error.

Return type:

None

async on_timeout(exc: BaseException) → None

Call when test execution times out.

Return type:

None

async on_pass() → None

Call when test execution returns successfully.

Return type:

None

log_info(msg: str, *args: Any) → None

Log information related to the current execution.

Return type:

None

end() → None

End test execution.

Return type:

None