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)
        self._early_results: list[WorkerResult] = []
        self._exec_started_at: Any = None
        self._exec_ended_at: Any = None
        self._exec_wall_duration: float | None = None

    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)

        if self._is_sharding_active():
            features = self._filter_features_by_shard(features)
            self._log_shard_info(len(features))

        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 -> shard -> split -> dispatch -> collect."""
        from datetime import UTC, datetime

        ctx = multiprocessing.get_context("spawn")
        task_queue: Any = ctx.JoinableQueue()
        result_queue: Any = ctx.Queue()
        stop_event: Any = ctx.Event()

        self._exec_started_at = datetime.now(UTC)
        started = time.monotonic()
        self._early_results = []
        try:
            work_units = self._plan()
            if self._is_sharding_active():
                work_units = self._apply_shard(work_units)
            parallel_batch, serial_batch = self._split_by_serial_tag(work_units)
            dispatched = self._dispatch(
                task_queue, result_queue, stop_event, parallel_batch, serial_batch, ctx
            )
            return self._collect(result_queue, dispatched)
        finally:
            self._exec_ended_at = datetime.now(UTC)
            self._exec_wall_duration = time.monotonic() - started
            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 _is_sharding_active(self) -> bool:
        """Check whether sharding is enabled for this run."""
        shard_index = getattr(self.config, "shard_index", None)
        total_shards = getattr(self.config, "total_shards", None)
        return shard_index is not None and total_shards is not None

    def _apply_shard(self, work_units: list[WorkUnit]) -> list[WorkUnit]:
        """Filter work units to only those in the current shard.

        Shard membership is computed on ids sorted deterministically, but
        the returned list preserves the planned (LPT/FIFO) order.

        Args:
            work_units: All planned work units.

        Returns:
            Work units belonging to the current shard.
        """
        shard_index = int(getattr(self.config, "shard_index", 0))
        total_shards = int(getattr(self.config, "total_shards", 0))
        selected = select_shard_work_units(work_units, shard_index, total_shards)
        self._log_shard_info(len(selected), total=len(work_units))
        return selected

    def _filter_features_by_shard(self, features: list[Any]) -> list[Any]:
        """Filter parsed features to only those in the current shard.

        In sequential mode, sharding operates at feature level: features are
        sorted by ``filename`` (matching the parallel mode's work-unit id sort)
        and split into shards.  Only features in the current shard are kept.

        Args:
            features: All parsed features.

        Returns:
            Features belonging to the current shard.
        """
        shard_index = int(getattr(self.config, "shard_index", 0))
        total_shards = int(getattr(self.config, "total_shards", 0))
        sorted_features = sorted(features, key=lambda f: getattr(f, "filename", "") or "")
        from behave_pool.shard import split_shards

        return split_shards(sorted_features, shard_index, total_shards)

    def _log_shard_info(self, selected_count: int, total: int | None = None) -> None:
        """Log shard metadata for visibility in CI output."""
        shard_index = getattr(self.config, "shard_index", None)
        total_shards = getattr(self.config, "total_shards", None)
        if shard_index is None or total_shards is None:
            return
        if total is not None:
            logger.info(
                "Shard %d/%d - %d work units selected (of %d total)",
                shard_index,
                total_shards,
                selected_count,
                total,
            )
        else:
            logger.info(
                "Shard %d/%d - %d features selected",
                shard_index,
                total_shards,
                selected_count,
            )

    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],
        ctx: Any,
    ) -> list[WorkUnit]:
        """Two-phase dispatch: parallel first, then serial.

        Phase 1: enqueue parallel_batch, launch up to N workers (never more
        than there are work units), collect each result as it arrives.
        Phase 2: enqueue serial_batch, launch 1 worker, collect results.

        Results are collected while dispatching so that the per-unit
        timeout (:data:`RESULT_TIMEOUT`) applies to each work unit rather
        than to a whole phase.

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

        # -- Phase 1: parallel batch with N workers.
        if parallel_batch:
            n_workers = min(self.config.parallel, len(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)

            self._early_results.extend(
                self._await_results(result_queue, stop_event, parallel_batch, workers)
            )
            self._shutdown_workers(workers, stop_event)
            self._drain_queue(task_queue)

        # -- 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()

            self._early_results.extend(
                self._await_results(result_queue, stop_event, serial_batch, [serial_worker])
            )
            self._shutdown_workers([serial_worker], stop_event)
            self._drain_queue(task_queue)

        return dispatched

    def _await_results(
        self,
        result_queue: Any,
        stop_event: Any,
        units: list[WorkUnit],
        workers: list[WorkerProcess],
    ) -> list[WorkerResult]:
        """Collect results for ``units`` while the workers are running.

        Each work unit may take up to :data:`RESULT_TIMEOUT` seconds; the
        clock restarts with every received result.  When a unit times out,
        the workers are terminated and the pending units count as missing.
        If ``config.stop`` is set, the first failure sets ``stop_event`` so
        workers drop all queued units.
        """
        pending = {u.id for u in units}
        results: list[WorkerResult] = []
        last_result_at = time.monotonic()
        while pending:
            try:
                result = result_queue.get(timeout=_POLL_INTERVAL)
            except queue.Empty:
                result = None
            except (EOFError, OSError):
                break
            if result is not None:
                results.append(result)
                pending.discard(result.work_unit_id)
                last_result_at = time.monotonic()
                self._print_progress(result)
                if result.failed and self.config.stop:
                    stop_event.set()
            if not pending:
                break
            if time.monotonic() - last_result_at >= RESULT_TIMEOUT:
                logger.warning(
                    "No worker result within %ds; terminating worker(s).",
                    int(RESULT_TIMEOUT),
                )
                stop_event.set()
                break
            if all(not w.is_alive() for w in workers):
                # All workers exited/crashed — drain anything still in flight.
                results.extend(self._drain_results(result_queue, pending))
                break
        return results

    def _drain_results(self, result_queue: Any, pending: set[str]) -> list[WorkerResult]:
        """Drain already-produced results after workers exited."""
        results: list[WorkerResult] = []
        deadline = time.monotonic() + 5.0
        while pending and time.monotonic() < deadline:
            try:
                result = result_queue.get(timeout=0.5)
            except queue.Empty:
                continue
            except (EOFError, OSError):
                break
            results.append(result)
            pending.discard(result.work_unit_id)
            self._print_progress(result)
        return results

    def _shutdown_workers(self, workers: list[WorkerProcess], stop_event: Any) -> None:
        """Wait briefly for workers to exit, then terminate stragglers."""
        for worker in workers:
            worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT)
            if worker.is_alive():
                logger.warning(
                    "Worker %d did not terminate; setting stop event and terminating.",
                    worker.worker_id,
                )
                stop_event.set()
                worker.terminate()

    @staticmethod
    def _drain_queue(task_queue: Any) -> None:
        """Drain any unconsumed items so the queue is empty for the next phase."""
        while not task_queue.empty():
            try:
                task_queue.get_nowait()
                task_queue.task_done()
            except queue.Empty:
                break

    @staticmethod
    def _print_progress(result: WorkerResult) -> None:
        """Print one line per finished work unit (replaces formatter output)."""
        status = "FAILED" if result.failed else "passed"
        if result.error:
            status = f"ERROR ({result.error})"
        print(f"{result.work_unit_id} ... {status} ({result.duration:.2f}s)")

    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] = list(self._early_results)
        received_ids: set[str] = {r.work_unit_id for r in results}

        # 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
            if result.work_unit_id in received_ids:
                continue
            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

        # Propagate undefined steps so --show-snippets works in parallel.
        seen_undefined: set[str] = set()
        for r in results:
            for step_text in r.undefined_steps:
                if step_text not in seen_undefined:
                    seen_undefined.add(step_text)
                    self.undefined_steps.append(step_text)

        self._update_timings(results, work_units)

        statistics = self._merge_reports(results)
        if statistics is not None:
            print(
                f"{statistics['features']} features, {statistics['scenarios']} scenarios, "
                f"{statistics['steps']} steps - "
                f"passed: {statistics['passed']}, failed: {statistics['failed']}, "
                f"skipped: {statistics['skipped']}, undefined: {statistics['undefined']}"
            )

        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], work_units: list[WorkUnit] | None = None
    ) -> None:
        """Update the TimingStore with observed durations from results.

        Entries whose feature file no longer exists are pruned so that
        renamed or deleted features do not accumulate forever.

        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.
            work_units: Dispatched work units (unused; kept for
                backward compatibility of the method signature).
        """
        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.prune_stale()
            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]) -> dict[str, Any] | 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.

        Returns:
            The computed statistics dict, or None if writing failed.
        """
        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)

        # Same file may appear in several work units (scenario scheme or a
        # mixed feature split into parallel/serial halves) — merge those
        # partial entries into a single feature dict.
        all_features = self._merge_feature_dicts(all_features)

        statistics: dict[str, Any] | None = 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)
            statistics = None

        tmp_dir = Path("tmp")
        if tmp_dir.is_dir():
            for report_file in tmp_dir.glob("worker_*.json"):
                with contextlib.suppress(OSError):
                    report_file.unlink()
            if not any(tmp_dir.iterdir()):
                with contextlib.suppress(OSError):
                    tmp_dir.rmdir()

        return statistics

    @staticmethod
    def _merge_feature_dicts(features: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """Merge partial feature dicts that share the same ``filename``.

        Work units that split a feature (``#serial`` halves or per-scenario
        units under ``--parallel-scheme scenario``) produce one report entry
        each; they are combined so every file appears once, with all its
        scenarios, summed duration, and the worst scenario status.
        """
        severity = {
            "untested": 0,
            "skipped": 1,
            "passed": 2,
            "pending": 3,
            "undefined": 3,
            "failed": 4,
            "error": 4,
            "hook_error": 4,
            "cleanup_error": 4,
        }
        merged: dict[str, dict[str, Any]] = {}
        order: list[str] = []
        for feature in features:
            key = str(feature.get("filename") or feature.get("id") or id(feature))
            if key not in merged:
                entry = dict(feature)
                entry["scenarios"] = list(feature.get("scenarios") or [])
                merged[key] = entry
                order.append(key)
                continue
            entry = merged[key]
            entry["scenarios"].extend(feature.get("scenarios") or [])
            entry["duration"] = (entry.get("duration") or 0.0) + (feature.get("duration") or 0.0)
        for key in order:
            entry = merged[key]
            scenarios = entry["scenarios"]
            if not scenarios:
                continue
            worst = max(
                (str(s.get("status", "untested")) for s in scenarios),
                key=lambda s: severity.get(s, 0),
            )
            entry["status"] = worst if worst in severity else "untested"
        return [merged[k] for k in order]

    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 = _sanitize_command(" ".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"] = _sanitize_url(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

        started_at = getattr(self, "_exec_started_at", None)
        ended_at = getattr(self, "_exec_ended_at", None) or datetime.now(UTC)
        if started_at is None:
            started_at = ended_at
        wall_duration = getattr(self, "_exec_wall_duration", None)
        if wall_duration is None:
            wall_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(wall_duration, 6),
            "startTime": started_at.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
            "endTime": ended_at.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
        }
        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()

ShardConfig dataclass

Configuration for sharding.

Attributes:

Name	Type	Description
shard_index	int	1-based index of the shard to execute.
total_shards	int	Total number of shards the suite is divided into.
features_dir	str	Path to the features directory to scan.
parallel	int	Number of local worker processes (0 or 1 = sequential).

Source code in behave_pool/shard.py

@dataclass(frozen=True)
class ShardConfig:
    """Configuration for sharding.

    Attributes:
        shard_index: 1-based index of the shard to execute.
        total_shards: Total number of shards the suite is divided into.
        features_dir: Path to the features directory to scan.
        parallel: Number of local worker processes (0 or 1 = sequential).
    """

    shard_index: int
    total_shards: int
    features_dir: str = "features/"
    parallel: int = 1

    def __post_init__(self) -> None:
        validate_shard(self.shard_index, self.total_shards)

run_with_shard

run_with_shard(config: ShardConfig) -> bool

Run a single shard of the test suite.

Plans the work units from config.features_dir, selects the shard deterministically, and executes it. When config.parallel > 1, work units within the shard are distributed among local workers.

Parameters:

Name	Type	Description	Default
config	ShardConfig	Shard configuration.	required

Returns:

Type	Description
bool	True if any test failed (Behave convention).

Source code in behave_pool/shard.py

def run_with_shard(config: ShardConfig) -> bool:
    """Run a single shard of the test suite.

    Plans the work units from ``config.features_dir``, selects the shard
    deterministically, and executes it.  When ``config.parallel > 1``,
    work units within the shard are distributed among local workers.

    Args:
        config: Shard configuration.

    Returns:
        ``True`` if any test failed (Behave convention).
    """
    from behave.configuration import Configuration

    from behave_pool.runner import ParallelRunner

    behave_config = Configuration(
        command_args=[config.features_dir],
        load_config=False,
        jobs=config.parallel,
    )
    behave_config.shard_index = config.shard_index
    behave_config.total_shards = config.total_shards

    runner = ParallelRunner(behave_config)
    return runner.run()

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)
        self._early_results: list[WorkerResult] = []
        self._exec_started_at: Any = None
        self._exec_ended_at: Any = None
        self._exec_wall_duration: float | None = None

    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)

        if self._is_sharding_active():
            features = self._filter_features_by_shard(features)
            self._log_shard_info(len(features))

        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 -> shard -> split -> dispatch -> collect."""
        from datetime import UTC, datetime

        ctx = multiprocessing.get_context("spawn")
        task_queue: Any = ctx.JoinableQueue()
        result_queue: Any = ctx.Queue()
        stop_event: Any = ctx.Event()

        self._exec_started_at = datetime.now(UTC)
        started = time.monotonic()
        self._early_results = []
        try:
            work_units = self._plan()
            if self._is_sharding_active():
                work_units = self._apply_shard(work_units)
            parallel_batch, serial_batch = self._split_by_serial_tag(work_units)
            dispatched = self._dispatch(
                task_queue, result_queue, stop_event, parallel_batch, serial_batch, ctx
            )
            return self._collect(result_queue, dispatched)
        finally:
            self._exec_ended_at = datetime.now(UTC)
            self._exec_wall_duration = time.monotonic() - started
            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 _is_sharding_active(self) -> bool:
        """Check whether sharding is enabled for this run."""
        shard_index = getattr(self.config, "shard_index", None)
        total_shards = getattr(self.config, "total_shards", None)
        return shard_index is not None and total_shards is not None

    def _apply_shard(self, work_units: list[WorkUnit]) -> list[WorkUnit]:
        """Filter work units to only those in the current shard.

        Shard membership is computed on ids sorted deterministically, but
        the returned list preserves the planned (LPT/FIFO) order.

        Args:
            work_units: All planned work units.

        Returns:
            Work units belonging to the current shard.
        """
        shard_index = int(getattr(self.config, "shard_index", 0))
        total_shards = int(getattr(self.config, "total_shards", 0))
        selected = select_shard_work_units(work_units, shard_index, total_shards)
        self._log_shard_info(len(selected), total=len(work_units))
        return selected

    def _filter_features_by_shard(self, features: list[Any]) -> list[Any]:
        """Filter parsed features to only those in the current shard.

        In sequential mode, sharding operates at feature level: features are
        sorted by ``filename`` (matching the parallel mode's work-unit id sort)
        and split into shards.  Only features in the current shard are kept.

        Args:
            features: All parsed features.

        Returns:
            Features belonging to the current shard.
        """
        shard_index = int(getattr(self.config, "shard_index", 0))
        total_shards = int(getattr(self.config, "total_shards", 0))
        sorted_features = sorted(features, key=lambda f: getattr(f, "filename", "") or "")
        from behave_pool.shard import split_shards

        return split_shards(sorted_features, shard_index, total_shards)

    def _log_shard_info(self, selected_count: int, total: int | None = None) -> None:
        """Log shard metadata for visibility in CI output."""
        shard_index = getattr(self.config, "shard_index", None)
        total_shards = getattr(self.config, "total_shards", None)
        if shard_index is None or total_shards is None:
            return
        if total is not None:
            logger.info(
                "Shard %d/%d - %d work units selected (of %d total)",
                shard_index,
                total_shards,
                selected_count,
                total,
            )
        else:
            logger.info(
                "Shard %d/%d - %d features selected",
                shard_index,
                total_shards,
                selected_count,
            )

    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],
        ctx: Any,
    ) -> list[WorkUnit]:
        """Two-phase dispatch: parallel first, then serial.

        Phase 1: enqueue parallel_batch, launch up to N workers (never more
        than there are work units), collect each result as it arrives.
        Phase 2: enqueue serial_batch, launch 1 worker, collect results.

        Results are collected while dispatching so that the per-unit
        timeout (:data:`RESULT_TIMEOUT`) applies to each work unit rather
        than to a whole phase.

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

        # -- Phase 1: parallel batch with N workers.
        if parallel_batch:
            n_workers = min(self.config.parallel, len(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)

            self._early_results.extend(
                self._await_results(result_queue, stop_event, parallel_batch, workers)
            )
            self._shutdown_workers(workers, stop_event)
            self._drain_queue(task_queue)

        # -- 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()

            self._early_results.extend(
                self._await_results(result_queue, stop_event, serial_batch, [serial_worker])
            )
            self._shutdown_workers([serial_worker], stop_event)
            self._drain_queue(task_queue)

        return dispatched

    def _await_results(
        self,
        result_queue: Any,
        stop_event: Any,
        units: list[WorkUnit],
        workers: list[WorkerProcess],
    ) -> list[WorkerResult]:
        """Collect results for ``units`` while the workers are running.

        Each work unit may take up to :data:`RESULT_TIMEOUT` seconds; the
        clock restarts with every received result.  When a unit times out,
        the workers are terminated and the pending units count as missing.
        If ``config.stop`` is set, the first failure sets ``stop_event`` so
        workers drop all queued units.
        """
        pending = {u.id for u in units}
        results: list[WorkerResult] = []
        last_result_at = time.monotonic()
        while pending:
            try:
                result = result_queue.get(timeout=_POLL_INTERVAL)
            except queue.Empty:
                result = None
            except (EOFError, OSError):
                break
            if result is not None:
                results.append(result)
                pending.discard(result.work_unit_id)
                last_result_at = time.monotonic()
                self._print_progress(result)
                if result.failed and self.config.stop:
                    stop_event.set()
            if not pending:
                break
            if time.monotonic() - last_result_at >= RESULT_TIMEOUT:
                logger.warning(
                    "No worker result within %ds; terminating worker(s).",
                    int(RESULT_TIMEOUT),
                )
                stop_event.set()
                break
            if all(not w.is_alive() for w in workers):
                # All workers exited/crashed — drain anything still in flight.
                results.extend(self._drain_results(result_queue, pending))
                break
        return results

    def _drain_results(self, result_queue: Any, pending: set[str]) -> list[WorkerResult]:
        """Drain already-produced results after workers exited."""
        results: list[WorkerResult] = []
        deadline = time.monotonic() + 5.0
        while pending and time.monotonic() < deadline:
            try:
                result = result_queue.get(timeout=0.5)
            except queue.Empty:
                continue
            except (EOFError, OSError):
                break
            results.append(result)
            pending.discard(result.work_unit_id)
            self._print_progress(result)
        return results

    def _shutdown_workers(self, workers: list[WorkerProcess], stop_event: Any) -> None:
        """Wait briefly for workers to exit, then terminate stragglers."""
        for worker in workers:
            worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT)
            if worker.is_alive():
                logger.warning(
                    "Worker %d did not terminate; setting stop event and terminating.",
                    worker.worker_id,
                )
                stop_event.set()
                worker.terminate()

    @staticmethod
    def _drain_queue(task_queue: Any) -> None:
        """Drain any unconsumed items so the queue is empty for the next phase."""
        while not task_queue.empty():
            try:
                task_queue.get_nowait()
                task_queue.task_done()
            except queue.Empty:
                break

    @staticmethod
    def _print_progress(result: WorkerResult) -> None:
        """Print one line per finished work unit (replaces formatter output)."""
        status = "FAILED" if result.failed else "passed"
        if result.error:
            status = f"ERROR ({result.error})"
        print(f"{result.work_unit_id} ... {status} ({result.duration:.2f}s)")

    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] = list(self._early_results)
        received_ids: set[str] = {r.work_unit_id for r in results}

        # 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
            if result.work_unit_id in received_ids:
                continue
            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

        # Propagate undefined steps so --show-snippets works in parallel.
        seen_undefined: set[str] = set()
        for r in results:
            for step_text in r.undefined_steps:
                if step_text not in seen_undefined:
                    seen_undefined.add(step_text)
                    self.undefined_steps.append(step_text)

        self._update_timings(results, work_units)

        statistics = self._merge_reports(results)
        if statistics is not None:
            print(
                f"{statistics['features']} features, {statistics['scenarios']} scenarios, "
                f"{statistics['steps']} steps - "
                f"passed: {statistics['passed']}, failed: {statistics['failed']}, "
                f"skipped: {statistics['skipped']}, undefined: {statistics['undefined']}"
            )

        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], work_units: list[WorkUnit] | None = None
    ) -> None:
        """Update the TimingStore with observed durations from results.

        Entries whose feature file no longer exists are pruned so that
        renamed or deleted features do not accumulate forever.

        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.
            work_units: Dispatched work units (unused; kept for
                backward compatibility of the method signature).
        """
        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.prune_stale()
            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]) -> dict[str, Any] | 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.

        Returns:
            The computed statistics dict, or None if writing failed.
        """
        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)

        # Same file may appear in several work units (scenario scheme or a
        # mixed feature split into parallel/serial halves) — merge those
        # partial entries into a single feature dict.
        all_features = self._merge_feature_dicts(all_features)

        statistics: dict[str, Any] | None = 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)
            statistics = None

        tmp_dir = Path("tmp")
        if tmp_dir.is_dir():
            for report_file in tmp_dir.glob("worker_*.json"):
                with contextlib.suppress(OSError):
                    report_file.unlink()
            if not any(tmp_dir.iterdir()):
                with contextlib.suppress(OSError):
                    tmp_dir.rmdir()

        return statistics

    @staticmethod
    def _merge_feature_dicts(features: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """Merge partial feature dicts that share the same ``filename``.

        Work units that split a feature (``#serial`` halves or per-scenario
        units under ``--parallel-scheme scenario``) produce one report entry
        each; they are combined so every file appears once, with all its
        scenarios, summed duration, and the worst scenario status.
        """
        severity = {
            "untested": 0,
            "skipped": 1,
            "passed": 2,
            "pending": 3,
            "undefined": 3,
            "failed": 4,
            "error": 4,
            "hook_error": 4,
            "cleanup_error": 4,
        }
        merged: dict[str, dict[str, Any]] = {}
        order: list[str] = []
        for feature in features:
            key = str(feature.get("filename") or feature.get("id") or id(feature))
            if key not in merged:
                entry = dict(feature)
                entry["scenarios"] = list(feature.get("scenarios") or [])
                merged[key] = entry
                order.append(key)
                continue
            entry = merged[key]
            entry["scenarios"].extend(feature.get("scenarios") or [])
            entry["duration"] = (entry.get("duration") or 0.0) + (feature.get("duration") or 0.0)
        for key in order:
            entry = merged[key]
            scenarios = entry["scenarios"]
            if not scenarios:
                continue
            worst = max(
                (str(s.get("status", "untested")) for s in scenarios),
                key=lambda s: severity.get(s, 0),
            )
            entry["status"] = worst if worst in severity else "untested"
        return [merged[k] for k in order]

    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 = _sanitize_command(" ".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"] = _sanitize_url(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

        started_at = getattr(self, "_exec_started_at", None)
        ended_at = getattr(self, "_exec_ended_at", None) or datetime.now(UTC)
        if started_at is None:
            started_at = ended_at
        wall_duration = getattr(self, "_exec_wall_duration", None)
        if wall_duration is None:
            wall_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(wall_duration, 6),
            "startTime": started_at.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
            "endTime": ended_at.isoformat(timespec="milliseconds").replace("+00:00", "Z"),
        }
        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"
    shard_index: int | None = None
    total_shards: int | None = None
    dry_run: bool = False
    use_nested_step_modules: bool = False
    tags: list[str] | str | None = None
    config_tags: str | list[str] | None = None
    default_tags: str | None = None
    tag_expression_protocol: str | None = None
    name: list[str] | None = None
    userdata: dict[str, Any] = field(default_factory=dict)

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 resolves the behave-pool options.

Resolution order for each option:

1. Explicit config value (CLI flag or ini dest key — only reachable when behave-pool's options were registered before argument parsing, e.g. via the behave-pool command or python -m behave_pool).
2. userdata entry pool.<name> (works with plain behave via -D pool.balance=fifo or a [behave.userdata] section).
3. Registered default.

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 resolves the behave-pool options.

    Resolution order for each option:

    1. Explicit config value (CLI flag or ini ``dest`` key — only
       reachable when behave-pool's options were registered before
       argument parsing, e.g. via the ``behave-pool`` command or
       ``python -m behave_pool``).
    2. ``userdata`` entry ``pool.<name>`` (works with plain ``behave``
       via ``-D pool.balance=fifo`` or a ``[behave.userdata]`` section).
    3. Registered default.

    Args:
        config: Behave Configuration instance to augment.
    """
    jobs = getattr(config, "jobs", 1)
    if jobs is None or jobs == 1:
        userdata_jobs = _userdata_get(config, "jobs")
        if userdata_jobs is not None:
            import contextlib

            with contextlib.suppress(TypeError, ValueError):
                jobs = int(str(userdata_jobs))
    config.parallel = jobs if jobs else 1

    config.parallel_scheme = _resolve_option(config, "parallel_scheme", "feature", "scheme")
    config.parallel_balance = _resolve_option(config, "parallel_balance", "lpt", "balance")
    config.parallel_timing_file = _resolve_option(
        config, "parallel_timing_file", ".behave-pool-timing.json", "timing_file"
    )
    config.parallel_report = _resolve_option(
        config, "parallel_report", "behave-pool-report.json", "report"
    )

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

    shard_value = getattr(config, "shard", None)
    if shard_value is None:
        shard_value = _userdata_get(config, "shard")
    if shard_value is not None:
        from behave_pool.shard import parse_shard_string

        shard_cfg = parse_shard_string(str(shard_value))
        config.shard_index = shard_cfg.shard_index
        config.total_shards = shard_cfg.total_shards
    else:
        if not hasattr(config, "shard_index"):
            config.shard_index = None
        if not hasattr(config, "total_shards"):
            config.total_shards = None

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."""
    tags = getattr(config, "tags", None)
    if isinstance(tags, (list, tuple)):
        tags = [str(t) for t in tags]
    elif tags is not None:
        tags = str(tags)
    name = getattr(config, "name", None)
    name_list = [str(n) for n in name] if isinstance(name, (list, tuple)) else None
    userdata = getattr(config, "userdata", None) or {}
    if not isinstance(userdata, dict):
        userdata = {}
    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"),
        shard_index=getattr(config, "shard_index", None),
        total_shards=getattr(config, "total_shards", None),
        dry_run=config.dry_run,
        use_nested_step_modules=getattr(config, "use_nested_step_modules", False),
        tags=tags,
        config_tags=getattr(config, "config_tags", None),
        default_tags=getattr(config, "default_tags", None),
        tag_expression_protocol=getattr(config, "tag_expression_protocol", None),
        name=name_list,
        userdata=dict(userdata),
    )

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",
        environment_file="environment.py",
        lang=None,
        stop=False,
    ),
    feature_path="features/login.feature",
    tags=["serial"],
    serial_mode="only",
)

