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
-
async
-
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.
-
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.
-
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
-
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.
-
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
-
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
- Returns
-
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)>¶
-