Utilities

General-purpose utilities for polling, dict navigation, time assertions, and temporary workspaces.

wait_until — polling with timeout

behave_kit.wait.wait_until(condition: Callable[[], object], *, timeout: float = 10.0, interval: float = 0.5, message: str = '') → None

Poll condition until it is truthy or timeout seconds elapse.

Parameters:

• condition – Zero-argument callable whose return value is evaluated with scalar-bool coercion (tolerates array-like objects).

• timeout – Maximum number of seconds to wait (default 10).

• interval – Seconds between polls (default 0.5).

• message – Custom message for the timeout error.

Raises:

TimeoutError – If condition does not become truthy within timeout seconds.

Poll a condition until it becomes truthy or the timeout elapses. Common in E2E and integration tests where a resource (API, database, browser) is not immediately available after an action.

Examples

from behave_kit import wait_until

@then("the API should be healthy")
def step(context):
    wait_until(
        lambda: context.client.get("/health").status_code == 200,
        timeout=10,
        interval=0.5,
    )

Custom timeout message:

wait_until(
    lambda: context.db.is_ready(),
    timeout=30,
    message="Database did not become ready in time",
)

The condition callable can return any truthy value, including array-like objects (NumPy arrays are supported via scalar-bool coercion).

get_path — dot-notation dict navigation

behave_kit.utils.get_path(data: object, path: str, default: Any = <object object>) → Any

Return the value at path within nested data, or default.

path uses dot notation to traverse nested dicts:

get_path(response, “user.address.city”)

Numeric segments traverse lists:

get_path(response, “users.0.name”)

Parameters:

• data – The root dict or list to traverse.

• path – Dot-separated path string.

• default – Value to return when the path does not exist. If not provided, BehaveKitError is raised instead.

Raises:

BehaveKitError – When the path is not found and no default is given.

Navigate nested dicts and lists using dot notation. Supports list indices and optional defaults.

Examples

from behave_kit import get_path

data = {
    "user": {
        "name": "Alice",
        "address": {"city": "Berlin"},
    },
    "users": [
        {"name": "Alice"},
        {"name": "Bob"},
    ],
}

get_path(data, "user.name")                # "Alice"
get_path(data, "user.address.city")        # "Berlin"
get_path(data, "users.0.name")             # "Alice"
get_path(data, "users.1.name")             # "Bob"
get_path(data, "user.phone", default="")   # "" (missing key, returns default)

When a key or index is missing and no default is provided, a BehaveKitError is raised with a descriptive message:

get_path(data, "user.missing")  # raises BehaveKitError: Key 'missing' not found

Time assertions — assert_under and @timed

behave_kit.timing.assert_under(seconds: float, func: Callable[[], R], *, message: str = '') → R

Run func and assert it completes within seconds.

Parameters:

• seconds – Maximum allowed execution time.

• func – Zero-argument callable to time.

• message – Custom message for the timeout error.

Returns:

The return value of func.

Raises:

TimeoutExceededError – If func takes longer than seconds.

behave_kit.timing.timed(seconds: float) → Callable[[Callable[[Concatenate[Context, P]], R]], Callable[[Concatenate[Context, P]], R]]

Decorator: assert a step function completes within seconds.

Usage:

@timed(2.0)
@when("I fetch the data")
def step(context): ...

Raises:

BehaveKitError – If seconds is negative.

class behave_kit.timing.TimeoutExceededError(message: str, *, cause: BaseException | None = None, suggestion: str | None = None)

Bases: BehaveKitError

Raised when a callable takes longer than the allowed time.

assert_under verifies that a callable completes within a deadline. @timed is a decorator that wraps a Behave step function with the same check.

Examples

from behave_kit import assert_under, timed

# Direct usage
result = assert_under(2.0, lambda: client.get("/health"))
assert result.status_code == 200

# As a step decorator
@timed(1.5)
@when("I fetch the data quickly")
def step(context):
    context.data = fetch_data()

If the callable takes longer than the allowed time, a TimeoutExceededError (subclass of BehaveKitError) is raised:

from behave_kit import assert_under, TimeoutExceededError

try:
    assert_under(0.5, lambda: slow_operation())
except TimeoutExceededError as exc:
    print(f"Too slow: {exc}")

temp_workspace — isolated temporary directory

behave_kit.workspace.temp_workspace(*, prefix: str = 'behave_kit_') → Iterator[Path]

Yield a temporary directory and restore the CWD on exit.

Parameters:

prefix – Prefix for the temporary directory name.

Yields:

Path to the temporary directory (the CWD while inside the block).

Create a temporary directory, change the CWD into it, and restore the original CWD on exit. The directory is removed automatically.

Examples

from behave_kit import temp_workspace

@when("I create a temporary workspace")
def step(context):
    with temp_workspace() as tmp:
        config_path = tmp / "config.json"
        config_path.write_text('{"debug": true}')
        context.config_dir = tmp
    # CWD is restored and the directory is cleaned up

The CWD is restored even if an exception is raised inside the block:

with temp_workspace() as tmp:
    raise ValueError("oops")
# Original CWD is restored