print(unit.is_serial)  # True

serial_mode selects which scenarios of a feature the unit executes: "all" (default), "only" (@serial scenarios), or "exclude" (everything except @serial). Mixed features are split by the planner into a "exclude" parallel unit and a "only" serial unit.

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.
serial_mode	str	Which scenarios of the feature to run: "all" (default, whole feature), "only" (only @serial scenarios), or "exclude" (all except @serial). Used to split a mixed feature into a parallel and a serial work unit.

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.
        serial_mode: Which scenarios of the feature to run: ``"all"``
            (default, whole feature), ``"only"`` (only ``@serial``
            scenarios), or ``"exclude"`` (all except ``@serial``).
            Used to split a mixed feature into a parallel and a serial
            work unit.
    """

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

    @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 prune_stale(self) -> int:
        """Remove entries whose feature file no longer exists.

        Work unit ids embed the feature path (``feature:<path>``,
        ``feature:<path>#serial``, ``scenario:<path>:<line>``).  Entries
        for deleted or renamed feature files are dropped so the timing
        file does not grow forever.  Entries whose path cannot be
        extracted are kept.

        Returns:
            Number of entries removed.
        """
        if not self._loaded:
            self.load()
        removed = 0
        for uid in list(self._data):
            path = _work_unit_feature_path(uid)
            if path and not Path(path).exists():
                del self._data[uid]
                removed += 1
        return removed

    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

prune_stale

prune_stale() -> int

Remove entries whose feature file no longer exists.

Work unit ids embed the feature path (feature:<path>, feature:<path>#serial, scenario:<path>:<line>). Entries for deleted or renamed feature files are dropped so the timing file does not grow forever. Entries whose path cannot be extracted are kept.

Returns:

Type	Description
int	Number of entries removed.

Source code in behave_pool/timing.py

def prune_stale(self) -> int:
    """Remove entries whose feature file no longer exists.

    Work unit ids embed the feature path (``feature:<path>``,
    ``feature:<path>#serial``, ``scenario:<path>:<line>``).  Entries
    for deleted or renamed feature files are dropped so the timing
    file does not grow forever.  Entries whose path cannot be
    extracted are kept.

    Returns:
        Number of entries removed.
    """
    if not self._loaded:
        self.load()
    removed = 0
    for uid in list(self._data):
        path = _work_unit_feature_path(uid)
        if path and not Path(path).exists():
            del self._data[uid]
            removed += 1
    return removed

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.

When a feature mixes @serial and non-serial scenarios, it is split into two work units: a parallel unit that runs the non-serial scenarios (serial_mode="exclude") and a serial unit that runs only the @serial scenarios (serial_mode="only").

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.

    When a feature mixes ``@serial`` and non-serial scenarios, it is split
    into two work units: a parallel unit that runs the non-serial
    scenarios (``serial_mode="exclude"``) and a serial unit that runs only
    the ``@serial`` scenarios (``serial_mode="only"``).
    """

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

    def iterate(self) -> Iterator[WorkUnit]:
        """Yield work units per feature, splitting on the ``@serial`` tag.

        A feature tagged ``@serial`` at feature level produces a single
        serial work unit.  A feature with a mix of ``@serial`` and
        non-serial scenarios produces up to two work units.  Otherwise
        one parallel work unit is produced per feature.
        """
        from behave_pool.serial import is_serial_tagged, item_is_serial

        for feature in self._features:
            feature_tags = {str(t) for t in (getattr(feature, "tags", None) or [])}
            run_items = list(getattr(feature, "run_items", None) or [])
            base_id = f"feature:{feature.filename}"

            if is_serial_tagged(feature):
                tags = set(feature_tags)
                for scenario in getattr(feature, "scenarios", None) or []:
                    tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))
                tags.add("serial")
                yield WorkUnit(
                    id=base_id,
                    config=snapshot_config(self._config),
                    feature_path=feature.filename,
                    scenario_line=None,
                    tags=sorted(tags),
                    serial_mode="all",
                )
                continue

            serial_items = [i for i in run_items if item_is_serial(i)]
            if not serial_items:
                tags = set(feature_tags)
                for scenario in getattr(feature, "scenarios", None) or []:
                    tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))
                yield WorkUnit(
                    id=base_id,
                    config=snapshot_config(self._config),
                    feature_path=feature.filename,
                    scenario_line=None,
                    tags=sorted(tags),
                )
                continue

            # -- MIXED FEATURE: split into parallel + serial units.
            all_tags = set(feature_tags)
            for scenario in getattr(feature, "scenarios", None) or []:
                all_tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))

            parallel_tags = {t for t in all_tags if t != "serial"}
            if any(not item_is_serial(i) for i in run_items):
                yield WorkUnit(
                    id=base_id,
                    config=snapshot_config(self._config),
                    feature_path=feature.filename,
                    scenario_line=None,
                    tags=sorted(parallel_tags),
                    serial_mode="exclude",
                )
            yield WorkUnit(
                id=f"{base_id}#serial",
                config=snapshot_config(self._config),
                feature_path=feature.filename,
                scenario_line=None,
                tags=sorted(all_tags | {"serial"}),
                serial_mode="only",
            )

