faust.livecheck.app

LiveCheck - Faust Application.

class faust.livecheck.app.LiveCheck(id: str, *, test_topic_name: str = None, bus_topic_name: str = None, report_topic_name: str = None, bus_concurrency: int = None, test_concurrency: int = None, send_reports: bool = None, **kwargs: Any) → None[source]

LiveCheck application.

SCAN_CATEGORIES = ['faust.agent', 'faust.command', 'faust.page', 'faust.service', 'faust.task', 'livecheck.case']
class Signal(name: str = '', case: faust.livecheck.signals._Case = None, index: int = -1) → None

Signal for test case using Kafka.

Used to wait for something to happen elsewhere.

async send(value: VT = None, *, key: Any = None, force: bool = False) → None

Notify test that this signal is now complete.

Return type

None

async wait(*, key: Any = None, timeout: Union[datetime.timedelta, float, str] = None) → VT

Wait for signal to be completed.

Return type

~VT

class 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

LiveCheck test case.

Runner

alias of faust.livecheck.runners.TestRunner

active = True
consecutive_failures = 0
property current_execution

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

property current_test

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

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

Execute test using TestRunner.

Return type

None

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

Perform GET request using HTTP client.

Return type

Optional[bytes]

property label

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

last_fail = None
last_test_received = None
latency_avg = None
logger = <Logger faust.livecheck.case (WARNING)>
max_consecutive_failures = 30
max_history = 100
maybe_trigger(id: str = None, *args: Any, **kwargs: Any) → AsyncGenerator[Optional[faust.livecheck.models.TestExecution], None]

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

Return type

AsyncGenerator[Optional[TestExecution], None]

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

Call when the suite fails.

Return type

None

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

Call when a test execution raises an exception.

Return type

None

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

Call when invariant in test execution fails.

Return type

None

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

Call when a test execution passes.

Return type

None

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

Call when a test is skipped.

Return type

None

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

Call when a test starts executing.

Return type

None

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

Call when a test execution times out.

Return type

None

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

Publish test report.

Return type

None

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

Perform POST request using HTTP client.

Return type

Optional[bytes]

probability = 0.5
realtime_logs = False
async resolve_signal(key: str, event: faust.livecheck.models.SignalEvent) → None

Mark test execution signal as resolved.

Return type

None

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

Override this to define your test case.

Return type

None

runtime_avg = None
property seconds_since_last_fail

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

state = 'INIT'
state_transition_delay = 60.0
test_expires = datetime.timedelta(0, 10800)
total_failures = 0
async trigger(id: str = None, *args: Any, **kwargs: Any) → faust.livecheck.models.TestExecution

Schedule test execution ASAP.

Return type

TestExecution

url_error_delay_backoff = 1.5
url_error_delay_max = 5.0
url_error_delay_min = 0.5
url_error_retries = 10
async url_request(method: str, url: Union[str, yarl.URL], **kwargs: Any) → Optional[bytes]

Perform URL request using HTTP client.

Return type

Optional[bytes]

url_timeout_connect = None
url_timeout_total = 300.0
warn_stalled_after = 1800.0
classmethod for_app(app: faust.types.app.AppT, *, prefix: str = 'livecheck-', web_port: int = 9999, test_topic_name: str = None, bus_topic_name: str = None, report_topic_name: str = None, bus_concurrency: int = None, test_concurrency: int = None, send_reports: bool = None, **kwargs: Any) → faust.livecheck.app.LiveCheck[source]

Create LiveCheck application targeting specific app.

The target app will be used to configure the LiveCheck app.

Return type

LiveCheck[]

test_topic_name = 'livecheck'
bus_topic_name = 'livecheck-bus'
report_topic_name = 'livecheck-report'
bus_concurrency = 30

Number of concurrent actors processing signal events.

test_concurrency = 100

Number of concurrent actors executing test cases.

send_reports = True

Unset this if you don’t want reports to be sent to the report_topic_name topic.

property current_test

Return the current test context (if any). :rtype: Optional[TestExecution]

on_produce_attach_test_headers(sender: faust.types.app.AppT, key: bytes = None, value: bytes = None, partition: int = None, timestamp: float = None, headers: List[Tuple[str, bytes]] = None, **kwargs: Any) → None[source]

Attach test headers to Kafka produce requests.

Return type

None

case(*, name: str = None, probability: float = None, warn_stalled_after: Union[datetime.timedelta, float, str] = datetime.timedelta(0, 1800), active: bool = None, test_expires: Union[datetime.timedelta, float, str] = None, frequency: Union[datetime.timedelta, float, str] = None, max_history: int = None, max_consecutive_failures: int = None, url_timeout_total: float = None, url_timeout_connect: float = None, url_error_retries: float = None, url_error_delay_min: float = None, url_error_delay_backoff: float = None, url_error_delay_max: float = None, base: Type[faust.livecheck.case.Case] = faust.livecheck.case.Case) → Callable[Type, faust.livecheck.case.Case][source]

Decorate class to be used as a test case.

Return type

Callable[[Type[+CT_co]], Case[]]

Returns

faust.livecheck.Case.

add_case(case: faust.livecheck.case.Case) → faust.livecheck.case.Case[source]

Add and register new test case.

Return type

Case[]

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

Publish test report to reporting topic.

Return type

None

logger = <Logger faust.livecheck.app (WARNING)>
async on_start() → None[source]

Call when LiveCheck application starts.

Return type

None

async on_started() → None[source]

Call when LiveCheck application is fully started.

Return type

None

bus[source]

Topic used for signal communication.

pending_tests[source]

Topic used to keep pending test executions.

reports[source]

Topic used to log test reports.