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 = <function _scenario_key>

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 – The scenario object.

Returns:

A string key unique to the scenario.

behave_priority.hooks.setup_priority(context: Any, *, order: bool = False, reverse: bool = False, priority_tag: str | None = None, stop_after_failures: int | None = None, stop_on_critical: bool = False, critical_tag: str = 'critical', default_priority: int = 999, report: bool = False, report_format: Literal['text', 'json', 'csv'] = 'text', parallel_coord: bool = False) → None

Set up priority execution in before_all hook.

All configuration is passed explicitly — no CLI flags.

Parameters:

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

• order – Sort scenarios by priority (highest first).

• reverse – Reverse sort order (lowest priority first).

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

• stop_after_failures – Stop after N failed scenarios.

• stop_on_critical – Stop if any critical scenario fails.

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

• default_priority – Priority for scenarios without a priority tag.

• report – Print execution report after run.

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

• parallel_coord – Enable cross-process fail-fast coordination. Requires BEHAVE_PRIORITY_COORD_DIR env var to be set.

behave_priority.hooks.before_scenario_hook(context: Any, scenario: Any) → None

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 – Behave’s context object.

• scenario – The scenario about to run.

behave_priority.hooks.after_scenario_hook(context: Any, scenario: Any) → None

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 – Behave’s context object.

• scenario – The scenario that just finished.

behave_priority.hooks.get_report(context: Any) → PriorityReport | None

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 – Behave’s context object.

Returns:

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

behave_priority.hooks.priority_report(context: Any) → None

Print the priority execution report.

Intended for use as after_all in behave’s environment.py.

Parameters:

context – Behave’s context object.

PriorityState

class behave_priority.hooks.PriorityState(config: PriorityConfig, report: PriorityReport, sorter: ScenarioSorter, coordinator: ParallelCoordinator | None = None, failed_count: int = 0, critical_failed: bool = False, should_stop: bool = False, executed_count: int = 0, skipped_count: int = 0, priority_map: dict[str, int]=<factory>, feature_map: dict[str, str]=<factory>, rule_tag_map: dict[str, list[str]]=<factory>, feature_tag_map: dict[str, list[str]]=<factory>)

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 (behave_priority.config.PriorityConfig) – The priority configuration.

• report (behave_priority.report.PriorityReport) – The execution report collector.

• sorter (behave_priority.sorter.ScenarioSorter) – The scenario sorter instance.

• coordinator (behave_priority.parallel.ParallelCoordinator | None) – Optional parallel coordinator for cross-process fail-fast. None when parallel_coord is disabled.

• 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.

• 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.

check_fail_fast() → bool

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.

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.
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).

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")