iterate

iterate() -> Iterator[WorkUnit]

Yield work units per feature, splitting on the @serial tag.

A feature tagged @serial at feature level produces a single serial work unit. A feature with a mix of @serial and non-serial scenarios produces up to two work units. Otherwise one parallel work unit is produced per feature.

Source code in behave_pool/iterator.py

def iterate(self) -> Iterator[WorkUnit]:
    """Yield work units per feature, splitting on the ``@serial`` tag.

    A feature tagged ``@serial`` at feature level produces a single
    serial work unit.  A feature with a mix of ``@serial`` and
    non-serial scenarios produces up to two work units.  Otherwise
    one parallel work unit is produced per feature.
    """
    from behave_pool.serial import is_serial_tagged, item_is_serial

    for feature in self._features:
        feature_tags = {str(t) for t in (getattr(feature, "tags", None) or [])}
        run_items = list(getattr(feature, "run_items", None) or [])
        base_id = f"feature:{feature.filename}"

        if is_serial_tagged(feature):
            tags = set(feature_tags)
            for scenario in getattr(feature, "scenarios", None) or []:
                tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))
            tags.add("serial")
            yield WorkUnit(
                id=base_id,
                config=snapshot_config(self._config),
                feature_path=feature.filename,
                scenario_line=None,
                tags=sorted(tags),
                serial_mode="all",
            )
            continue

        serial_items = [i for i in run_items if item_is_serial(i)]
        if not serial_items:
            tags = set(feature_tags)
            for scenario in getattr(feature, "scenarios", None) or []:
                tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))
            yield WorkUnit(
                id=base_id,
                config=snapshot_config(self._config),
                feature_path=feature.filename,
                scenario_line=None,
                tags=sorted(tags),
            )
            continue

        # -- MIXED FEATURE: split into parallel + serial units.
        all_tags = set(feature_tags)
        for scenario in getattr(feature, "scenarios", None) or []:
            all_tags.update(str(t) for t in (getattr(scenario, "tags", None) or []))

        parallel_tags = {t for t in all_tags if t != "serial"}
        if any(not item_is_serial(i) for i in run_items):
            yield WorkUnit(
                id=base_id,
                config=snapshot_config(self._config),
                feature_path=feature.filename,
                scenario_line=None,
                tags=sorted(parallel_tags),
                serial_mode="exclude",
            )
        yield WorkUnit(
            id=f"{base_id}#serial",
            config=snapshot_config(self._config),
            feature_path=feature.filename,
            scenario_line=None,
            tags=sorted(all_tags | {"serial"}),
            serial_mode="only",
        )

