API reference

This page documents the full public API of behave-pool. Each module is documented with its classes, methods, and usage examples.

Package

behave_pool

behave-pool: parallel test execution for Behave BDD via native ITestRunner.

Public API:

from behave_pool import ParallelRunner

# Register in behave.ini:
# [behave.runners]
# parallel = behave_pool:ParallelRunner

# Then run:
# behave --runner=parallel --parallel 4 --parallel-scheme feature features/

ParallelRunner

Bases: Runner

Coordinator that dispatches work units to worker processes.

When config.parallel <= 1 it falls back to the standard Behave sequential runner. Otherwise it plans, dispatches, and collects results from N worker processes.

Source code in behave_pool/runner.py

class ParallelRunner(Runner):  # type: ignore[misc]
    """Coordinator that dispatches work units to worker processes.

    When ``config.parallel <= 1`` it falls back to the standard Behave
    sequential runner.  Otherwise it plans, dispatches, and collects
    results from N worker processes.
    """

    def __init__(self, config: Configuration) -> None:
        super().__init__(config)
        add_parallel_options(config)

    def run(self) -> bool:
        """Run the test suite — parallel or sequential depending on config."""
        with self.path_manager:
            self.setup_paths()
            return self.run_with_paths()

    def run_with_paths(self) -> bool:
        """Run tests with configured paths.

        If ``config.parallel <= 1`` delegates to the standard sequential
        runner.  Otherwise enters the parallel pipeline.
        """
        if self.config.parallel <= 1:
            return self._run_sequential()

        return self._run_parallel()

    def _run_sequential(self) -> bool:
        """Standard Behave sequential execution."""
        from behave.runner import Context

        self.context = Context(self)
        self.load_hooks()
        self.load_step_definitions()

        feature_locations = [
            filename for filename in self.feature_locations() if not self.config.exclude(filename)
        ]
        features = parse_features(feature_locations, language=self.config.lang)
        self.features.extend(features)

        stream_openers = self.config.outputs
        self.formatters = make_formatters(self.config, stream_openers)
        failed: bool = self.run_model()
        return failed

    def _run_parallel(self) -> bool:
        """Execute the parallel pipeline: plan -> split -> dispatch -> collect."""
        ctx = multiprocessing.get_context("spawn")
        task_queue: Any = ctx.JoinableQueue()
        result_queue: Any = ctx.Queue()
        stop_event: Any = ctx.Event()

        try:
            work_units = self._plan()
            parallel_batch, serial_batch = self._split_by_serial_tag(work_units)
            dispatched = self._dispatch(
                task_queue, result_queue, stop_event, parallel_batch, serial_batch
            )
            return self._collect(result_queue, dispatched)
        finally:
            stop_event.set()
            task_queue.close()
            result_queue.close()

    def _plan(self) -> list[WorkUnit]:
        """Parse features and create work units.

        Returns:
            List of work units to execute.
        """
        from behave.runner import Context

        self.context = Context(self)
        self.load_hooks()

        feature_locations = [
            filename for filename in self.feature_locations() if not self.config.exclude(filename)
        ]
        features = parse_features(feature_locations, language=self.config.lang)
        self.features.extend(features)

        iterator = WorkUnitIterator.for_scheme(
            scheme=self.config.parallel_scheme,
            features=features,
            config=self.config,
        )
        work_units = list(iterator.iterate())
        work_units = self._sort_by_duration(work_units)

        return work_units

    def _sort_by_duration(self, units: list[WorkUnit]) -> list[WorkUnit]:
        """Sort work units by historical duration (LPT) or keep FIFO order.

        When ``config.parallel_balance`` is ``"lpt"``, units are sorted
        descending by their stored duration in the TimingStore so that
        the longest jobs start first, improving overall wall-clock time.

        When ``config.parallel_balance`` is ``"fifo"``, the original
        order is preserved.

        Args:
            units: Work units to sort.

        Returns:
            Sorted list of work units.
        """
        balance = getattr(self.config, "parallel_balance", "lpt")
        if balance == "fifo":
            return units

        timing_file = (
            getattr(self.config, "parallel_timing_file", None) or ".behave-pool-timing.json"
        )
        store = TimingStore(path=Path(timing_file))
        store.load()
        return sorted(units, key=lambda u: store.get_duration(u.id), reverse=True)

    @staticmethod
    def _split_by_serial_tag(
        units: list[WorkUnit],
    ) -> tuple[list[WorkUnit], list[WorkUnit]]:
        """Split work units into parallel and serial batches.

        Args:
            units: All work units to split.

        Returns:
            Tuple of (parallel_batch, serial_batch).
        """
        parallel_batch = [u for u in units if not u.is_serial]
        serial_batch = [u for u in units if u.is_serial]
        return parallel_batch, serial_batch

    def _dispatch(
        self,
        task_queue: Any,
        result_queue: Any,
        stop_event: Any,
        parallel_batch: list[WorkUnit],
        serial_batch: list[WorkUnit],
    ) -> list[WorkUnit]:
        """Two-phase dispatch: parallel first, then serial.

        Phase 1: enqueue parallel_batch, launch N workers, wait for completion.
        Phase 2: enqueue serial_batch one at a time, launch 1 worker, wait.

        Returns:
            List of work units that were actually enqueued (dispatched).
        """
        n_workers = self.config.parallel
        config_snapshot = snapshot_config(self.config)
        dispatched: list[WorkUnit] = []
        ctx = multiprocessing.get_context("spawn")

        # -- Phase 1: parallel batch with N workers.
        if parallel_batch:
            for unit in parallel_batch:
                task_queue.put(unit)
            for _ in range(n_workers):
                task_queue.put(None)
            dispatched.extend(parallel_batch)

            workers: list[WorkerProcess] = []
            for worker_id in range(n_workers):
                worker = WorkerProcess(
                    worker_id=worker_id,
                    task_queue=task_queue,
                    result_queue=result_queue,
                    stop_event=stop_event,
                    config_snapshot=config_snapshot,
                    ctx=ctx,
                )
                worker.start()
                workers.append(worker)

            for worker in workers:
                worker.join(timeout=300)
                if worker.is_alive():
                    logger.warning(
                        "Worker %d did not terminate within 300s; "
                        "setting stop event and terminating.",
                        worker.worker_id,
                    )
                    stop_event.set()
                    worker.terminate()

            # Drain any unconsumed items so the queue is empty for Phase 2.
            while not task_queue.empty():
                try:
                    task_queue.get_nowait()
                    task_queue.task_done()
                except queue.Empty:
                    break

        # -- Phase 2: serial batch with 1 worker.
        if serial_batch and not stop_event.is_set():
            for unit in serial_batch:
                task_queue.put(unit)
            task_queue.put(None)
            dispatched.extend(serial_batch)

            serial_worker = WorkerProcess(
                worker_id=0,
                task_queue=task_queue,
                result_queue=result_queue,
                stop_event=stop_event,
                config_snapshot=config_snapshot,
                ctx=ctx,
            )
            serial_worker.start()
            serial_worker.join(timeout=300)
            if serial_worker.is_alive():
                logger.warning(
                    "Serial worker did not terminate within 300s; "
                    "setting stop event and terminating."
                )
                stop_event.set()
                serial_worker.terminate()

            # Drain any unconsumed items.
            while not task_queue.empty():
                try:
                    task_queue.get_nowait()
                    task_queue.task_done()
                except queue.Empty:
                    break

        return dispatched

    def _collect(
        self,
        result_queue: Any,
        work_units: list[WorkUnit],
        deadline_seconds: float = 30,
    ) -> bool:
        """Drain result queue, merge results, and compute exit code.

        Returns:
            True if any test failed (Behave convention).
        """
        expected = len(work_units)
        results: list[WorkerResult] = []
        received_ids: set[str] = set()

        # Drain all available results, waiting up to deadline_seconds for late arrivals.
        deadline = time.monotonic() + deadline_seconds
        while len(results) < expected and time.monotonic() < deadline:
            try:
                result = result_queue.get(timeout=1)
            except queue.Empty:
                continue
            except (EOFError, OSError):
                break
            results.append(result)
            received_ids.add(result.work_unit_id)

        # Detect missing results from crashed or timed-out workers.
        missing = [u.id for u in work_units if u.id not in received_ids]
        if missing:
            logger.warning(
                "Missing %d result(s) from worker(s): %s",
                len(missing),
                ", ".join(missing),
            )

        any_failed = any(r.failed for r in results)

        # Missing results indicate worker crashes — treat as failures.
        if missing:
            any_failed = True

        self._update_timings(results)

        self._merge_reports(results)

        logger.info(
            "Parallel run complete: %d work units, %d results, failed=%s",
            len(work_units),
            len(results),
            any_failed,
        )

        return any_failed

    def _update_timings(self, results: list[WorkerResult]) -> None:
        """Update the TimingStore with observed durations from results.

        Timing persistence is best-effort: any failure is logged and
        does not affect the test run outcome.

        Args:
            results: Worker results containing durations to persist.
        """
        timing_file = (
            getattr(self.config, "parallel_timing_file", None) or ".behave-pool-timing.json"
        )
        try:
            store = TimingStore(path=Path(timing_file))
            store.load()
            for result in results:
                store.update(result.work_unit_id, result.duration)
            store.save_if_changed()
        except Exception:
            logger.warning(
                "Failed to update timing file %s; timings will not persist.", timing_file
            )

    def _merge_reports(self, results: list[WorkerResult]) -> None:
        """Merge per-worker JSON reports into a unified Behave-compatible JSON.

        Reads each worker's report file (pointed to by WorkerResult.report_path),
        collects all feature dicts, computes aggregate statistics, detects the
        runtime environment, and writes a full behave-modern-json-report
        ExecutionReport JSON to the path specified by ``--parallel-report``.

        After merging, the temporary ``tmp/`` directory is cleaned up.

        Args:
            results: Worker results with report paths to merge.
        """
        import json

        all_features: list[dict[str, Any]] = []

        for result in results:
            if not result.report_path:
                continue
            try:
                report_file = Path(result.report_path)
                if report_file.exists():
                    data = json.loads(report_file.read_text(encoding="utf-8"))
                    all_features.extend(data.get("features", []))
            except Exception:
                logger.warning("Failed to read worker report %s; skipping.", result.report_path)

        statistics = self._compute_statistics(all_features)
        environment = self._detect_environment()
        execution = self._build_execution(results)

        report = {
            "schemaVersion": "1.1.0",
            "execution": execution,
            "statistics": statistics,
            "environment": environment,
            "features": all_features,
            "metadata": {},
        }

        report_path = Path(
            getattr(self.config, "parallel_report", None) or "behave-pool-report.json"
        )
        try:
            report_path.write_text(
                json.dumps(report, indent=2, ensure_ascii=False),
                encoding="utf-8",
            )
            logger.info("Unified report written to %s", report_path)
        except Exception:
            logger.warning("Failed to write unified report to %s", report_path)

        tmp_dir = Path("tmp")
        if tmp_dir.is_dir():
            shutil.rmtree(tmp_dir, ignore_errors=True)

    def _compute_statistics(self, features: list[dict[str, Any]]) -> dict[str, Any]:
        """Compute aggregate statistics from merged feature dicts."""
        _status_fields = {
            "passed": "passed",
            "failed": "failed",
            "skipped": "skipped",
            "undefined": "undefined",
            "pending": "pending",
        }
        _failed_statuses = frozenset({"failed", "error", "hook_error", "cleanup_error"})

        feature_count = 0
        scenario_count = 0
        step_count = 0
        counts: dict[str, int] = dict.fromkeys(_status_fields.values(), 0)
        total_duration = 0.0
        error_count = 0
        slowest_step_duration = 0.0
        all_durations: list[float] = []
        exception_counts: dict[str, int] = {}
        by_tag: dict[str, dict[str, Any]] = {}

        for feature in features:
            feature_count += 1
            feature_duration = 0.0

            for scenario in feature.get("scenarios", []) or []:
                scenario_count += 1
                scenario_duration = 0.0

                for step in scenario.get("steps", []) or []:
                    step_count += 1
                    status = step.get("status", "untested")
                    field_name = _status_fields.get(status)
                    if field_name is not None:
                        counts[field_name] += 1
                    step_duration = step.get("duration", 0.0) or 0.0
                    scenario_duration += step_duration
                    if status in _failed_statuses:
                        error_count += 1
                    slowest_step_duration = max(slowest_step_duration, step_duration)
                    error = step.get("error")
                    if error and error.get("type"):
                        etype = error["type"]
                        exception_counts[etype] = exception_counts.get(etype, 0) + 1

                scenario_duration = scenario.get("duration", 0.0) or scenario_duration
                all_durations.append(scenario_duration)
                feature_duration += scenario_duration

                scenario_tags = set(scenario.get("tags", []) or [])
                feature_tags = set(feature.get("tags", []) or [])
                for tag in scenario_tags | feature_tags:
                    tag_data = by_tag.setdefault(
                        tag,
                        {
                            "count": 0,
                            "duration": 0.0,
                            "passed": 0,
                            "failed": 0,
                            "skipped": 0,
                            "undefined": 0,
                            "pending": 0,
                            "untested": 0,
                            "error": 0,
                            "hook_error": 0,
                            "cleanup_error": 0,
                            "xfailed": 0,
                            "xpassed": 0,
                        },
                    )
                    tag_data["count"] += 1
                    tag_data["duration"] += scenario_duration
                    s = scenario.get("status", "passed")
                    if s in tag_data:
                        tag_data[s] += 1

            total_duration += feature.get("duration", 0.0) or feature_duration

        total_terminal = counts["passed"] + counts["failed"]
        pass_rate = (counts["passed"] / total_terminal) if total_terminal else 0.0
        avg_scenario_duration = sum(all_durations) / len(all_durations) if all_durations else 0.0
        common_exception_type = (
            max(exception_counts, key=lambda k: exception_counts.get(k, 0))
            if exception_counts
            else None
        )

        stats: dict[str, Any] = {
            "features": feature_count,
            "scenarios": scenario_count,
            "steps": step_count,
            "passed": counts["passed"],
            "failed": counts["failed"],
            "skipped": counts["skipped"],
            "undefined": counts["undefined"],
            "pending": counts["pending"],
            "passRate": round(pass_rate, 6),
            "duration": round(total_duration, 6),
            "errorCount": error_count,
            "totalAttachments": 0,
            "totalLogs": 0,
            "slowestStepDuration": round(slowest_step_duration, 6),
            "avgScenarioDuration": round(avg_scenario_duration, 6),
            "byTag": by_tag,
        }
        if common_exception_type is not None:
            stats["commonExceptionType"] = common_exception_type
        return stats

    def _detect_environment(self) -> dict[str, Any]:
        """Detect runtime environment for the report."""
        import os
        import platform as _platform
        import socket
        import subprocess
        import sys

        env: dict[str, Any] = {}

        env["pythonVersion"] = sys.version.split(" ", 1)[0]
        env["platform"] = sys.platform
        env["os"] = _platform.system() or "Unknown"
        env["osVersion"] = _platform.release() or "Unknown"

        try:
            env["hostname"] = socket.gethostname() or "unknown"
        except Exception:
            env["hostname"] = "unknown"

        ci_env = os.environ
        if ci_env.get("GITHUB_ACTIONS") == "true":
            env["ciProvider"] = "github-actions"
        elif ci_env.get("GITLAB_CI"):
            env["ciProvider"] = "gitlab-ci"
        elif ci_env.get("JENKINS_URL"):
            env["ciProvider"] = "jenkins"
        elif ci_env.get("CI"):
            env["ciProvider"] = "ci"

        cwd = os.getcwd()
        if cwd:
            env["cwd"] = cwd

        cmd = " ".join(sys.argv)
        if cmd:
            env["command"] = cmd

        try:
            import getpass

            env["user"] = getpass.getuser()
        except Exception:
            pass

        cpu = os.cpu_count()
        if cpu:
            env["cpuCount"] = cpu

        try:
            import behave

            bv = str(getattr(behave, "__version__", "") or "")
            if bv:
                env["behaveVersion"] = bv
        except Exception:
            pass

        git_info: dict[str, str] = {}
        try:
            for key, git_cmd in [
                ("branch", ["git", "rev-parse", "--abbrev-ref", "HEAD"]),
                ("commit", ["git", "rev-parse", "--short", "HEAD"]),
                ("remote", ["git", "remote", "get-url", "origin"]),
            ]:
                result = subprocess.run(
                    git_cmd, capture_output=True, text=True, timeout=2, check=False
                )
                if result.returncode == 0:
                    git_info[key] = result.stdout.strip()
        except Exception:
            pass

        if git_info.get("branch"):
            env["gitBranch"] = git_info["branch"]
        if git_info.get("commit"):
            env["gitCommit"] = git_info["commit"]
        if git_info.get("remote"):
            env["gitRemote"] = git_info["remote"]

        return env

    def _build_execution(self, results: list[WorkerResult]) -> dict[str, Any]:
        """Build the execution metadata block."""
        import uuid
        from datetime import UTC, datetime

        now = datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
        total_duration = sum(r.duration for r in results)
        any_failed = any(r.failed for r in results)

        execution: dict[str, Any] = {
            "executionId": f"exec-{uuid.uuid4().hex}",
            "status": "failed" if any_failed else "passed",
            "duration": round(total_duration, 6),
            "startTime": now,
            "endTime": now,
        }
        return execution

