prefect.testing.fixtures
Functions
add_prefect_loggers_to_caplog
add_prefect_loggers_to_caplog(caplog: pytest.LogCaptureFixture) -> Generator[None, None, None]
is_port_in_use
is_port_in_use(port: int) -> bool
use_hosted_api_server
use_hosted_api_server(hosted_api_server: str) -> Generator[str, None, None]
Sets PREFECT_API_URL to the test session’s hosted API endpoint.
disable_hosted_api_server
disable_hosted_api_server() -> Generator[None, None, None]
Disables the hosted API server by setting PREFECT_API_URL to None.
enable_ephemeral_server
enable_ephemeral_server(disable_hosted_api_server: None) -> Generator[None, None, None]
Enables the ephemeral server by setting PREFECT_SERVER_ALLOW_EPHEMERAL_MODE to True.
mock_anyio_sleep
mock_anyio_sleep(monkeypatch: pytest.MonkeyPatch) -> Generator[Callable[[float], None], None, None]
Mock sleep used to not actually sleep but to set the current time to now + sleep
delay seconds while still yielding to other tasks in the event loop.
Provides “assert_sleeps_for” context manager which asserts a sleep time occurred
within the context while using the actual runtime of the context as a tolerance.
recorder
puppeteer
events_api_url
events_api_url(events_server: Server, unused_tcp_port: int) -> str
events_cloud_api_url
events_cloud_api_url(events_server: Server, unused_tcp_port: int) -> str
mock_should_emit_events
mock_should_emit_events(monkeypatch: pytest.MonkeyPatch) -> mock.Mock
asserting_events_worker
asserting_events_worker(monkeypatch: pytest.MonkeyPatch) -> Generator[EventsWorker, None, None]
asserting_and_emitting_events_worker
asserting_and_emitting_events_worker(monkeypatch: pytest.MonkeyPatch) -> Generator[EventsWorker, None, None]
reset_worker_events
reset_worker_events(asserting_events_worker: EventsWorker) -> Generator[None, None, None]
enable_lineage_events
enable_lineage_events() -> Generator[None, None, None]
A fixture that ensures lineage events are enabled.
Classes
Recorder
Puppeteer
AssertingEventsPipeline
AssertingAndEmittingEventsPipeline