ScenarioIterator

Bases: WorkUnitIterator

Generate one WorkUnit per scenario run item.

A unit maps to a single run item: a Scenario or a whole ScenarioOutline (all its examples run together — outlines are the declaration-level unit). Scenarios inside Rule blocks are flattened to individual units.

Items that carry the @serial tag (directly, via the feature/rule, or — for outlines — via any generated example) are produced as serial units and run in the serial phase. In particular, an outline with only some @serial example rows runs its whole outline serially: conservative, but correct.

Source code in behave_pool/iterator.py

class ScenarioIterator(WorkUnitIterator):
    """Generate one WorkUnit per scenario run item.

    A unit maps to a single run item: a Scenario or a whole
    ScenarioOutline (all its examples run together — outlines are the
    declaration-level unit).  Scenarios inside ``Rule`` blocks are
    flattened to individual units.

    Items that carry the ``@serial`` tag (directly, via the feature/rule,
    or — for outlines — via any generated example) are produced as serial
    units and run in the serial phase.  In particular, an outline with
    only *some* ``@serial`` example rows runs its whole outline serially:
    conservative, but correct.
    """

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

    def iterate(self) -> Iterator[WorkUnit]:
        """Yield one work unit per scenario or scenario outline."""
        from behave.model import Rule

        from behave_pool.serial import is_serial_tagged, item_is_serial

        def _items(container: Any) -> Iterator[Any]:
            for item in getattr(container, "run_items", None) or []:
                if isinstance(item, Rule):
                    yield from _items(item)
                else:
                    yield item

        for feature in self._features:
            feature_tags = {str(t) for t in (getattr(feature, "tags", None) or [])}
            feature_serial = is_serial_tagged(feature)
            for item in _items(feature):
                line = getattr(item, "line", None)
                if line is None:
                    continue
                tags = set(feature_tags)
                item_tags = (
                    getattr(item, "effective_tags", None) or getattr(item, "tags", None) or []
                )
                tags.update(str(t) for t in item_tags)
                if feature_serial or item_is_serial(item):
                    tags.add("serial")
                yield WorkUnit(
                    id=f"scenario:{feature.filename}:{line}",
                    config=snapshot_config(self._config),
                    feature_path=feature.filename,
                    scenario_line=line,
                    tags=sorted(tags),
                    serial_mode="all",
                )

