faust.livecheck.case

LiveCheck - Test cases.

class faust.livecheck.case.Case(*, app: _LiveCheck, name: str, probability: Optional[float] = None, warn_stalled_after: Optional[Union[timedelta, float, str]] = None, active: Optional[bool] = None, signals: Optional[Iterable[BaseSignal]] = None, test_expires: Optional[Union[timedelta, float, str]] = None, frequency: Optional[Union[timedelta, float, str]] = None, realtime_logs: Optional[bool] = None, max_history: Optional[int] = None, max_consecutive_failures: Optional[int] = None, url_timeout_total: Optional[float] = None, url_timeout_connect: Optional[float] = None, url_error_retries: Optional[int] = None, url_error_delay_min: Optional[float] = None, url_error_delay_backoff: Optional[float] = None, url_error_delay_max: Optional[float] = None, **kwargs: Any)

LiveCheck test case.

Runner

alias of TestRunner

status: State = 'INIT'

Current state of this test.

last_test_received: Optional[float] = None

The warn_stalled_after timer uses this to keep track of either when a test was last received, or the last time the timer timed out.

last_fail: Optional[float] = None

Timestamp of when the suite last failed.

runtime_avg: Optional[float] = None

latency_avg: Optional[float] = None

frequency_avg: Optional[float] = None

state_transition_delay: float = 60.0

consecutive_failures: int = 0

total_failures: int = 0

app: _LiveCheck

name: str

Name of the test If not set this will be generated out of the subclass name.

active: bool = True

probability: float = 0.5

Probability of test running when live traffic is going through.

warn_stalled_after: float = 1800.0

Timeout in seconds for when after we warn that nothing is processing.

signals: Dict[str, BaseSignal]

test_expires: timedelta = datetime.timedelta(seconds=10800)

frequency: Optional[float] = None

How often we execute the test using fake data (define Case.make_fake_request()).

Set to None if production traffic is frequent enough to satisfy warn_stalled_after.

realtime_logs = False

max_history: int = 100

Max items to store in latency_history and runtime_history.

max_consecutive_failures: int = 30

url_timeout_total: Optional[float] = 300.0

url_timeout_connect: Optional[float] = None

url_error_retries: int = 10

url_error_delay_min: float = 0.5

url_error_delay_backoff: float = 1.5

url_error_delay_max: float = 5.0

frequency_history: Deque[float]

latency_history: Deque[float]

runtime_history: Deque[float]

total_by_state: Counter[State]

total_signals: int

maybe_trigger(id: Optional[str] = None, *args: Any, **kwargs: Any) → AsyncGenerator[Optional[TestExecution], None]

Schedule test execution, or not, based on probability setting.

Return type:

_GenericAlias[_UnionGenericAlias[TestExecution, None], None]

async trigger(id: Optional[str] = None, *args: Any, **kwargs: Any) → TestExecution

Schedule test execution ASAP.

Return type:

TestExecution

async run(*test_args: Any, **test_kwargs: Any) → None

Override this to define your test case.

Return type:

None

logger: logging.Logger = <Logger faust.livecheck.case (WARNING)>

async resolve_signal(key: str, event: SignalEvent) → None

Mark test execution signal as resolved.

Return type:

None

log: CompositeLogger

diag: DiagT

async_exit_stack: AsyncExitStack

exit_stack: ExitStack

async execute(test: TestExecution) → None

Execute test using TestRunner.

Return type:

None

async on_test_start(runner: TestRunner) → None

Call when a test starts executing.

Return type:

None

async on_test_skipped(runner: TestRunner) → None

Call when a test is skipped.

Return type:

None

async on_test_failed(runner: TestRunner, exc: BaseException) → None

Call when invariant in test execution fails.

Return type:

None

async on_test_error(runner: TestRunner, exc: BaseException) → None

Call when a test execution raises an exception.

Return type:

None

async on_test_timeout(runner: TestRunner, exc: BaseException) → None

Call when a test execution times out.

Return type:

None

async on_test_pass(runner: TestRunner) → None

Call when a test execution passes.

Return type:

None

async post_report(report: TestReport) → None

Publish test report.

Return type:

None

async make_fake_request() → None

Return type:

None

async on_suite_fail(exc: SuiteFailed, new_state: State = State.FAIL) → None

Call when the suite fails.

Return type:

None

property seconds_since_last_fail: Optional[float]

Return number of seconds since any test failed. :rtype: _UnionGenericAlias[float, None]

async get_url(url: Union[str, URL], **kwargs: Any) → Optional[bytes]

Perform GET request using HTTP client.

Return type:

_UnionGenericAlias[bytes, None]

async post_url(url: Union[str, URL], **kwargs: Any) → Optional[bytes]

Perform POST request using HTTP client.

Return type:

_UnionGenericAlias[bytes, None]

async url_request(method: str, url: Union[str, URL], **kwargs: Any) → Optional[bytes]

Perform URL request using HTTP client.

Return type:

_UnionGenericAlias[bytes, None]

property current_test: Optional[TestExecution]

Return the currently active test in this task (if any). :rtype: _UnionGenericAlias[TestExecution, None]

property current_execution: Optional[TestRunner]

Return the currently executing TestRunner in this task. :rtype: _UnionGenericAlias[TestRunner, None]

property label: str

Return human-readable label for this test case. :rtype: str