`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]

​prefect.testing.utilities

​Functions

​exceptions_equal

​kubernetes_environments_equal

​assert_does_not_warn

​prefect_test_harness

​get_most_recent_flow_run

​assert_blocks_equal

​assert_uses_result_serializer

​assert_uses_result_storage

​a_test_step

​b_test_step