prefect.testing.utilities

Internal utilities for tests.

Functions

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

exceptions_equal(a: Exception, b: Exception) -> bool

Exceptions cannot be compared by ==. They can be compared using is but this will fail if the exception is serialized/deserialized so this utility does its best to assert equality using the type and args used to initialize the exception

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

kubernetes_environments_equal(actual: list[dict[str, str]], expected: list[dict[str, str]] | dict[str, str]) -> bool

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

assert_does_not_warn(ignore_warnings: list[type[Warning]] | None = None) -> Generator[None, None, None]

Converts warnings to errors within this context to assert warnings are not raised, except for those specified in ignore_warnings.

Parameters:

• ignore_warnings: List of warning types to ignore. Example: [DeprecationWarning, UserWarning]

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

prefect_test_harness(server_startup_timeout: int | None = 30)

Temporarily run flows against a local SQLite database for testing.

Args:

• server_startup_timeout: The maximum time to wait for the server to start. Defaults to 30 seconds. If set to None, the value of PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS will be used.

Examples:

from prefect import flow
from prefect.testing.utilities import prefect_test_harness


@flow
def my_flow():
    return 'Done!'

with prefect_test_harness():
    assert my_flow() == 'Done!' # run against temporary db

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

get_most_recent_flow_run(client: 'PrefectClient | None' = None, flow_name: str | None = None) -> 'FlowRun'

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

assert_blocks_equal(found: Block, expected: Block, exclude_private: bool = True, **kwargs: Any) -> None

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

assert_uses_result_serializer(state: State, serializer: str | Serializer, client: 'PrefectClient') -> None

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

assert_uses_result_storage(state: State, storage: 'str | ReadableFileSystem', client: 'PrefectClient') -> None

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

a_test_step(**kwargs: Any) -> dict[str, Any]

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

b_test_step(**kwargs: Any) -> dict[str, Any]