run

run() -> bool

Run the test suite — parallel or sequential depending on config.

Source code in behave_pool/runner.py

def run(self) -> bool:
    """Run the test suite — parallel or sequential depending on config."""
    with self.path_manager:
        self.setup_paths()
        return self.run_with_paths()

run_with_paths

run_with_paths() -> bool

Run tests with configured paths.

If config.parallel <= 1 delegates to the standard sequential runner. Otherwise enters the parallel pipeline.

Source code in behave_pool/runner.py

def run_with_paths(self) -> bool:
    """Run tests with configured paths.

    If ``config.parallel <= 1`` delegates to the standard sequential
    runner.  Otherwise enters the parallel pipeline.
    """
    if self.config.parallel <= 1:
        return self._run_sequential()

    return self._run_parallel()

ParallelRunner

The coordinator that orchestrates parallel feature execution. Extends behave.runner.Runner and implements Behave's ITestRunner interface.

from behave_pool import ParallelRunner
from behave.configuration import Configuration

config = Configuration(["--parallel", "4", "features/"])
runner = ParallelRunner(config)
failed = runner.run()  # True if any test failed

behave_pool.runner

ParallelRunner: coordinator that orchestrates parallel feature execution.

ParallelRunner

Bases: Runner

Coordinator that dispatches work units to worker processes.

When config.parallel <= 1 it falls back to the standard Behave sequential runner. Otherwise it plans, dispatches, and collects results from N worker processes.

Source code in behave_pool/runner.py

