utilities

prefect.testing.utilities

Internal utilities for tests.

Functions

exceptions_equal

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

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

assert_does_not_warn

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

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

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

assert_blocks_equal

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

assert_uses_result_serializer

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

assert_uses_result_storage

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

a_test_step

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

b_test_step

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