iterate

iterate() -> Iterator[WorkUnit]

Yield one work unit per scenario or scenario outline.

Source code in behave_pool/iterator.py

def iterate(self) -> Iterator[WorkUnit]:
    """Yield one work unit per scenario or scenario outline."""
    from behave.model import Rule

    from behave_pool.serial import is_serial_tagged, item_is_serial

    def _items(container: Any) -> Iterator[Any]:
        for item in getattr(container, "run_items", None) or []:
            if isinstance(item, Rule):
                yield from _items(item)
            else:
                yield item

    for feature in self._features:
        feature_tags = {str(t) for t in (getattr(feature, "tags", None) or [])}
        feature_serial = is_serial_tagged(feature)
        for item in _items(feature):
            line = getattr(item, "line", None)
            if line is None:
                continue
            tags = set(feature_tags)
            item_tags = (
                getattr(item, "effective_tags", None) or getattr(item, "tags", None) or []
            )
            tags.update(str(t) for t in item_tags)
            if feature_serial or item_is_serial(item):
                tags.add("serial")
            yield WorkUnit(
                id=f"scenario:{feature.filename}:{line}",
                config=snapshot_config(self._config),
                feature_path=feature.filename,
                scenario_line=line,
                tags=sorted(tags),
                serial_mode="all",
            )

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.
        """
        if scheme == "feature":
            return FeatureIterator(features, config)
        if scheme == "scenario":
            return ScenarioIterator(features, config)
        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.

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.
    """
    if scheme == "feature":
        return FeatureIterator(features, config)
    if scheme == "scenario":
        return ScenarioIterator(features, config)
    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."""
    ...

Sharding

ShardConfig configures sharding for CI parallelism. run_with_shard is a convenience function that creates a ParallelRunner with shard settings.

from behave_pool import ShardConfig, run_with_shard

config = ShardConfig(
    shard_index=1,
    total_shards=3,
    features_dir="features/",
    parallel=4,
)
failed = run_with_shard(config)

parse_shard_string parses INDEX/TOTAL strings:

from behave_pool.shard import parse_shard_string

config = parse_shard_string("2/3")
print(config.shard_index)  # 2
print(config.total_shards)  # 3

split_shards divides a sorted list into contiguous groups:

from behave_pool.shard import split_shards

items = list(range(10))
shard1 = split_shards(items, 1, 3)  # [0, 1, 2, 3]
shard2 = split_shards(items, 2, 3)  # [4, 5, 6]
shard3 = split_shards(items, 3, 3)  # [7, 8, 9]

behave_pool.shard

Sharding support for CI parallelism across multiple machines.

Sharding divides the work units (one per feature file, plus the extra #serial units produced by mixed features) into total_shards contiguous groups. Each CI runner executes only its assigned shard (shard_index).

The algorithm:

1. Build the work units from the parsed features.
2. Sort a copy deterministically by work unit id (i.e. feature path) to compute shard membership identically on every machine.
3. Split the sorted list into total_shards groups. The first len % total_shards shards receive one extra unit.
4. Return the shard_index-th group (1-based) in the original input order, so the planned LPT/FIFO order is preserved.

Sharding composes with --parallel (local parallelism within a shard), --parallel-balance (ordering is preserved inside the shard), and @serial (serial work units within a shard run sequentially).

ShardConfig dataclass

Configuration for sharding.

Attributes:

Name	Type	Description
shard_index	int	1-based index of the shard to execute.
total_shards	int	Total number of shards the suite is divided into.
features_dir	str	Path to the features directory to scan.
parallel	int	Number of local worker processes (0 or 1 = sequential).

Source code in behave_pool/shard.py

@dataclass(frozen=True)
class ShardConfig:
    """Configuration for sharding.

    Attributes:
        shard_index: 1-based index of the shard to execute.
        total_shards: Total number of shards the suite is divided into.
        features_dir: Path to the features directory to scan.
        parallel: Number of local worker processes (0 or 1 = sequential).
    """

    shard_index: int
    total_shards: int
    features_dir: str = "features/"
    parallel: int = 1

    def __post_init__(self) -> None:
        validate_shard(self.shard_index, self.total_shards)

ShardError

Bases: ValueError, ConfigError

Raised when shard configuration or parsing is invalid.

Also inherits from :class:behave.exception.ConfigError so that behave's main() catches it and prints a clean error message instead of a raw traceback.

Source code in behave_pool/shard.py

class ShardError(ValueError, ConfigError):  # type: ignore[misc]
    """Raised when shard configuration or parsing is invalid.

    Also inherits from :class:`behave.exception.ConfigError` so that
    behave's ``main()`` catches it and prints a clean error message
    instead of a raw traceback.
    """

parse_shard_string

parse_shard_string(value: str) -> ShardConfig

Parse a INDEX/TOTAL shard string into a :class:ShardConfig.

Parameters:

Name	Type	Description	Default
value	str	String like "1/3".	required

Returns:

Name	Type	Description
A	ShardConfig	class:ShardConfig with shard_index and total_shards set.

Raises:

Type	Description
ShardError	If the string is malformed or values are out of range.

Source code in behave_pool/shard.py

def parse_shard_string(value: str) -> ShardConfig:
    """Parse a ``INDEX/TOTAL`` shard string into a :class:`ShardConfig`.

    Args:
        value: String like ``"1/3"``.

    Returns:
        A :class:`ShardConfig` with ``shard_index`` and ``total_shards`` set.

    Raises:
        ShardError: If the string is malformed or values are out of range.
    """
    match = _SHARD_RE.match(value.strip())
    if not match:
        msg = f"Invalid shard format {value!r}. Expected 'INDEX/TOTAL' (e.g. '1/3')."
        raise ShardError(msg)

    shard_index = int(match.group("index"))
    total_shards = int(match.group("total"))
    validate_shard(shard_index, total_shards)
    return ShardConfig(shard_index=shard_index, total_shards=total_shards)

run_with_shard

run_with_shard(config: ShardConfig) -> bool

Run a single shard of the test suite.

Plans the work units from config.features_dir, selects the shard deterministically, and executes it. When config.parallel > 1, work units within the shard are distributed among local workers.

Parameters:

Name	Type	Description	Default
config	ShardConfig	Shard configuration.	required

Returns:

Type	Description
bool	True if any test failed (Behave convention).

Source code in behave_pool/shard.py

def run_with_shard(config: ShardConfig) -> bool:
    """Run a single shard of the test suite.

    Plans the work units from ``config.features_dir``, selects the shard
    deterministically, and executes it.  When ``config.parallel > 1``,
    work units within the shard are distributed among local workers.

    Args:
        config: Shard configuration.

    Returns:
        ``True`` if any test failed (Behave convention).
    """
    from behave.configuration import Configuration

    from behave_pool.runner import ParallelRunner

    behave_config = Configuration(
        command_args=[config.features_dir],
        load_config=False,
        jobs=config.parallel,
    )
    behave_config.shard_index = config.shard_index
    behave_config.total_shards = config.total_shards

    runner = ParallelRunner(behave_config)
    return runner.run()

select_shard_work_units

select_shard_work_units(work_units: list[WorkUnit], shard_index: int, total_shards: int) -> list[WorkUnit]

Select the work units belonging to a shard.

Work units are sorted by their id (which encodes the feature path) to compute the shard assignment deterministically across machines, but the returned list preserves the incoming (e.g. LPT) order so that balancing is not lost.

Parameters:

Name	Type	Description	Default
work_units	list[WorkUnit]	All work units from the planning phase.	required
shard_index	int	1-based shard index.	required
total_shards	int	Total number of shards.	required

Returns:

Type	Description
list[WorkUnit]	Work units assigned to the requested shard, in input order.

Source code in behave_pool/shard.py

def select_shard_work_units(
    work_units: list[WorkUnit],
    shard_index: int,
    total_shards: int,
) -> list[WorkUnit]:
    """Select the work units belonging to a shard.

    Work units are sorted by their ``id`` (which encodes the feature path)
    to compute the shard assignment deterministically across machines,
    but the returned list preserves the incoming (e.g. LPT) order so that
    balancing is not lost.

    Args:
        work_units: All work units from the planning phase.
        shard_index: 1-based shard index.
        total_shards: Total number of shards.

    Returns:
        Work units assigned to the requested shard, in input order.
    """
    sorted_units = sorted(work_units, key=lambda u: u.id)
    selected = split_shards(sorted_units, shard_index, total_shards)
    selected_ids = {id(u) for u in selected}
    return [u for u in work_units if id(u) in selected_ids]

split_shards

split_shards(items: list[T], shard_index: int, total_shards: int) -> list[T]

Split a list into total_shards contiguous groups and return group shard_index.

The first len(items) % total_shards shards receive one extra item.

Parameters:

Name	Type	Description	Default
items	list[T]	Sorted list of items to split.	required
shard_index	int	1-based shard index.	required
total_shards	int	Total number of shards.	required

Returns:

Type	Description
list[T]	The slice of items belonging to the requested shard.

Source code in behave_pool/shard.py

def split_shards(
    items: list[T],
    shard_index: int,
    total_shards: int,
) -> list[T]:
    """Split a list into ``total_shards`` contiguous groups and return group ``shard_index``.

    The first ``len(items) % total_shards`` shards receive one extra item.

    Args:
        items: Sorted list of items to split.
        shard_index: 1-based shard index.
        total_shards: Total number of shards.

    Returns:
        The slice of items belonging to the requested shard.
    """
    validate_shard(shard_index, total_shards)
    n = len(items)
    base_size = n // total_shards
    remainder = n % total_shards

    start = (shard_index - 1) * base_size + min(shard_index - 1, remainder)
    count = base_size + (1 if shard_index <= remainder else 0)
    return items[start : start + count]

validate_shard

validate_shard(shard_index: int, total_shards: int) -> None

Validate shard parameters.

Parameters:

Name	Type	Description	Default
shard_index	int	1-based shard index.	required
total_shards	int	Total number of shards.	required

Raises:

Type	Description
ShardError	If total_shards < 1 or shard_index is out of range.

Source code in behave_pool/shard.py

def validate_shard(shard_index: int, total_shards: int) -> None:
    """Validate shard parameters.

    Args:
        shard_index: 1-based shard index.
        total_shards: Total number of shards.

    Raises:
        ShardError: If ``total_shards`` < 1 or ``shard_index`` is out of range.
    """
    if total_shards < 1:
        msg = f"total_shards must be >= 1, got {total_shards}"
        raise ShardError(msg)
    if shard_index < 1:
        msg = f"shard_index must be >= 1, got {shard_index}"
        raise ShardError(msg)
    if shard_index > total_shards:
        msg = f"shard_index ({shard_index}) must be <= total_shards ({total_shards})"
        raise ShardError(msg)

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,
            )
            if unit.serial_mode != "all":
                from behave_pool.serial import filter_serial_scenarios

                for feature in self.features:
                    filter_serial_scenarios(feature, unit.serial_mode)
            if unit.scenario_line is not None:
                from behave_pool.serial import filter_scenario_by_line

                for feature in self.features:
                    filter_scenario_by_line(feature, unit.scenario_line)
            self.undefined_steps.clear()
            self.hook_failures = 0
            if self.context is not None:
                # aborted lives on the Context created in setup()
                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 ""),
            "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."""
        # NOTE: use scenario.steps, not all_steps — all_steps also contains
        # the background steps, which are serialized via feature.background
        # and would otherwise be counted twice.
        steps = []
        for step in getattr(scenario, "steps", None) 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}"
        # Walk run_items so ScenarioOutlines contribute their expanded
        # example scenarios (with real statuses), not the untested template
        # that appears in feature.scenarios.
        from behave.model import Rule, ScenarioOutline

        def _iter_scenarios(container: Any) -> Any:
            for item in getattr(container, "run_items", None) or []:
                if isinstance(item, Rule):
                    yield from _iter_scenarios(item)
                elif isinstance(item, ScenarioOutline):
                    yield from item.scenarios
                else:
                    yield item

        scenarios = [self._serialize_scenario(s, feature_id) for s in _iter_scenarios(feature)]
        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(
                    self._map_status(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,
        )
        if unit.serial_mode != "all":
            from behave_pool.serial import filter_serial_scenarios

            for feature in self.features:
                filter_serial_scenarios(feature, unit.serial_mode)
        if unit.scenario_line is not None:
            from behave_pool.serial import filter_scenario_by_line

            for feature in self.features:
                filter_scenario_by_line(feature, unit.scenario_line)
        self.undefined_steps.clear()
        self.hook_failures = 0
        if self.context is not None:
            # aborted lives on the Context created in setup()
            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)