class ParallelRunner(Runner):  # type: ignore[misc]
    """Coordinator that dispatches work units to worker processes.

    When ``config.parallel <= 1`` it falls back to the standard Behave
    sequential runner.  Otherwise it plans, dispatches, and collects
    results from N worker processes.
    """

    def __init__(self, config: Configuration) -> None:
        super().__init__(config)
        add_parallel_options(config)

    def run(self) -> bool:
        """Run the test suite — parallel or sequential depending on config."""
        with self.path_manager:
            self.setup_paths()
            return self.run_with_paths()

    def run_with_paths(self) -> bool:
        """Run tests with configured paths.

        If ``config.parallel <= 1`` delegates to the standard sequential
        runner.  Otherwise enters the parallel pipeline.
        """
        if self.config.parallel <= 1:
            return self._run_sequential()

        return self._run_parallel()

    def _run_sequential(self) -> bool:
        """Standard Behave sequential execution."""
        from behave.runner import Context

        self.context = Context(self)
        self.load_hooks()
        self.load_step_definitions()

        feature_locations = [
            filename for filename in self.feature_locations() if not self.config.exclude(filename)
        ]
        features = parse_features(feature_locations, language=self.config.lang)
        self.features.extend(features)

        stream_openers = self.config.outputs
        self.formatters = make_formatters(self.config, stream_openers)
        failed: bool = self.run_model()
        return failed

    def _run_parallel(self) -> bool:
        """Execute the parallel pipeline: plan -> split -> dispatch -> collect."""
        ctx = multiprocessing.get_context("spawn")
        task_queue: Any = ctx.JoinableQueue()
        result_queue: Any = ctx.Queue()
        stop_event: Any = ctx.Event()

        try:
            work_units = self._plan()
            parallel_batch, serial_batch = self._split_by_serial_tag(work_units)
            dispatched = self._dispatch(
                task_queue, result_queue, stop_event, parallel_batch, serial_batch
            )
            return self._collect(result_queue, dispatched)
        finally:
            stop_event.set()
            task_queue.close()
            result_queue.close()

    def _plan(self) -> list[WorkUnit]:
        """Parse features and create work units.

        Returns:
            List of work units to execute.
        """
        from behave.runner import Context

        self.context = Context(self)
        self.load_hooks()

        feature_locations = [
            filename for filename in self.feature_locations() if not self.config.exclude(filename)
        ]
        features = parse_features(feature_locations, language=self.config.lang)
        self.features.extend(features)

        iterator = WorkUnitIterator.for_scheme(
            scheme=self.config.parallel_scheme,
            features=features,
            config=self.config,
        )
        work_units = list(iterator.iterate())
        work_units = self._sort_by_duration(work_units)

        return work_units

    def _sort_by_duration(self, units: list[WorkUnit]) -> list[WorkUnit]:
        """Sort work units by historical duration (LPT) or keep FIFO order.

        When ``config.parallel_balance`` is ``"lpt"``, units are sorted
        descending by their stored duration in the TimingStore so that
        the longest jobs start first, improving overall wall-clock time.

        When ``config.parallel_balance`` is ``"fifo"``, the original
        order is preserved.

        Args:
            units: Work units to sort.

        Returns:
            Sorted list of work units.
        """
        balance = getattr(self.config, "parallel_balance", "lpt")
        if balance == "fifo":
            return units

        timing_file = (
            getattr(self.config, "parallel_timing_file", None) or ".behave-pool-timing.json"
        )
        store = TimingStore(path=Path(timing_file))
        store.load()
        return sorted(units, key=lambda u: store.get_duration(u.id), reverse=True)

    @staticmethod
    def _split_by_serial_tag(
        units: list[WorkUnit],
    ) -> tuple[list[WorkUnit], list[WorkUnit]]:
        """Split work units into parallel and serial batches.

        Args:
            units: All work units to split.

        Returns:
            Tuple of (parallel_batch, serial_batch).
        """
        parallel_batch = [u for u in units if not u.is_serial]
        serial_batch = [u for u in units if u.is_serial]
        return parallel_batch, serial_batch

    def _dispatch(
        self,
        task_queue: Any,
        result_queue: Any,
        stop_event: Any,
        parallel_batch: list[WorkUnit],
        serial_batch: list[WorkUnit],
    ) -> list[WorkUnit]:
        """Two-phase dispatch: parallel first, then serial.

        Phase 1: enqueue parallel_batch, launch N workers, wait for completion.
        Phase 2: enqueue serial_batch one at a time, launch 1 worker, wait.

        Returns:
            List of work units that were actually enqueued (dispatched).
        """
        n_workers = self.config.parallel
        config_snapshot = snapshot_config(self.config)
        dispatched: list[WorkUnit] = []
        ctx = multiprocessing.get_context("spawn")

        # -- Phase 1: parallel batch with N workers.
        if parallel_batch:
            for unit in parallel_batch:
                task_queue.put(unit)
            for _ in range(n_workers):
                task_queue.put(None)
            dispatched.extend(parallel_batch)

            workers: list[WorkerProcess] = []
            for worker_id in range(n_workers):
                worker = WorkerProcess(
                    worker_id=worker_id,
                    task_queue=task_queue,
                    result_queue=result_queue,
                    stop_event=stop_event,
                    config_snapshot=config_snapshot,
                    ctx=ctx,
                )
                worker.start()
                workers.append(worker)

            for worker in workers:
                worker.join(timeout=300)
                if worker.is_alive():
                    logger.warning(
                        "Worker %d did not terminate within 300s; "
                        "setting stop event and terminating.",
                        worker.worker_id,
                    )
                    stop_event.set()
                    worker.terminate()

            # Drain any unconsumed items so the queue is empty for Phase 2.
            while not task_queue.empty():
                try:
                    task_queue.get_nowait()
                    task_queue.task_done()
                except queue.Empty:
                    break

        # -- Phase 2: serial batch with 1 worker.
        if serial_batch and not stop_event.is_set():
            for unit in serial_batch:
                task_queue.put(unit)
            task_queue.put(None)
            dispatched.extend(serial_batch)

            serial_worker = WorkerProcess(
                worker_id=0,
                task_queue=task_queue,
                result_queue=result_queue,
                stop_event=stop_event,
                config_snapshot=config_snapshot,
                ctx=ctx,
            )
            serial_worker.start()
            serial_worker.join(timeout=300)
            if serial_worker.is_alive():
                logger.warning(
                    "Serial worker did not terminate within 300s; "
                    "setting stop event and terminating."
                )
                stop_event.set()
                serial_worker.terminate()

            # Drain any unconsumed items.
            while not task_queue.empty():
                try:
                    task_queue.get_nowait()
                    task_queue.task_done()
                except queue.Empty:
                    break

        return dispatched

    def _collect(
        self,
        result_queue: Any,
        work_units: list[WorkUnit],
        deadline_seconds: float = 30,
    ) -> bool:
        """Drain result queue, merge results, and compute exit code.

        Returns:
            True if any test failed (Behave convention).
        """
        expected = len(work_units)
        results: list[WorkerResult] = []
        received_ids: set[str] = set()

        # Drain all available results, waiting up to deadline_seconds for late arrivals.
        deadline = time.monotonic() + deadline_seconds
        while len(results) < expected and time.monotonic() < deadline:
            try:
                result = result_queue.get(timeout=1)
            except queue.Empty:
                continue
            except (EOFError, OSError):
                break
            results.append(result)
            received_ids.add(result.work_unit_id)

        # Detect missing results from crashed or timed-out workers.
        missing = [u.id for u in work_units if u.id not in received_ids]
        if missing:
            logger.warning(
                "Missing %d result(s) from worker(s): %s",
                len(missing),
                ", ".join(missing),
            )

        any_failed = any(r.failed for r in results)

        # Missing results indicate worker crashes — treat as failures.
        if missing:
            any_failed = True

        self._update_timings(results)

        self._merge_reports(results)

        logger.info(
            "Parallel run complete: %d work units, %d results, failed=%s",
            len(work_units),
            len(results),
            any_failed,
        )

        return any_failed

    def _update_timings(self, results: list[WorkerResult]) -> None:
        """Update the TimingStore with observed durations from results.

        Timing persistence is best-effort: any failure is logged and
        does not affect the test run outcome.

        Args:
            results: Worker results containing durations to persist.
        """
        timing_file = (
            getattr(self.config, "parallel_timing_file", None) or ".behave-pool-timing.json"
        )
        try:
            store = TimingStore(path=Path(timing_file))
            store.load()
            for result in results:
                store.update(result.work_unit_id, result.duration)
            store.save_if_changed()
        except Exception:
            logger.warning(
                "Failed to update timing file %s; timings will not persist.", timing_file
            )

    def _merge_reports(self, results: list[WorkerResult]) -> None:
        """Merge per-worker JSON reports into a unified Behave-compatible JSON.

        Reads each worker's report file (pointed to by WorkerResult.report_path),
        collects all feature dicts, computes aggregate statistics, detects the
        runtime environment, and writes a full behave-modern-json-report
        ExecutionReport JSON to the path specified by ``--parallel-report``.

        After merging, the temporary ``tmp/`` directory is cleaned up.

        Args:
            results: Worker results with report paths to merge.
        """
        import json

        all_features: list[dict[str, Any]] = []

        for result in results:
            if not result.report_path:
                continue
            try:
                report_file = Path(result.report_path)
                if report_file.exists():
                    data = json.loads(report_file.read_text(encoding="utf-8"))
                    all_features.extend(data.get("features", []))
            except Exception:
                logger.warning("Failed to read worker report %s; skipping.", result.report_path)

        statistics = self._compute_statistics(all_features)
        environment = self._detect_environment()
        execution = self._build_execution(results)

        report = {
            "schemaVersion": "1.1.0",
            "execution": execution,
            "statistics": statistics,
            "environment": environment,
            "features": all_features,
            "metadata": {},
        }

        report_path = Path(
            getattr(self.config, "parallel_report", None) or "behave-pool-report.json"
        )
        try:
            report_path.write_text(
                json.dumps(report, indent=2, ensure_ascii=False),
                encoding="utf-8",
            )
            logger.info("Unified report written to %s", report_path)
        except Exception:
            logger.warning("Failed to write unified report to %s", report_path)

        tmp_dir = Path("tmp")
        if tmp_dir.is_dir():
            shutil.rmtree(tmp_dir, ignore_errors=True)

    def _compute_statistics(self, features: list[dict[str, Any]]) -> dict[str, Any]:
        """Compute aggregate statistics from merged feature dicts."""
        _status_fields = {
            "passed": "passed",
            "failed": "failed",
            "skipped": "skipped",
            "undefined": "undefined",
            "pending": "pending",
        }
        _failed_statuses = frozenset({"failed", "error", "hook_error", "cleanup_error"})

        feature_count = 0
        scenario_count = 0
        step_count = 0
        counts: dict[str, int] = dict.fromkeys(_status_fields.values(), 0)
        total_duration = 0.0
        error_count = 0
        slowest_step_duration = 0.0
        all_durations: list[float] = []
        exception_counts: dict[str, int] = {}
        by_tag: dict[str, dict[str, Any]] = {}

        for feature in features:
            feature_count += 1
            feature_duration = 0.0

            for scenario in feature.get("scenarios", []) or []:
                scenario_count += 1
                scenario_duration = 0.0

                for step in scenario.get("steps", []) or []:
                    step_count += 1
                    status = step.get("status", "untested")
                    field_name = _status_fields.get(status)
                    if field_name is not None:
                        counts[field_name] += 1
                    step_duration = step.get("duration", 0.0) or 0.0
                    scenario_duration += step_duration
                    if status in _failed_statuses:
                        error_count += 1
                    slowest_step_duration = max(slowest_step_duration, step_duration)
                    error = step.get("error")
                    if error and error.get("type"):
                        etype = error["type"]
                        exception_counts[etype] = exception_counts.get(etype, 0) + 1

                scenario_duration = scenario.get("duration", 0.0) or scenario_duration
                all_durations.append(scenario_duration)
                feature_duration += scenario_duration

                scenario_tags = set(scenario.get("tags", []) or [])
                feature_tags = set(feature.get("tags", []) or [])
                for tag in scenario_tags | feature_tags:
                    tag_data = by_tag.setdefault(
                        tag,
                        {
                            "count": 0,
                            "duration": 0.0,
                            "passed": 0,
                            "failed": 0,
                            "skipped": 0,
                            "undefined": 0,
                            "pending": 0,
                            "untested": 0,
                            "error": 0,
                            "hook_error": 0,
                            "cleanup_error": 0,
                            "xfailed": 0,
                            "xpassed": 0,
                        },
                    )
                    tag_data["count"] += 1
                    tag_data["duration"] += scenario_duration
                    s = scenario.get("status", "passed")
                    if s in tag_data:
                        tag_data[s] += 1

            total_duration += feature.get("duration", 0.0) or feature_duration

        total_terminal = counts["passed"] + counts["failed"]
        pass_rate = (counts["passed"] / total_terminal) if total_terminal else 0.0
        avg_scenario_duration = sum(all_durations) / len(all_durations) if all_durations else 0.0
        common_exception_type = (
            max(exception_counts, key=lambda k: exception_counts.get(k, 0))
            if exception_counts
            else None
        )

        stats: dict[str, Any] = {
            "features": feature_count,
            "scenarios": scenario_count,
            "steps": step_count,
            "passed": counts["passed"],
            "failed": counts["failed"],
            "skipped": counts["skipped"],
            "undefined": counts["undefined"],
            "pending": counts["pending"],
            "passRate": round(pass_rate, 6),
            "duration": round(total_duration, 6),
            "errorCount": error_count,
            "totalAttachments": 0,
            "totalLogs": 0,
            "slowestStepDuration": round(slowest_step_duration, 6),
            "avgScenarioDuration": round(avg_scenario_duration, 6),
            "byTag": by_tag,
        }
        if common_exception_type is not None:
            stats["commonExceptionType"] = common_exception_type
        return stats

    def _detect_environment(self) -> dict[str, Any]:
        """Detect runtime environment for the report."""
        import os
        import platform as _platform
        import socket
        import subprocess
        import sys

        env: dict[str, Any] = {}

        env["pythonVersion"] = sys.version.split(" ", 1)[0]
        env["platform"] = sys.platform
        env["os"] = _platform.system() or "Unknown"
        env["osVersion"] = _platform.release() or "Unknown"

        try:
            env["hostname"] = socket.gethostname() or "unknown"
        except Exception:
            env["hostname"] = "unknown"

        ci_env = os.environ
        if ci_env.get("GITHUB_ACTIONS") == "true":
            env["ciProvider"] = "github-actions"
        elif ci_env.get("GITLAB_CI"):
            env["ciProvider"] = "gitlab-ci"
        elif ci_env.get("JENKINS_URL"):
            env["ciProvider"] = "jenkins"
        elif ci_env.get("CI"):
            env["ciProvider"] = "ci"

        cwd = os.getcwd()
        if cwd:
            env["cwd"] = cwd

        cmd = " ".join(sys.argv)
        if cmd:
            env["command"] = cmd

        try:
            import getpass

            env["user"] = getpass.getuser()
        except Exception:
            pass

        cpu = os.cpu_count()
        if cpu:
            env["cpuCount"] = cpu

        try:
            import behave

            bv = str(getattr(behave, "__version__", "") or "")
            if bv:
                env["behaveVersion"] = bv
        except Exception:
            pass

        git_info: dict[str, str] = {}
        try:
            for key, git_cmd in [
                ("branch", ["git", "rev-parse", "--abbrev-ref", "HEAD"]),
                ("commit", ["git", "rev-parse", "--short", "HEAD"]),
                ("remote", ["git", "remote", "get-url", "origin"]),
            ]:
                result = subprocess.run(
                    git_cmd, capture_output=True, text=True, timeout=2, check=False
                )
                if result.returncode == 0:
                    git_info[key] = result.stdout.strip()
        except Exception:
            pass

        if git_info.get("branch"):
            env["gitBranch"] = git_info["branch"]
        if git_info.get("commit"):
            env["gitCommit"] = git_info["commit"]
        if git_info.get("remote"):
            env["gitRemote"] = git_info["remote"]

        return env

    def _build_execution(self, results: list[WorkerResult]) -> dict[str, Any]:
        """Build the execution metadata block."""
        import uuid
        from datetime import UTC, datetime

        now = datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
        total_duration = sum(r.duration for r in results)
        any_failed = any(r.failed for r in results)

        execution: dict[str, Any] = {
            "executionId": f"exec-{uuid.uuid4().hex}",
            "status": "failed" if any_failed else "passed",
            "duration": round(total_duration, 6),
            "startTime": now,
            "endTime": now,
        }
        return execution

