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) → 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.
-
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) → 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
]
-
frequency
= None¶
-
frequency_avg
= None¶
-
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, **kwargs) → AsyncGenerator[Optional[faust.livecheck.models.TestExecution], None]¶ Schedule test execution, or not, based on probability setting.
- Return type
AsyncGenerator
[Optional
[TestExecution
],None
]
-
probability
= 0.5¶
-
realtime_logs
= False¶
-
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¶
-
url_error_delay_backoff
= 1.5¶
-
url_error_delay_max
= 5.0¶
-
url_error_delay_min
= 0.5¶
-
url_error_retries
= 10¶
-
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) → 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) → 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
[]
-
logger
= <Logger faust.livecheck.app (WARNING)>¶
-