Serial detection

Shared helpers used by both the coordinator (planning) and the workers (execution) to identify @serial scenarios consistently.

behave_pool.serial

Shared @serial tag detection and scenario filtering.

The @serial tag can appear at feature, rule, scenario, scenario-outline, or examples level. Both the coordinator (planning) and the workers (execution) use these helpers so that the two phases agree on which scenarios are serial.

filter_scenario_by_line

filter_scenario_by_line(feature: Feature, line: int) -> None

Keep only the run item declared at line, in place.

A run item is a Scenario or a whole ScenarioOutline (outlines keep all their expanded examples as one unit, since feature.scenarios exposes only the outline template anyway). Rules are traversed recursively and dropped when left empty.

Parameters:

Name	Type	Description	Default
feature	Feature	Parsed Behave Feature (freshly parsed by the worker).	required
line	int	Declaration line of the scenario/outline to keep.	required

Source code in behave_pool/serial.py

def filter_scenario_by_line(feature: Feature, line: int) -> None:
    """Keep only the run item declared at ``line``, in place.

    A run item is a Scenario or a whole ScenarioOutline (outlines keep
    all their expanded examples as one unit, since ``feature.scenarios``
    exposes only the outline template anyway).  Rules are traversed
    recursively and dropped when left empty.

    Args:
        feature: Parsed Behave Feature (freshly parsed by the worker).
        line: Declaration line of the scenario/outline to keep.
    """
    _filter_container_by_line(feature, line)