run

run() -> bool

Run the test suite — parallel or sequential depending on config.

Source code in behave_pool/runner.py

def run(self) -> bool:
    """Run the test suite — parallel or sequential depending on config."""
    with self.path_manager:
        self.setup_paths()
        return self.run_with_paths()

run_with_paths

run_with_paths() -> bool

Run tests with configured paths.

If config.parallel <= 1 delegates to the standard sequential runner. Otherwise enters the parallel pipeline.

Source code in behave_pool/runner.py

def run_with_paths(self) -> bool:
    """Run tests with configured paths.

    If ``config.parallel <= 1`` delegates to the standard sequential
    runner.  Otherwise enters the parallel pipeline.
    """
    if self.config.parallel <= 1:
        return self._run_sequential()

    return self._run_parallel()

Configuration

ConfigSnapshot is a picklable snapshot of Behave's Configuration that can be safely sent to worker processes via spawn.

from behave_pool.config import ConfigSnapshot, snapshot_config

snapshot = snapshot_config(config)
print(snapshot.parallel)  # 4
print(snapshot.parallel_scheme)  # "feature"
print(snapshot.base_dir)  # "features"

behave_pool.config

Parallel configuration options for behave-pool.

ConfigSnapshot dataclass

Picklable snapshot of essential Configuration fields for worker processes.

The full behave Configuration contains non-picklable objects (file handles, reporters). This snapshot captures only the fields needed by WorkerRunner to execute features.

Source code in behave_pool/config.py

@dataclass(frozen=True)
class ConfigSnapshot:
    """Picklable snapshot of essential Configuration fields for worker processes.

    The full behave Configuration contains non-picklable objects (file
    handles, reporters).  This snapshot captures only the fields needed
    by WorkerRunner to execute features.
    """

    base_dir: str
    steps_dir: str
    environment_file: str
    lang: str | None
    stop: bool
    paths: list[str] = field(default_factory=list)
    parallel: int = 1
    parallel_scheme: str = "feature"
    parallel_balance: str = "lpt"
    parallel_timing_file: str = ".behave-pool-timing.json"
    parallel_report: str = "behave-pool-report.json"
    dry_run: bool = False
    use_nested_step_modules: bool = False

add_parallel_options

add_parallel_options(config: Configuration) -> None

Add parallel-related attributes to a Configuration instance.

Maps behave's config.jobs (from --parallel/--jobs) to config.parallel and ensures config.parallel_scheme exists.

Parameters:

Name	Type	Description	Default
config	Configuration	Behave Configuration instance to augment.	required

Source code in behave_pool/config.py

def add_parallel_options(config: Configuration) -> None:
    """Add parallel-related attributes to a Configuration instance.

    Maps behave's ``config.jobs`` (from ``--parallel``/``--jobs``) to
    ``config.parallel`` and ensures ``config.parallel_scheme`` exists.

    Args:
        config: Behave Configuration instance to augment.
    """
    jobs = getattr(config, "jobs", 1)
    if jobs is None:
        jobs = 1
    config.parallel = jobs

    if not hasattr(config, "parallel_scheme") or config.parallel_scheme is None:
        config.parallel_scheme = "feature"

    if not hasattr(config, "parallel_balance") or config.parallel_balance is None:
        config.parallel_balance = "lpt"

    if not hasattr(config, "parallel_timing_file") or config.parallel_timing_file is None:
        config.parallel_timing_file = ".behave-pool-timing.json"

    if not hasattr(config, "parallel_report") or config.parallel_report is None:
        config.parallel_report = "behave-pool-report.json"

    if not hasattr(config, "use_nested_step_modules"):
        config.use_nested_step_modules = False

snapshot_config

