Behave Hooks

The hooks module provides the integration layer between behave-priority and behave’s lifecycle. It exposes functions for before_all, before_scenario, after_scenario, and after_all hooks.

behave_priority.hooks._scenario_key(scenario)

Build a deterministic key for a scenario.

Uses filename and line attributes when available (as behave provides), falling back to id() for objects without them.

Parameters:

scenario (Any) – The scenario object.

Return type:

str

Returns:

A string key unique to the scenario.

behave_priority.hooks.setup_priority(context, *, order=None, reverse=None, priority_tag=None, stop_after_failures=None, stop_on_critical=None, critical_tag='critical', default_priority=999, report=None, report_format='text', parallel_coord=None)

Set up priority execution in before_all hook.

Configuration can be passed explicitly or via environment variables when arguments are None.

Parameters:

• context (Any) – Behave’s context object (context in before_all).

• order (bool | None) – Sort scenarios by priority (highest first). If None, reads BEHAVE_PRIORITY_ORDER (default 0).

• reverse (bool | None) – Reverse sort order (lowest priority first). If None, reads BEHAVE_PRIORITY_REVERSE (default 0).

• priority_tag (str | None) – Tag name to run first (e.g. "smoke").

• stop_after_failures (int | None) – Stop after N failed scenarios. If None, reads BEHAVE_PRIORITY_STOP_AFTER (default disabled).

• stop_on_critical (bool | None) – Stop if any critical scenario fails. If None, reads BEHAVE_PRIORITY_FAIL_FAST (default 0).

• critical_tag (str) – Tag name that marks a scenario as critical.

• default_priority (int) – Priority for scenarios without a priority tag.

• report (bool | None) – Print execution report after run. If None, reads BEHAVE_PRIORITY_REPORT (default 0).

• report_format (Literal['text', 'json', 'csv']) – Output format for the report ("text", "json", "csv").

• parallel_coord (bool | None) – Enable cross-process fail-fast coordination. If None, reads BEHAVE_PRIORITY_PARALLEL_COORD (default 0). Requires BEHAVE_PRIORITY_COORD_DIR env var to be set.

Return type:

None

behave_priority.hooks.before_scenario_hook(context, scenario)

Skip scenario if fail-fast has been triggered.

Intended for use as before_scenario in behave’s environment.py. Recording of skipped scenarios is handled by after_scenario_hook to avoid duplicate entries.

Parameters:

• context (Any) – Behave’s context object.

• scenario (Any) – The scenario about to run.

Return type:

None

behave_priority.hooks.after_scenario_hook(context, scenario)

Record scenario result and update fail-fast state.

Intended for use as after_scenario in behave’s environment.py. Both executed and skipped scenarios are recorded here to avoid duplicate entries.

Parameters:

• context (Any) – Behave’s context object.

• scenario (Any) – The scenario that just finished.

Return type:

None

behave_priority.hooks.get_report(context)

Retrieve the priority execution report from context.

Allows programmatic access to the PriorityReport object after a run, without accessing the private context._priority_state.

Parameters:

context (Any) – Behave’s context object.

Return type:

PriorityReport | None

Returns:

The PriorityReport if priority state was set up, otherwise None.

behave_priority.hooks.priority_report(context)

Print the priority execution report.

Intended for use as after_all in behave’s environment.py. Writes to sys.__stdout__ (falling back to sys.stdout) so the report stays visible even when behave’s output capture is enabled, which is the default — behave captures stdout/stderr during hooks.

Parameters:

context (Any) – Behave’s context object.

Return type:

None

behave_priority.hooks.cleanup_parallel_coord(context)

Clean up the parallel coordinator for this worker.

Intended for use in after_all alongside priority_report. Removes this worker’s file from the coordination directory.

Parameters:

context (Any) – Behave’s context object.

Return type:

None

PriorityState

class behave_priority.hooks.PriorityState(config, report, sorter, coordinator=None, failed_count=0, critical_failed=False, should_stop=False, executed_count=0, skipped_count=0, priority_map=<factory>, feature_map=<factory>, rule_tag_map=<factory>, feature_tag_map=<factory>, missing_skip_warned=False)

Bases: object

Mutable execution state, persisted across hooks via context.

Note

This state is not shared across processes. When behave runs with --parallel, each worker process gets its own isolated PriorityState. However, when parallel_coord=True and BEHAVE_PRIORITY_COORD_DIR is set, a ParallelCoordinator shares fail-fast state across workers via file-based IPC.

Variables:

• config – The priority configuration.

• report – The execution report collector.

• sorter – The scenario sorter instance.

• coordinator – Optional parallel coordinator for cross-process fail-fast. None when parallel_coord is disabled.

