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: Any) → 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: Any, **kwargs: Any) → AsyncGenerator[Optional[faust.livecheck.models.TestExecution], None][source]

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

Return type

AsyncGenerator[Optional[TestExecution], None]

async trigger(id: str = None, *args: Any, **kwargs: Any) → faust.livecheck.models.TestExecution[source]

Schedule test execution ASAP.

Return type

TestExecution

logger = <Logger faust.livecheck.case (WARNING)>
async run(*test_args: Any, **test_kwargs: Any) → None[source]

Override this to define your test case.

Return type

None

async resolve_signal(key: str, event: faust.livecheck.models.SignalEvent) → None[source]

Mark test execution signal as resolved.

Return type

None

async execute(test: faust.livecheck.models.TestExecution) → None[source]

Execute test using TestRunner.

Return type

None

async on_test_start(runner: faust.livecheck.runners.TestRunner) → None[source]

Call when a test starts executing.

Return type

None

async on_test_skipped(runner: faust.livecheck.runners.TestRunner) → None[source]

Call when a test is skipped.

Return type

None

async on_test_failed(runner: faust.livecheck.runners.TestRunner, exc: BaseException) → None[source]

Call when invariant in test execution fails.

Return type

None

async on_test_error(runner: faust.livecheck.runners.TestRunner, exc: BaseException) → None[source]

Call when a test execution raises an exception.

Return type

None

async on_test_timeout(runner: faust.livecheck.runners.TestRunner, exc: BaseException) → None[source]

Call when a test execution times out.

Return type

None

async on_test_pass(runner: faust.livecheck.runners.TestRunner) → None[source]

Call when a test execution passes.

Return type

None

async post_report(report: faust.livecheck.models.TestReport) → None[source]

Publish test report.

Return type

None

async on_suite_fail(exc: faust.livecheck.exceptions.SuiteFailed, new_state: faust.livecheck.models.State = <State.FAIL: 'FAIL'>) → None[source]

Call when the suite fails.

Return type

None

property seconds_since_last_fail

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

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

Perform GET request using HTTP client.

Return type

Optional[bytes]

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

Perform POST request using HTTP client.

Return type

Optional[bytes]

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

Perform URL request using HTTP client.

Return type

Optional[bytes]

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