snapshot_config(config: Configuration) -> ConfigSnapshot

Create a picklable snapshot from a Configuration instance.

Source code in behave_pool/config.py

def snapshot_config(config: Configuration) -> ConfigSnapshot:
    """Create a picklable snapshot from a Configuration instance."""
    return ConfigSnapshot(
        base_dir=str(getattr(config, "base_dir", None) or "features"),
        steps_dir=str(getattr(config, "steps_dir", None) or "steps"),
        environment_file=str(getattr(config, "environment_file", None) or "environment.py"),
        lang=config.lang,
        stop=config.stop,
        paths=[str(p) for p in config.paths] if config.paths else [],
        parallel=getattr(config, "parallel", None) or getattr(config, "jobs", None) or 1,
        parallel_scheme=getattr(config, "parallel_scheme", "feature"),
        parallel_balance=getattr(config, "parallel_balance", "lpt"),
        parallel_timing_file=str(
            getattr(config, "parallel_timing_file", None) or ".behave-pool-timing.json"
        ),
        parallel_report=str(getattr(config, "parallel_report", None) or "behave-pool-report.json"),
        dry_run=config.dry_run,
        use_nested_step_modules=getattr(config, "use_nested_step_modules", False),
    )

WorkUnit

A frozen dataclass representing a single unit of test work dispatched to a worker process.

from behave_pool.work_unit import WorkUnit
from behave_pool.config import ConfigSnapshot

unit = WorkUnit(
    id="feature:features/login.feature",
    config=ConfigSnapshot(base_dir="features", steps_dir="steps"),
    feature_path="features/login.feature",
    tags=["serial"],
)

print(unit.is_serial)  # True

behave_pool.work_unit

WorkUnit: a single unit of test work to be dispatched to a worker.

WorkUnit dataclass

A single unit of test work for parallel dispatch.

A WorkUnit represents either a whole feature file or a single scenario (identified by line number) within a feature file. It carries a picklable ConfigSnapshot so that workers can execute independently after being spawned via multiprocessing.

Attributes:

Name	Type	Description
id	str	Unique identifier, e.g. "feature:login.feature" or "scenario:login.feature:12".
config	ConfigSnapshot	Picklable ConfigSnapshot with essential configuration.
feature_path	str | None	Path to the .feature file.
scenario_line	int | None	Line number of the scenario within the feature file. None when the work unit represents an entire feature.
tags	list[str]	Tags associated with the scenario or feature.

Source code in behave_pool/work_unit.py

@dataclass(frozen=True)
class WorkUnit:
    """A single unit of test work for parallel dispatch.

    A WorkUnit represents either a whole feature file or a single scenario
    (identified by line number) within a feature file. It carries a
    picklable ConfigSnapshot so that workers can execute independently
    after being spawned via multiprocessing.

    Attributes:
        id: Unique identifier, e.g. "feature:login.feature" or
            "scenario:login.feature:12".
        config: Picklable ConfigSnapshot with essential configuration.
        feature_path: Path to the .feature file.
        scenario_line: Line number of the scenario within the feature file.
            None when the work unit represents an entire feature.
        tags: Tags associated with the scenario or feature.
    """

    id: str
    config: ConfigSnapshot
    feature_path: str | None = None
    scenario_line: int | None = None
    tags: list[str] = field(default_factory=list)

    @property
    def is_serial(self) -> bool:
        """True if this work unit is tagged with @serial.

        Serial work units are executed sequentially after all parallel
        work units have completed.
        """
        return "serial" in self.tags

is_serial property

is_serial: bool

True if this work unit is tagged with @serial.

Serial work units are executed sequentially after all parallel work units have completed.

WorkerResult

The outcome of executing a WorkUnit in a worker process.

from behave_pool.result import WorkerResult

result = WorkerResult(
    worker_id=0,
    work_unit_id="feature:features/login.feature",
    failed=False,
    duration=1.23,
)
print(result.failed)  # False
print(result.duration)  # 1.23

behave_pool.result

WorkerResult: the outcome of executing a WorkUnit in a worker process.

WorkerResult dataclass

Result of a worker executing a single WorkUnit.

Produced by WorkerRunner.run_work_unit() and sent back to the coordinator via the result queue for aggregation.

Attributes:

Name	Type	Description
worker_id	int	Identifier of the worker process that produced this result.
work_unit_id	str	ID of the WorkUnit that was executed.
failed	bool	True if any scenario or step in the work unit failed.
duration	float	Wall-clock execution time in seconds.
report_path	str | None	Path to the temporary JSON report file, or None if no report was written.
undefined_steps	list[str]	List of undefined step text patterns encountered.
error	str | None	Error message if the worker process crashed, None otherwise.

Source code in behave_pool/result.py

@dataclass(frozen=True)
class WorkerResult:
    """Result of a worker executing a single WorkUnit.

    Produced by WorkerRunner.run_work_unit() and sent back to the
    coordinator via the result queue for aggregation.

    Attributes:
        worker_id: Identifier of the worker process that produced this result.
        work_unit_id: ID of the WorkUnit that was executed.
        failed: True if any scenario or step in the work unit failed.
        duration: Wall-clock execution time in seconds.
        report_path: Path to the temporary JSON report file, or None if
            no report was written.
        undefined_steps: List of undefined step text patterns encountered.
        error: Error message if the worker process crashed, None otherwise.
    """

    worker_id: int
    work_unit_id: str
    failed: bool
    duration: float
    report_path: str | None = None
    undefined_steps: list[str] = field(default_factory=list)
    error: str | None = None

TimingStore

Loads and saves historical work unit durations as JSON for LPT balancing.

from pathlib import Path
from behave_pool.timing import TimingStore

store = TimingStore(path=Path(".behave-pool-timing.json"))
store.load()
print(store.get_duration("feature:features/login.feature"))  # 1.23

store.update("feature:features/login.feature", 1.45)
store.save_if_changed()  # True if file was written

behave_pool.timing

TimingStore: persist historical work unit durations for LPT balancing.

TimingStore

Load and save historical work unit durations as JSON.

The file format is a simple mapping of work unit IDs to durations in seconds::

{"feature:login.feature": 1.23, "feature:checkout.feature": 0.45}

Attributes:

Name	Type	Description
path		Path to the JSON timing file.

Source code in behave_pool/timing.py

class TimingStore:
    """Load and save historical work unit durations as JSON.

    The file format is a simple mapping of work unit IDs to durations
    in seconds::

        {"feature:login.feature": 1.23, "feature:checkout.feature": 0.45}

    Attributes:
        path: Path to the JSON timing file.
    """

    def __init__(self, path: Path = Path(".behave-pool-timing.json")) -> None:
        self.path = path
        self._data: dict[str, float] = {}
        self._loaded: bool = False

    def load(self) -> dict[str, float]:
        """Load timing data from the JSON file.

        Returns:
            Dict mapping work unit IDs to durations in seconds.
            Returns an empty dict if the file is missing or corrupt.
        """
        if not self.path.exists():
            self._data = {}
            self._loaded = True
            return self._data

        try:
            text = self.path.read_text(encoding="utf-8")
            raw = json.loads(text)
            if not isinstance(raw, dict):
                logger.warning("Timing file %s is not a JSON object; ignoring.", self.path)
                self._data = {}
            else:
                self._data = {}
                for k, v in raw.items():
                    try:
                        coerced = float(v)
                    except (TypeError, ValueError):
                        logger.warning(
                            "Timing file %s: skipping invalid entry %r=%r", self.path, k, v
                        )
                        continue
                    if not math.isfinite(coerced):
                        logger.warning(
                            "Timing file %s: skipping non-finite entry %r=%r", self.path, k, v
                        )
                        continue
                    self._data[str(k)] = coerced
        except (json.JSONDecodeError, ValueError, TypeError, OSError) as exc:
            logger.warning("Timing file %s is corrupt (%s); ignoring.", self.path, exc)
            self._data = {}

        self._loaded = True
        return self._data

    def save(self, data: dict[str, float]) -> None:
        """Write timing data to the JSON file atomically with indent=2.

        Writes to a temporary file in the same directory and then
        atomically replaces the target file, preventing corruption
        if the process crashes during a write.

        Args:
            data: Dict mapping work unit IDs to durations in seconds.
        """
        text = json.dumps(data, indent=2, sort_keys=True)
        parent = self.path.parent
        parent.mkdir(parents=True, exist_ok=True)
        fd, tmp_path = tempfile.mkstemp(dir=str(parent), suffix=".tmp", prefix=self.path.name + "_")
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                f.write(text)
            os.replace(tmp_path, self.path)
        except OSError:
            if os.path.exists(tmp_path):
                os.unlink(tmp_path)
            raise
        self._data = dict(data)

    def get_duration(self, work_unit_id: str) -> float:
        """Return the stored duration for a work unit, or 0.0 if unknown."""
        if not self._loaded:
            self.load()
        return self._data.get(work_unit_id, 0.0)

    def update(self, work_unit_id: str, duration: float) -> None:
        """Insert or update the duration for a work unit.

        Non-finite values (inf, NaN) are rejected because they produce
        non-standard JSON and break save_if_changed equality checks
        (NaN != NaN).
        """
        if not self._loaded:
            self.load()
        if not math.isfinite(duration):
            logger.warning(
                "Ignoring non-finite duration %r for work unit %r", duration, work_unit_id
            )
            return
        self._data[work_unit_id] = duration

    def save_if_changed(self) -> bool:
        """Save data only if it differs from what was loaded.

        Returns:
            True if the file was written, False if no changes were detected.
        """
        if not self._loaded:
            self.load()

        # Read the file contents without overwriting self._data.
        original: dict[str, float] = {}
        if self.path.exists():
            try:
                text = self.path.read_text(encoding="utf-8")
                raw = json.loads(text)
                if isinstance(raw, dict):
                    for k, v in raw.items():
                        with contextlib.suppress(TypeError, ValueError):
                            coerced = float(v)
                            if math.isfinite(coerced):
                                original[str(k)] = coerced
            except (json.JSONDecodeError, ValueError, TypeError, OSError):
                pass

        if self._data == original:
            return False

        self.save(self._data)
        return True