filter_serial_scenarios

filter_serial_scenarios(feature: Feature, serial_mode: str) -> None

Remove non-matching scenarios from a feature, in place.

Parameters:

Name	Type	Description	Default
feature	Feature	Parsed Behave Feature (freshly parsed by the worker).	required
serial_mode	str	"only" keeps serial scenarios, "exclude" drops them, "all" is a no-op.	required

Source code in behave_pool/serial.py

def filter_serial_scenarios(feature: Feature, serial_mode: str) -> None:
    """Remove non-matching scenarios from a feature, in place.

    Args:
        feature: Parsed Behave Feature (freshly parsed by the worker).
        serial_mode: ``"only"`` keeps serial scenarios, ``"exclude"`` drops
            them, ``"all"`` is a no-op.
    """
    if serial_mode == "all":
        return
    _filter_container(feature, serial_mode)

is_serial_tagged

is_serial_tagged(obj: Any) -> bool

Return True if serial is among the object's effective tags.

Effective tags include tags inherited from the parent feature or rule.

Source code in behave_pool/serial.py

def is_serial_tagged(obj: Any) -> bool:
    """Return True if ``serial`` is among the object's effective tags.

    Effective tags include tags inherited from the parent feature or rule.
    """
    tags = getattr(obj, "effective_tags", None)
    if tags is None:
        tags = getattr(obj, "tags", None) or []
    return SERIAL_TAG in {str(t) for t in tags}

