faust.livecheck.case

LiveCheck - Test cases.

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

LiveCheck test case.

Runner

alias of faust.livecheck.runners.TestRunner

state = 'INIT'

Current state of this test.

last_test_received = 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 = None

Timestamp of when the suite last failed.

runtime_avg = None
latency_avg = None
frequency_avg = None
state_transition_delay = 60.0
consecutive_failures = 0
total_failures = 0
name = None

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

active = True
probability = 0.5

Probability of test running when live traffic is going through.

warn_stalled_after = 1800.0

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

test_expires = datetime.timedelta(0, 10800)
frequency = 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 = 100

Max items to store in latency_history and runtime_history.

max_consecutive_failures = 30
url_timeout_total = 300.0
url_timeout_connect = None
url_error_retries = 10
url_error_delay_min = 0.5
url_error_delay_backoff = 1.5
url_error_delay_max = 5.0
maybe_trigger(id: str = None, *args, **kwargs) → AsyncGenerator[Optional[faust.livecheck.models.TestExecution], None][source]

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

Return type

AsyncGenerator[Optional[TestExecution], None]

logger = <Logger faust.livecheck.case (WARNING)>
property seconds_since_last_fail

Return number of seconds since any test failed. :rtype: Optional[float]

property current_test

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

property current_execution

Return the currently executing TestRunner in this task. :rtype: Optional[TestRunner]

property label

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