get_duration

get_duration(work_unit_id: str) -> float

Return the stored duration for a work unit, or 0.0 if unknown.

Source code in behave_pool/timing.py

def get_duration(self, work_unit_id: str) -> float:
    """Return the stored duration for a work unit, or 0.0 if unknown."""
    if not self._loaded:
        self.load()
    return self._data.get(work_unit_id, 0.0)

load

load() -> dict[str, float]

Load timing data from the JSON file.

Returns:

Type	Description
dict[str, float]	Dict mapping work unit IDs to durations in seconds.
dict[str, float]	Returns an empty dict if the file is missing or corrupt.

Source code in behave_pool/timing.py

def load(self) -> dict[str, float]:
    """Load timing data from the JSON file.

    Returns:
        Dict mapping work unit IDs to durations in seconds.
        Returns an empty dict if the file is missing or corrupt.
    """
    if not self.path.exists():
        self._data = {}
        self._loaded = True
        return self._data

    try:
        text = self.path.read_text(encoding="utf-8")
        raw = json.loads(text)
        if not isinstance(raw, dict):
            logger.warning("Timing file %s is not a JSON object; ignoring.", self.path)
            self._data = {}
        else:
            self._data = {}
            for k, v in raw.items():
                try:
                    coerced = float(v)
                except (TypeError, ValueError):
                    logger.warning(
                        "Timing file %s: skipping invalid entry %r=%r", self.path, k, v
                    )
                    continue
                if not math.isfinite(coerced):
                    logger.warning(
                        "Timing file %s: skipping non-finite entry %r=%r", self.path, k, v
                    )
                    continue
                self._data[str(k)] = coerced
    except (json.JSONDecodeError, ValueError, TypeError, OSError) as exc:
        logger.warning("Timing file %s is corrupt (%s); ignoring.", self.path, exc)
        self._data = {}

    self._loaded = True
    return self._data

save

save(data: dict[str, float]) -> None

Write timing data to the JSON file atomically with indent=2.

Writes to a temporary file in the same directory and then atomically replaces the target file, preventing corruption if the process crashes during a write.

Parameters:

Name	Type	Description	Default
data	dict[str, float]	Dict mapping work unit IDs to durations in seconds.	required

Source code in behave_pool/timing.py

def save(self, data: dict[str, float]) -> None:
    """Write timing data to the JSON file atomically with indent=2.

    Writes to a temporary file in the same directory and then
    atomically replaces the target file, preventing corruption
    if the process crashes during a write.

    Args:
        data: Dict mapping work unit IDs to durations in seconds.
    """
    text = json.dumps(data, indent=2, sort_keys=True)
    parent = self.path.parent
    parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_path = tempfile.mkstemp(dir=str(parent), suffix=".tmp", prefix=self.path.name + "_")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(text)
        os.replace(tmp_path, self.path)
    except OSError:
        if os.path.exists(tmp_path):
            os.unlink(tmp_path)
        raise
    self._data = dict(data)

save_if_changed

save_if_changed() -> bool

Save data only if it differs from what was loaded.

Returns:

Type	Description
bool	True if the file was written, False if no changes were detected.

Source code in behave_pool/timing.py

def save_if_changed(self) -> bool:
    """Save data only if it differs from what was loaded.

    Returns:
        True if the file was written, False if no changes were detected.
    """
    if not self._loaded:
        self.load()

    # Read the file contents without overwriting self._data.
    original: dict[str, float] = {}
    if self.path.exists():
        try:
            text = self.path.read_text(encoding="utf-8")
            raw = json.loads(text)
            if isinstance(raw, dict):
                for k, v in raw.items():
                    with contextlib.suppress(TypeError, ValueError):
                        coerced = float(v)
                        if math.isfinite(coerced):
                            original[str(k)] = coerced
        except (json.JSONDecodeError, ValueError, TypeError, OSError):
            pass

    if self._data == original:
        return False

    self.save(self._data)
    return True

update

update(work_unit_id: str, duration: float) -> None

Insert or update the duration for a work unit.

Non-finite values (inf, NaN) are rejected because they produce non-standard JSON and break save_if_changed equality checks (NaN != NaN).

Source code in behave_pool/timing.py

def update(self, work_unit_id: str, duration: float) -> None:
    """Insert or update the duration for a work unit.

    Non-finite values (inf, NaN) are rejected because they produce
    non-standard JSON and break save_if_changed equality checks
    (NaN != NaN).
    """
    if not self._loaded:
        self.load()
    if not math.isfinite(duration):
        logger.warning(
            "Ignoring non-finite duration %r for work unit %r", duration, work_unit_id
        )
        return
    self._data[work_unit_id] = duration

WorkUnitIterator

Strategy pattern for generating WorkUnit objects from parsed features.

from behave_pool.iterator import WorkUnitIterator

iterator = WorkUnitIterator.for_scheme(
    scheme="feature",
    features=features,
    config=config,
)
for unit in iterator.iterate():
    print(unit.id, unit.feature_path)

behave_pool.iterator

Strategy pattern for iterating work units from Behave features.

FeatureIterator

Bases: WorkUnitIterator

Generate one WorkUnit per feature file.

Each WorkUnit contains an isolated deep copy of the Configuration so that workers can execute independently without shared mutable state.

Source code in behave_pool/iterator.py

class FeatureIterator(WorkUnitIterator):
    """Generate one WorkUnit per feature file.

    Each WorkUnit contains an isolated deep copy of the Configuration
    so that workers can execute independently without shared mutable state.
    """

    def __init__(self, features: list[Feature], config: Configuration) -> None:
        self._features = features
        self._config = config

    def iterate(self) -> Iterator[WorkUnit]:
        """Yield one WorkUnit per feature.

        Tags are collected from both the feature and its scenarios.
        If any scenario has the ``serial`` tag, the work unit is
        marked serial so it runs in the serial phase.
        """
        for feature in self._features:
            tags = set(str(t) for t in (getattr(feature, "tags", None) or []))
            for scenario in getattr(feature, "scenarios", None) or []:
                tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))
            yield WorkUnit(
                id=f"feature:{feature.filename}",
                config=snapshot_config(self._config),
                feature_path=feature.filename,
                scenario_line=None,
                tags=list(tags),
            )

iterate

iterate() -> Iterator[WorkUnit]

Yield one WorkUnit per feature.

Tags are collected from both the feature and its scenarios. If any scenario has the serial tag, the work unit is marked serial so it runs in the serial phase.

Source code in behave_pool/iterator.py

def iterate(self) -> Iterator[WorkUnit]:
    """Yield one WorkUnit per feature.

    Tags are collected from both the feature and its scenarios.
    If any scenario has the ``serial`` tag, the work unit is
    marked serial so it runs in the serial phase.
    """
    for feature in self._features:
        tags = set(str(t) for t in (getattr(feature, "tags", None) or []))
        for scenario in getattr(feature, "scenarios", None) or []:
            tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))
        yield WorkUnit(
            id=f"feature:{feature.filename}",
            config=snapshot_config(self._config),
            feature_path=feature.filename,
            scenario_line=None,
            tags=list(tags),
        )

WorkUnitIterator

Bases: ABC

Abstract strategy for generating WorkUnits from parsed features.

Source code in behave_pool/iterator.py

class WorkUnitIterator(ABC):
    """Abstract strategy for generating WorkUnits from parsed features."""

    @abstractmethod
    def iterate(self) -> Iterator[WorkUnit]:
        """Yield WorkUnit instances one at a time."""
        ...

    @staticmethod
    def for_scheme(
        scheme: str,
        features: list[Feature],
        config: Configuration,
    ) -> WorkUnitIterator:
        """Factory: return the iterator for the given parallel scheme.

        Args:
            scheme: "feature" or "scenario".
            features: Parsed Behave Feature objects.
            config: Coordinator's Configuration (will be deep-copied per unit).

        Returns:
            A WorkUnitIterator instance for the requested scheme.

        Raises:
            ValueError: If scheme is not recognised.
            NotImplementedError: If scheme is "scenario" (not yet implemented).
        """
        if scheme == "feature":
            return FeatureIterator(features, config)
        if scheme == "scenario":
            raise NotImplementedError("ScenarioIterator is not yet implemented")
        msg = f"Unknown parallel scheme: {scheme!r}. Use 'feature' or 'scenario'."
        raise ValueError(msg)

for_scheme staticmethod

for_scheme(scheme: str, features: list[Feature], config: Configuration) -> WorkUnitIterator

Factory: return the iterator for the given parallel scheme.

Parameters:

Name	Type	Description	Default
scheme	str	"feature" or "scenario".	required
features	list[Feature]	Parsed Behave Feature objects.	required
config	Configuration	Coordinator's Configuration (will be deep-copied per unit).	required

Returns:

Type	Description
WorkUnitIterator	A WorkUnitIterator instance for the requested scheme.

Raises:

Type	Description
ValueError	If scheme is not recognised.
NotImplementedError	If scheme is "scenario" (not yet implemented).

Source code in behave_pool/iterator.py

@staticmethod
def for_scheme(
    scheme: str,
    features: list[Feature],
    config: Configuration,
) -> WorkUnitIterator:
    """Factory: return the iterator for the given parallel scheme.

    Args:
        scheme: "feature" or "scenario".
        features: Parsed Behave Feature objects.
        config: Coordinator's Configuration (will be deep-copied per unit).

    Returns:
        A WorkUnitIterator instance for the requested scheme.

    Raises:
        ValueError: If scheme is not recognised.
        NotImplementedError: If scheme is "scenario" (not yet implemented).
    """
    if scheme == "feature":
        return FeatureIterator(features, config)
    if scheme == "scenario":
        raise NotImplementedError("ScenarioIterator is not yet implemented")
    msg = f"Unknown parallel scheme: {scheme!r}. Use 'feature' or 'scenario'."
    raise ValueError(msg)

