prefect.testing.fixtures

Functions

add_prefect_loggers_to_caplog ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L48" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

add_prefect_loggers_to_caplog(caplog: pytest.LogCaptureFixture) -> Generator[None, None, None]

is_port_in_use ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L62" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

is_port_in_use(port: int) -> bool

hosted_api_server ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L68" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

hosted_api_server(unused_tcp_port_factory: Callable[[], int], test_database_connection_url: Optional[str]) -> AsyncGenerator[str, None]

Runs an instance of the Prefect API server in a subprocess instead of the using the ephemeral application.

Uses the same database as the rest of the tests.

use_hosted_api_server ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L166" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

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 ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L180" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

disable_hosted_api_server() -> Generator[None, None, None]

Disables the hosted API server by setting PREFECT_API_URL to None.

enable_ephemeral_server ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L193" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

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 ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L210" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

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 ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L314" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

recorder() -> Recorder

puppeteer ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L319" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

puppeteer() -> Puppeteer

events_server ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L324" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

events_server(unused_tcp_port: int, recorder: Recorder, puppeteer: Puppeteer) -> AsyncGenerator[Server, None]

events_api_url ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L427" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

events_api_url(events_server: Server, unused_tcp_port: int) -> str

events_cloud_api_url ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L432" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

events_cloud_api_url(events_server: Server, unused_tcp_port: int) -> str

mock_should_emit_events ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L437" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

mock_should_emit_events(monkeypatch: pytest.MonkeyPatch) -> mock.Mock

asserting_events_worker ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L445" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

asserting_events_worker(monkeypatch: pytest.MonkeyPatch) -> Generator[EventsWorker, None, None]

asserting_and_emitting_events_worker ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L458" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

asserting_and_emitting_events_worker(monkeypatch: pytest.MonkeyPatch) -> Generator[EventsWorker, None, None]

events_pipeline ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L471" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

events_pipeline(asserting_events_worker: EventsWorker) -> AsyncGenerator[EventsPipeline, None]

emitting_events_pipeline ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L522" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

emitting_events_pipeline(asserting_and_emitting_events_worker: EventsWorker) -> AsyncGenerator[EventsPipeline, None]

reset_worker_events ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L541" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

reset_worker_events(asserting_events_worker: EventsWorker) -> Generator[None, None, None]

Classes

Recorder ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L283" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Puppeteer ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/testing/fixtures.py#L296" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}