Method _build

tests/test_workflows.py:2107–2139

Wire an engine + run state to a probe step that echoes context.item. Per-item output is ``{"seen": }`` so order and per-thread item isolation are checkable. ``on_item(item)`` may run a side effect and optionally return a StepStatus to override COMPLETED (or raise).

(tmp_path, on_item=None)

    @staticmethod
    def _build(tmp_path, on_item=None):
        """Wire an engine + run state to a probe step that echoes context.item.

        Per-item output is ``{"seen": <item>}`` so order and per-thread item
        isolation are checkable. ``on_item(item)`` may run a side effect and
        optionally return a StepStatus to override COMPLETED (or raise).
        """
        from specify_cli.workflows.base import (
            RunStatus,
            StepBase,
            StepContext,
            StepResult,
            StepStatus,
        )
        from specify_cli.workflows.engine import RunState, WorkflowEngine

        class _ProbeStep(StepBase):
            type_key = "probe"

            def execute(self, config, context):
                status = StepStatus.COMPLETED
                if on_item is not None:
                    override = on_item(context.item)
                    if override is not None:
                        status = override
                return StepResult(status=status, output={"seen": context.item})

        engine = WorkflowEngine(project_root=tmp_path)
        context = StepContext()
        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
        state.status = RunStatus.RUNNING
        template = {"id": "impl", "type": "probe"}
        return engine, context, state, {"probe": _ProbeStep()}, template

    def _run(self, tmp_path, items, max_concurrency, on_item=None):
        engine, context, state, registry, template = self._build(tmp_path, on_item)