iterate abstractmethod

iterate() -> Iterator[WorkUnit]

Yield WorkUnit instances one at a time.

Source code in behave_pool/iterator.py

@abstractmethod
def iterate(self) -> Iterator[WorkUnit]:
    """Yield WorkUnit instances one at a time."""
    ...

Worker

WorkerRunner executes work units inside an isolated worker process. WorkerProcess wraps multiprocessing.Process for lifecycle management.

These classes are used internally by ParallelRunner and are not typically instantiated directly by end users.

behave_pool.worker

WorkerRunner and WorkerProcess for parallel test execution.

WorkerProcess

Wrapper around multiprocessing.Process for consuming WorkUnits.

Each WorkerProcess runs _worker_run_loop in a separate process, consuming WorkUnits from a JoinableQueue and producing WorkerResults in a result Queue.

Source code in behave_pool/worker.py

class WorkerProcess:
    """Wrapper around multiprocessing.Process for consuming WorkUnits.

    Each WorkerProcess runs _worker_run_loop in a separate process,
    consuming WorkUnits from a JoinableQueue and producing WorkerResults
    in a result Queue.
    """

    def __init__(
        self,
        worker_id: int,
        task_queue: Any,
        result_queue: QueueType[WorkerResult],
        stop_event: Any,
        config_snapshot: ConfigSnapshot,
        ctx: Any | None = None,
    ) -> None:
        self.worker_id = worker_id
        self._task_queue = task_queue
        self._result_queue = result_queue
        self._stop_event = stop_event
        self._config_snapshot = config_snapshot
        self._ctx = ctx or multiprocessing.get_context("spawn")
        self._process: multiprocessing.process.BaseProcess | None = None

    def start(self) -> None:
        """Launch the worker process."""
        self._process = self._ctx.Process(
            target=_worker_run_loop,
            args=(
                self.worker_id,
                self._task_queue,
                self._result_queue,
                self._stop_event,
                self._config_snapshot,
            ),
            daemon=True,
        )
        assert self._process is not None
        self._process.start()

    def join(self, timeout: float | None = None) -> None:
        """Wait for the worker process to terminate."""
        if self._process is not None:
            self._process.join(timeout)

    def is_alive(self) -> bool:
        """Return True if the worker process is still running."""
        if self._process is None:
            return False
        return self._process.is_alive()

    def terminate(self) -> None:
        """Forcefully terminate the worker process.

        Should only be called after ``join(timeout=...)`` returns and
        ``is_alive()`` is still True, as a last resort to avoid
        indefinite hangs from stuck workers.
        """
        if self._process is not None and self._process.is_alive():
            self._process.terminate()
            logger.warning("Worker %d forcibly terminated.", self.worker_id)

is_alive

is_alive() -> bool

Return True if the worker process is still running.

Source code in behave_pool/worker.py

def is_alive(self) -> bool:
    """Return True if the worker process is still running."""
    if self._process is None:
        return False
    return self._process.is_alive()

join

join(timeout: float | None = None) -> None

Wait for the worker process to terminate.

Source code in behave_pool/worker.py

def join(self, timeout: float | None = None) -> None:
    """Wait for the worker process to terminate."""
    if self._process is not None:
        self._process.join(timeout)

start

start() -> None

Launch the worker process.

Source code in behave_pool/worker.py

def start(self) -> None:
    """Launch the worker process."""
    self._process = self._ctx.Process(
        target=_worker_run_loop,
        args=(
            self.worker_id,
            self._task_queue,
            self._result_queue,
            self._stop_event,
            self._config_snapshot,
        ),
        daemon=True,
    )
    assert self._process is not None
    self._process.start()

terminate

terminate() -> None

Forcefully terminate the worker process.

Should only be called after join(timeout=...) returns and is_alive() is still True, as a last resort to avoid indefinite hangs from stuck workers.

Source code in behave_pool/worker.py

def terminate(self) -> None:
    """Forcefully terminate the worker process.

    Should only be called after ``join(timeout=...)`` returns and
    ``is_alive()`` is still True, as a last resort to avoid
    indefinite hangs from stuck workers.
    """
    if self._process is not None and self._process.is_alive():
        self._process.terminate()
        logger.warning("Worker %d forcibly terminated.", self.worker_id)

WorkerRunner

Bases: ModelRunner

Runner that executes work units in an isolated worker process.

Lifecycle

1. setup() — once at worker start (load hooks, steps, before_all).
2. run_work_unit(unit) — called per work unit.
3. teardown() — once at worker end (after_all, close formatters).

Source code in behave_pool/worker.py