item_is_serial

item_is_serial(item: Any) -> bool

Return True if a run item (Scenario/ScenarioOutline/Rule) is serial.

A rule or outline is serial when it is tagged @serial itself or when any of its contained scenarios are serial.

Source code in behave_pool/serial.py

def item_is_serial(item: Any) -> bool:
    """Return True if a run item (Scenario/ScenarioOutline/Rule) is serial.

    A rule or outline is serial when it is tagged ``@serial`` itself or when
    any of its contained scenarios are serial.
    """
    from behave.model import Rule, ScenarioOutline

    if isinstance(item, Rule):
        return is_serial_tagged(item) or any(item_is_serial(i) for i in item.run_items)
    if isinstance(item, ScenarioOutline):
        return is_serial_tagged(item) or any(is_serial_tagged(s) for s in item.scenarios)
    return is_serial_tagged(item)

CLI

The behave-pool console entry point — a behave wrapper that registers the pool options before argument parsing.

behave_pool.cli

behave-pool command: behave wrapper that registers pool options early.

Behave parses command-line arguments and config files inside Configuration.__init__, before the runner class is loaded. Custom options therefore must be registered in behave.configuration.OPTIONS before Configuration is created. Importing :mod:behave_pool registers them, so this entry point simply imports the package first and then delegates to behave's own main() flow.

Unless --runner (or the runner config key) selects a different runner, the behave-pool command uses :class:ParallelRunner automatically — no [behave.runners] alias is required.

main

main(args: list[str] | None = None) -> int

Entry point for the behave-pool command.

Parameters:

Name	Type	Description	Default
args	list[str] | None	Command-line arguments (defaults to sys.argv[1:]).	None

Returns:

Type	Description
int	Exit code: 0 on success, non-zero on failure.

Source code in behave_pool/cli.py

def main(args: list[str] | None = None) -> int:
    """Entry point for the ``behave-pool`` command.

    Args:
        args: Command-line arguments (defaults to ``sys.argv[1:]``).

    Returns:
        Exit code: 0 on success, non-zero on failure.
    """
    # behave_pool is imported at module level so that its options are
    # registered in behave.configuration.OPTIONS before Configuration
    # parses the command line.
    from behave.__main__ import run_behave
    from behave.configuration import DEFAULT_RUNNER_CLASS_NAME, Configuration
    from behave.exception import ConfigError, TagExpressionError

    try:
        # NOTE: behave only falls back to sys.argv when argv[0] contains
        # "behave"; under "python -m behave_pool" it does not, so the args
        # must be passed explicitly.
        args_list = list(args) if args is not None else sys.argv[1:]
        config = Configuration(args_list)
        runner_class = None
        user_runner_given = any(
            a == "--runner" or a.startswith("--runner=") or a.startswith("--runner:")
            for a in args_list
        )
        if config.runner == DEFAULT_RUNNER_CLASS_NAME and not user_runner_given:
            runner_class = behave_pool.ParallelRunner
        return int(run_behave(config, runner_class=runner_class))
    except ConfigError as e:
        print(f"{e.__class__.__name__}: {e}")
    except TagExpressionError as e:
        print(f"TagExpressionError: {e}")
    return 1