• failed_count – Number of failed scenarios so far.

• critical_failed – Whether any critical scenario has failed.

• should_stop – Whether fail-fast conditions have been triggered.

• executed_count – Number of scenarios actually executed.

• skipped_count – Number of scenarios skipped by fail-fast.

• priority_map – Maps scenario key to resolved priority.

• feature_map – Maps scenario key to parent feature name.

• rule_tag_map – Maps scenario key to parent rule tags (Gherkin v6).

• feature_tag_map – Maps scenario key to parent feature tags.

• missing_skip_warned – Whether the missing scenario.skip() warning has already been emitted (warned once per run).

check_fail_fast()

Check if fail-fast conditions are met.

When a parallel coordinator is active, global failure counts across all workers are considered in addition to local state.

Return type:

bool

Returns:

True if execution should stop after the current scenario.

The PriorityState dataclass holds all mutable execution state that persists across hooks via context._priority_state.

Note

This state is not shared across processes. When behave runs with --parallel, each worker process gets its own isolated PriorityState. Fail-fast, counters, and reports are per-process and not coordinated across workers.

State Fields

PriorityState Fields

Field	Type	Description
config	PriorityConfig	The priority configuration for this run.
report	PriorityReport	The execution report collector.
sorter	ScenarioSorter	The sorter instance used to reorder scenarios.
coordinator	ParallelCoordinator | None	Optional parallel coordinator for cross-process fail-fast.
failed_count	int	Number of failed scenarios so far.
critical_failed	bool	Whether any critical scenario has failed.
should_stop	bool	Whether fail-fast conditions have been triggered.
executed_count	int	Number of scenarios actually executed (excludes skipped).
skipped_count	int	Number of scenarios skipped by fail-fast.
priority_map	dict[str, int]	Maps scenario key to resolved priority.
feature_map	dict[str, str]	Maps scenario key to parent feature name.
rule_tag_map	dict[str, list[str]]	Maps scenario key to parent rule tags (Gherkin v6).
feature_tag_map	dict[str, list[str]]	Maps scenario key to parent feature tags.
missing_skip_warned	bool	Whether the missing scenario.skip() warning was already emitted.

Fail-Fast Logic

The check_fail_fast method evaluates two conditions:

1. stop_after_failures: If failed_count >= stop_after_failures, returns True.

2. stop_on_critical: If stop_on_critical is enabled and critical_failed is True, returns True.

If either condition is met, should_stop is set to True and all subsequent scenarios are skipped via before_scenario_hook.

Scenario Key Generation

The internal _scenario_key function builds a deterministic key for each scenario:

• If the scenario has filename and line attributes (as behave provides), the key is "{filename}:{line}".

• Otherwise, falls back to "id:{id(scenario)}".

This key is used to look up priority, feature name, and rule tags in the state maps.

Hook Lifecycle

before_all(context)
  └─ setup_priority(context, ...)
       ├─ Create PriorityConfig
       ├─ Access runner.features
       ├─ Sort features & scenarios
       ├─ Populate priority_map, feature_map, rule_tag_map
       └─ Store PriorityState in context._priority_state

before_scenario(context, scenario)
  └─ before_scenario_hook(context, scenario)
       └─ If should_stop: scenario.skip("fail-fast triggered")

after_scenario(context, scenario)
  └─ after_scenario_hook(context, scenario)
       ├─ Record scenario in report
       ├─ If skipped: update should_stop, return
       ├─ Increment executed_count
       ├─ If failed: increment failed_count, check critical
       └─ Update should_stop

after_all(context)
  └─ priority_report(context)
       └─ If config.report: print report

Integration Example

from behave_priority import (
    setup_priority,
    before_scenario_hook,
    after_scenario_hook,
    priority_report,
    get_report,
)


def before_all(context):
    setup_priority(
        context,
        order=True,
        stop_after_failures=3,
        stop_on_critical=True,
        report=True,
    )


def before_scenario(context, scenario):
    before_scenario_hook(context, scenario)


def after_scenario(context, scenario):
    after_scenario_hook(context, scenario)


def after_all(context):
    priority_report(context)
    # Optionally access the report programmatically:
    report = get_report(context)
    if report:
        data = report.to_dict()
        print(f"Pass rate: {data['summary']['pass_rate']:.1f}%")

Programmatic Report Access

Use get_report() to access the PriorityReport object after a run, without accessing context._priority_state directly:

from behave_priority import get_report

def after_all(context):
    report = get_report(context)
    if report:
        summary = report.summary()
        print(f"Executed: {summary.passed} passed, {summary.failed} failed")
        print(f"Time saved: {summary.time_saved:.2f}s")