class WorkerRunner(ModelRunner):  # type: ignore[misc]
    """Runner that executes work units in an isolated worker process.

    Lifecycle:
        1. setup() — once at worker start (load hooks, steps, before_all).
        2. run_work_unit(unit) — called per work unit.
        3. teardown() — once at worker end (after_all, close formatters).
    """

    def __init__(
        self,
        config: Configuration,
        worker_id: int,
        result_queue: QueueType[WorkerResult],
        stop_event: Any,
    ) -> None:
        super().__init__(config)
        self.worker_id = worker_id
        self.result_queue = result_queue
        self.stop_event = stop_event
        self._last_result: WorkerResult | None = None
        self._setup_done = False
        self.base_dir = config.base_dir if getattr(config, "base_dir", None) else "features"

    def load_hooks(self, filename: str | None = None) -> None:
        """Load environment hooks from the environment file."""
        env_filename = (
            filename or getattr(self.config, "environment_file", None) or "environment.py"
        )
        hooks_path = os.path.join(self.base_dir, env_filename)
        if os.path.exists(hooks_path):
            exec_file(hooks_path, self.hooks)

        if "before_all" not in self.hooks:
            self.hooks["before_all"] = _noop_hook

    def load_step_definitions(self, extra_step_paths: list[str] | None = None) -> None:
        """Load step definitions from the steps directory."""
        if extra_step_paths is None:
            extra_step_paths = []
        steps_dir = os.path.join(self.base_dir, getattr(self.config, "steps_dir", None) or "steps")
        step_paths = [steps_dir]
        if self.config.use_nested_step_modules:
            step_subdirectories = select_subdirectories(steps_dir)
            step_paths.extend(step_subdirectories)
        step_paths = list(step_paths) + list(extra_step_paths)
        load_step_modules(step_paths)
        from behave.step_registry import registry as global_registry

        self.step_registry = global_registry

    def setup(self) -> None:
        """Load hooks, step definitions, create Context, run before_all."""
        self.load_hooks()
        self.load_step_definitions()
        self.context = Context(self)
        self.run_hook("before_all")
        self._setup_done = True
        logger.debug("WorkerRunner %d setup complete", self.worker_id)

    def run_work_unit(self, unit: WorkUnit) -> WorkerResult:
        """Execute a single work unit and return the result.

        Args:
            unit: The WorkUnit to execute.

        Returns:
            WorkerResult with timing, failure status, and report path.
        """
        start = time.perf_counter()
        try:
            if not unit.feature_path:
                raise ValueError(f"Work unit {unit.id} has no feature_path")
            self.features = parse_features(
                [unit.feature_path],
                language=self.config.lang,
            )
            self.undefined_steps.clear()
            self.hook_failures = 0
            if self.context is not None:
                self.aborted = False
            failed = self._run_features()
            duration = time.perf_counter() - start
            undefined = list(self.undefined_steps)
            report_path = self._write_report(unit)
            result = WorkerResult(
                worker_id=self.worker_id,
                work_unit_id=unit.id,
                failed=failed,
                duration=duration,
                report_path=report_path,
                undefined_steps=undefined,
            )
        except Exception as exc:
            duration = time.perf_counter() - start
            logger.exception("WorkerRunner %d error in work unit %s", self.worker_id, unit.id)
            result = WorkerResult(
                worker_id=self.worker_id,
                work_unit_id=unit.id,
                failed=True,
                duration=duration,
                error=str(exc),
            )
        self._last_result = result
        return result

    def teardown(self) -> None:
        """Run after_all hooks and close formatters.

        Safe to call even if setup() did not complete: skips hooks
        and formatters that were never initialised.
        """
        if self._setup_done:
            self.run_hook("after_all")
        for formatter in getattr(self, "formatters", []):
            formatter.close()
        logger.debug("WorkerRunner %d teardown complete", self.worker_id)

    def collect_result(self) -> WorkerResult | None:
        """Return the last WorkerResult produced, or None."""
        return self._last_result

    def _run_features(self) -> bool:
        """Run self.features without before_all/after_all hooks.

        Returns:
            True if any feature failed.
        """
        run_feature = not self.aborted
        failed_count = 0
        undefined_steps_initial_size = len(self.undefined_steps)
        for feature in self.features:
            if run_feature:
                try:
                    self.feature = feature
                    for formatter in self.formatters:
                        formatter.uri(feature.filename)
                    failed = feature.run(self)
                    if failed:
                        failed_count += 1
                        if self.config.stop or self.aborted:
                            run_feature = False
                except KeyboardInterrupt:
                    self.abort(reason="KeyboardInterrupt")
                    failed_count += 1
                    run_feature = False
            for reporter in self.config.reporters:
                reporter.feature(feature)
        return (
            failed_count > 0
            or self.aborted
            or self.hook_failures > 0
            or len(self.undefined_steps) > undefined_steps_initial_size
        )

    def _serialize_location(self, obj: Any) -> dict[str, Any] | None:
        """Extract a location dict from a Behave model object."""
        filename = getattr(obj, "filename", None)
        line = getattr(obj, "line", None)
        if filename is None and line is None:
            loc = getattr(obj, "location", None)
            if loc is not None:
                filename = getattr(loc, "filename", None) or str(loc)
                line = getattr(loc, "line", None)
        if filename is None and line is None:
            return None
        result: dict[str, Any] = {
            "filename": str(filename or ""),
            "line": int(line or 0),
        }
        return result

    def _map_status(self, raw: Any) -> str:
        """Map a Behave status to a canonical string."""
        if raw is None:
            return "untested"
        name = getattr(raw, "name", None) or str(raw)
        return name.lower().strip() or "passed"

    def _serialize_error(self, step: Any) -> dict[str, Any] | None:
        """Extract an error dict from a Behave step."""
        exc = getattr(step, "error", None) or getattr(step, "exception", None)
        if exc is not None and isinstance(exc, BaseException):
            import traceback as _tb

            return {
                "id": f"err-{id(step):x}",
                "type": type(exc).__name__,
                "message": str(exc),
                "traceback": "".join(_tb.format_exception(type(exc), exc, exc.__traceback__)),
                "location": self._serialize_location(step),
            }
        error_message = getattr(step, "error_message", None)
        if error_message:
            return {
                "id": f"err-{id(step):x}",
                "type": "Error",
                "message": str(error_message),
                "traceback": str(getattr(step, "exc_traceback", None) or error_message),
                "location": self._serialize_location(step),
            }
        return None

    def _serialize_step(self, step: Any) -> dict[str, Any]:
        """Convert a Behave Step to a behave-modern-json-report step dict."""
        return {
            "id": f"step-{id(step):x}",
            "keyword": str(getattr(step, "keyword", "")),
            "text": str(getattr(step, "name", "") or getattr(step, "text", "")),
            "status": self._map_status(getattr(step, "status", "passed")),
            "duration": float(getattr(step, "duration", 0.0) or 0.0),
            "location": self._serialize_location(step),
            "error": self._serialize_error(step),
            "attachments": [],
            "logs": [],
        }

    def _serialize_background(self, background: Any) -> dict[str, Any]:
        """Convert a Behave Background to a behave-modern-json-report background dict."""
        steps = [self._serialize_step(s) for s in (getattr(background, "steps", None) or [])]
        return {
            "id": f"bg-{id(background):x}",
            "name": str(getattr(background, "name", "") or ""),
            "keyword": str(getattr(background, "keyword", "Background") or "Background"),
            "location": self._serialize_location(background),
            "steps": steps,
        }

    def _serialize_scenario(self, scenario: Any, feature_id: str) -> dict[str, Any]:
        """Convert a Behave Scenario to a behave-modern-json-report scenario dict."""
        steps = []
        for step in getattr(scenario, "all_steps", None) or getattr(scenario, "steps", []) or []:
            steps.append(self._serialize_step(step))
        scenario_type = str(getattr(scenario, "type", "") or "")
        is_outline = scenario_type in ("scenario_outline", "outline")
        return {
            "id": f"scenario-{id(scenario):x}",
            "name": str(getattr(scenario, "name", "") or "<unnamed>"),
            "featureId": feature_id,
            "description": str(getattr(scenario, "description", "") or "") or None,
            "tags": [str(t) for t in (getattr(scenario, "tags", None) or [])],
            "examples": [],
            "location": self._serialize_location(scenario),
            "status": self._map_status(getattr(scenario, "status", "passed")),
            "duration": float(getattr(scenario, "duration", 0.0) or 0.0),
            "steps": steps,
            "background": None,
            "rule": None,
            "isOutline": is_outline,
            "outlineName": None,
            "retry": None,
        }

    def _serialize_feature(self, feature: Any) -> dict[str, Any]:
        """Convert a Behave Feature to a behave-modern-json-report feature dict.

        The output matches the feature structure of behave-modern-json-report's
        ExecutionReport schema so downstream tools can consume it directly.
        """
        feature_id = f"feature-{id(feature):x}"
        scenarios = []
        for scenario in getattr(feature, "scenarios", None) or []:
            scenarios.append(self._serialize_scenario(scenario, feature_id))
        background = None
        behave_background = getattr(feature, "background", None)
        if behave_background:
            background = self._serialize_background(behave_background)
        return {
            "id": feature_id,
            "name": str(getattr(feature, "name", "") or "<unnamed>"),
            "description": str(getattr(feature, "description", "") or "") or None,
            "tags": [str(t) for t in (getattr(feature, "tags", None) or [])],
            "filename": getattr(feature, "filename", None),
            "line": getattr(feature, "line", None),
            "status": self._map_status(getattr(feature, "status", "passed")),
            "duration": float(getattr(feature, "duration", 0.0) or 0.0),
            "scenarios": scenarios,
            "background": background,
        }

    def _write_report(self, unit: WorkUnit) -> str | None:
        """Write a JSON report for the work unit in behave-modern-json-report format.

        Each report contains a list of feature dicts matching the
        behave-modern-json-report ExecutionReport feature schema. The
        coordinator merges these into a full ExecutionReport.

        Returns:
            Path to the report file, or None if writing failed.
        """
        tmp_dir = Path("tmp")
        safe_id = unit.id.replace(":", "_").replace("/", "_").replace("\\", "_")
        report_path = tmp_dir / f"worker_{self.worker_id}_{safe_id}.json"
        try:
            tmp_dir.mkdir(exist_ok=True)
            report_data = {
                "worker_id": self.worker_id,
                "work_unit_id": unit.id,
                "features": [self._serialize_feature(f) for f in self.features],
                "failed": any(getattr(f, "status", "passed") == "failed" for f in self.features),
            }
            report_path.write_text(json.dumps(report_data, indent=2), encoding="utf-8")
            return str(report_path)
        except Exception:
            logger.warning("Failed to write report to %s", report_path)
            return None

collect_result

collect_result() -> WorkerResult | None

Return the last WorkerResult produced, or None.

Source code in behave_pool/worker.py

def collect_result(self) -> WorkerResult | None:
    """Return the last WorkerResult produced, or None."""
    return self._last_result

load_hooks

load_hooks(filename: str | None = None) -> None

Load environment hooks from the environment file.

Source code in behave_pool/worker.py

def load_hooks(self, filename: str | None = None) -> None:
    """Load environment hooks from the environment file."""
    env_filename = (
        filename or getattr(self.config, "environment_file", None) or "environment.py"
    )
    hooks_path = os.path.join(self.base_dir, env_filename)
    if os.path.exists(hooks_path):
        exec_file(hooks_path, self.hooks)

    if "before_all" not in self.hooks:
        self.hooks["before_all"] = _noop_hook

load_step_definitions

load_step_definitions(extra_step_paths: list[str] | None = None) -> None

Load step definitions from the steps directory.

Source code in behave_pool/worker.py

def load_step_definitions(self, extra_step_paths: list[str] | None = None) -> None:
    """Load step definitions from the steps directory."""
    if extra_step_paths is None:
        extra_step_paths = []
    steps_dir = os.path.join(self.base_dir, getattr(self.config, "steps_dir", None) or "steps")
    step_paths = [steps_dir]
    if self.config.use_nested_step_modules:
        step_subdirectories = select_subdirectories(steps_dir)
        step_paths.extend(step_subdirectories)
    step_paths = list(step_paths) + list(extra_step_paths)
    load_step_modules(step_paths)
    from behave.step_registry import registry as global_registry

    self.step_registry = global_registry

run_work_unit

run_work_unit(unit: WorkUnit) -> WorkerResult

Execute a single work unit and return the result.

Parameters:

Name	Type	Description	Default
unit	WorkUnit	The WorkUnit to execute.	required

Returns:

Type	Description
WorkerResult	WorkerResult with timing, failure status, and report path.

Source code in behave_pool/worker.py

def run_work_unit(self, unit: WorkUnit) -> WorkerResult:
    """Execute a single work unit and return the result.

    Args:
        unit: The WorkUnit to execute.

    Returns:
        WorkerResult with timing, failure status, and report path.
    """
    start = time.perf_counter()
    try:
        if not unit.feature_path:
            raise ValueError(f"Work unit {unit.id} has no feature_path")
        self.features = parse_features(
            [unit.feature_path],
            language=self.config.lang,
        )
        self.undefined_steps.clear()
        self.hook_failures = 0
        if self.context is not None:
            self.aborted = False
        failed = self._run_features()
        duration = time.perf_counter() - start
        undefined = list(self.undefined_steps)
        report_path = self._write_report(unit)
        result = WorkerResult(
            worker_id=self.worker_id,
            work_unit_id=unit.id,
            failed=failed,
            duration=duration,
            report_path=report_path,
            undefined_steps=undefined,
        )
    except Exception as exc:
        duration = time.perf_counter() - start
        logger.exception("WorkerRunner %d error in work unit %s", self.worker_id, unit.id)
        result = WorkerResult(
            worker_id=self.worker_id,
            work_unit_id=unit.id,
            failed=True,
            duration=duration,
            error=str(exc),
        )
    self._last_result = result
    return result

setup

setup() -> None

Load hooks, step definitions, create Context, run before_all.

Source code in behave_pool/worker.py

def setup(self) -> None:
    """Load hooks, step definitions, create Context, run before_all."""
    self.load_hooks()
    self.load_step_definitions()
    self.context = Context(self)
    self.run_hook("before_all")
    self._setup_done = True
    logger.debug("WorkerRunner %d setup complete", self.worker_id)

teardown

teardown() -> None

Run after_all hooks and close formatters.

Safe to call even if setup() did not complete: skips hooks and formatters that were never initialised.

Source code in behave_pool/worker.py

def teardown(self) -> None:
    """Run after_all hooks and close formatters.

    Safe to call even if setup() did not complete: skips hooks
    and formatters that were never initialised.
    """
    if self._setup_done:
        self.run_hook("after_all")
    for formatter in getattr(self, "formatters", []):
        formatter.close()
    logger.debug("WorkerRunner %d teardown complete", self.worker_id)