diff --git a/src/specify_cli/workflows/base.py b/src/specify_cli/workflows/base.py index 3a42679309..6ad14cc257 100644 --- a/src/specify_cli/workflows/base.py +++ b/src/specify_cli/workflows/base.py @@ -74,6 +74,9 @@ class StepContext: #: Current run ID. run_id: str | None = None + #: Source directory of the workflow definition file. + workflow_dir: str | None = None + @dataclass class StepResult: diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 6025c30c5c..64900a1adf 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -455,6 +455,7 @@ def __init__( # append_log is never called while _lock is held, the two never nest. self._log_lock = threading.Lock() self.inputs: dict[str, Any] = {} + self.workflow_dir: str | None = None self.created_at = datetime.now(timezone.utc).isoformat() self.updated_at = self.created_at self.log_entries: list[dict[str, Any]] = [] @@ -507,6 +508,7 @@ def save(self) -> None: "current_step_index": self.current_step_index, "current_step_id": self.current_step_id, "step_results": self.step_results, + "workflow_dir": self.workflow_dir, "created_at": self.created_at, "updated_at": self.updated_at, } @@ -564,6 +566,7 @@ def load(cls, run_id: str, project_root: Path) -> RunState: state.current_step_index = state_data.get("current_step_index", 0) state.current_step_id = state_data.get("current_step_id") state.step_results = state_data.get("step_results", {}) + state.workflow_dir = state_data.get("workflow_dir") state.created_at = state_data.get("created_at", "") state.updated_at = state_data.get("updated_at", "") @@ -697,6 +700,12 @@ def execute( # Resolve inputs resolved_inputs = self._resolve_inputs(definition, inputs or {}) state.inputs = resolved_inputs + workflow_dir = ( + str(definition.source_path.resolve().parent) + if definition.source_path is not None + else None + ) + state.workflow_dir = workflow_dir state.status = RunStatus.RUNNING state.save() @@ -707,6 +716,7 @@ def execute( default_options=definition.default_options, project_root=str(self.project_root), run_id=state.run_id, + workflow_dir=workflow_dir, ) # Execute steps @@ -772,6 +782,7 @@ def resume( default_options=definition.default_options, project_root=str(self.project_root), run_id=state.run_id, + workflow_dir=state.workflow_dir, ) from . import STEP_REGISTRY diff --git a/src/specify_cli/workflows/expressions.py b/src/specify_cli/workflows/expressions.py index d6736bc321..e2dc225cd8 100644 --- a/src/specify_cli/workflows/expressions.py +++ b/src/specify_cli/workflows/expressions.py @@ -142,7 +142,8 @@ def _build_namespace(context: Any) -> dict[str, Any]: # runs use an 8-character uuid4 hex; operator-supplied ids may be # any alphanumeric string with hyphens or underscores. run_id = getattr(context, "run_id", None) or "" - ns["context"] = {"run_id": run_id} + workflow_dir = getattr(context, "workflow_dir", None) or "" + ns["context"] = {"run_id": run_id, "workflow_dir": workflow_dir} return ns diff --git a/src/specify_cli/workflows/steps/shell/__init__.py b/src/specify_cli/workflows/steps/shell/__init__.py index ed5df8fc36..68b273c7f3 100644 --- a/src/specify_cli/workflows/steps/shell/__init__.py +++ b/src/specify_cli/workflows/steps/shell/__init__.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os import subprocess from typing import Any @@ -34,6 +35,12 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: if isinstance(timeout, bool) or not isinstance(timeout, int) or timeout <= 0: timeout = 300 + env = {**os.environ} + if context.workflow_dir: + env["SPECKIT_WORKFLOW_DIR"] = context.workflow_dir + else: + env.pop("SPECKIT_WORKFLOW_DIR", None) + # NOTE: shell=True is required to support pipes, redirects, and # multi-command expressions in workflow YAML. Workflow authors # control commands; catalog-installed workflows should be reviewed @@ -45,6 +52,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: capture_output=True, text=True, cwd=cwd, + env=env, timeout=timeout, ) output = { diff --git a/tests/test_workflows.py b/tests/test_workflows.py index d7cff20f6d..de6da6d851 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -4163,6 +4163,286 @@ def test_run_id_arg_takes_precedence_over_env_override(self, project_dir, monkey assert state.step_results["stamp"]["output"]["stdout"].strip() == "explicit-456" +# ===== context.workflow_dir Tests ===== + + +class TestContextWorkflowDir: + """Tests for `{{ context.workflow_dir }}` and `SPECKIT_WORKFLOW_DIR`.""" + + def test_context_workflow_dir_resolves(self): + """``{{ context.workflow_dir }}`` resolves to ``StepContext.workflow_dir``.""" + from specify_cli.workflows.expressions import evaluate_expression + from specify_cli.workflows.base import StepContext + + ctx = StepContext(workflow_dir="/home/user/my-workflow") + assert evaluate_expression("{{ context.workflow_dir }}", ctx) == "/home/user/my-workflow" + + def test_context_workflow_dir_defaults_to_empty_when_unset(self): + """``{{ context.workflow_dir }}`` resolves to ``""`` when no source + path is available (string-loaded workflows, dry-run). + """ + from specify_cli.workflows.expressions import evaluate_expression + from specify_cli.workflows.base import StepContext + + ctx = StepContext() + assert evaluate_expression("{{ context.workflow_dir }}", ctx) == "" + + def test_context_workflow_dir_string_interpolation(self): + """Workflow dir interpolates inside a larger template string.""" + from specify_cli.workflows.expressions import evaluate_expression + from specify_cli.workflows.base import StepContext + + ctx = StepContext(workflow_dir="/opt/workflows/setup") + result = evaluate_expression("cp {{ context.workflow_dir }}/config.yml .", ctx) + assert result == "cp /opt/workflows/setup/config.yml ." + + def test_step_context_workflow_dir(self): + """StepContext accepts and stores workflow_dir.""" + from specify_cli.workflows.base import StepContext + + ctx = StepContext(workflow_dir="/some/path") + assert ctx.workflow_dir == "/some/path" + + ctx_none = StepContext() + assert ctx_none.workflow_dir is None + + def test_from_yaml_sets_workflow_dir(self, project_dir): + """Workflow loaded from a YAML file has workflow_dir set to the + file's parent directory. + """ + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + + wf_dir = project_dir / "my-workflows" + wf_dir.mkdir() + wf_file = wf_dir / "setup.yml" + wf_file.write_text(""" +schema_version: "1.0" +workflow: + id: "from-yaml" + name: "From YAML" + version: "1.0.0" +steps: + - id: check-dir + type: shell + run: "echo DIR={{ context.workflow_dir }}" +""") + definition = WorkflowDefinition.from_yaml(wf_file) + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + + stdout = state.step_results["check-dir"]["output"]["stdout"] + assert stdout.strip() == f"DIR={wf_dir.resolve()}" + + def test_from_string_has_empty_workflow_dir(self, project_dir): + """String-loaded workflows have empty workflow_dir.""" + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + + definition = WorkflowDefinition.from_string(""" +schema_version: "1.0" +workflow: + id: "from-string" + name: "From String" + version: "1.0.0" +steps: + - id: check-dir + type: shell + run: "echo DIR={{ context.workflow_dir }}" +""") + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + + stdout = state.step_results["check-dir"]["output"]["stdout"] + assert stdout.strip() == "DIR=" + + def test_shell_step_receives_speckit_workflow_dir_env_var(self, project_dir): + """Shell steps receive SPECKIT_WORKFLOW_DIR in their environment.""" + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + import sys + + wf_dir = project_dir / "wf" + wf_dir.mkdir() + wf_file = wf_dir / "workflow.yml" + python = sys.executable.replace("\\", "/") + wf_file.write_text(f""" +schema_version: "1.0" +workflow: + id: "env-var-test" + name: "Env Var Test" + version: "1.0.0" +steps: + - id: print-env + type: shell + run: '{python} -c "import os; print(os.environ.get(''SPECKIT_WORKFLOW_DIR'', ''UNSET''))"' +""") + definition = WorkflowDefinition.from_yaml(wf_file) + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + + stdout = state.step_results["print-env"]["output"]["stdout"] + assert stdout.strip() == str(wf_dir.resolve()) + + def test_shell_step_no_env_var_when_workflow_dir_unset(self, project_dir, monkeypatch): + """Shell steps do not set SPECKIT_WORKFLOW_DIR for string-loaded workflows.""" + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + import sys + + monkeypatch.delenv("SPECKIT_WORKFLOW_DIR", raising=False) + + python = sys.executable.replace("\\", "/") + definition = WorkflowDefinition.from_string(f""" +schema_version: "1.0" +workflow: + id: "no-env-var" + name: "No Env Var" + version: "1.0.0" +steps: + - id: check-env + type: shell + run: '{python} -c "import os; print(os.environ.get(''SPECKIT_WORKFLOW_DIR'', ''UNSET''))"' +""") + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + + stdout = state.step_results["check-env"]["output"]["stdout"] + assert stdout.strip() == "UNSET" + + def test_resume_preserves_original_workflow_dir(self, project_dir): + """Resumed workflow uses the original source directory, not the + run-directory copy path. + """ + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + from specify_cli.workflows.base import RunStatus + + wf_dir = project_dir / "original-source" + wf_dir.mkdir() + wf_file = wf_dir / "resumable.yml" + wf_file.write_text(""" +schema_version: "1.0" +workflow: + id: "resumable" + name: "Resumable" + version: "1.0.0" +steps: + - id: gate-step + type: gate + message: "Approve?" + - id: after-gate + type: shell + run: "echo DIR={{ context.workflow_dir }}" +""") + definition = WorkflowDefinition.from_yaml(wf_file) + engine = WorkflowEngine(project_dir) + + # Execute -- gate pauses the workflow + state = engine.execute(definition) + assert state.status == RunStatus.PAUSED + assert state.workflow_dir == str(wf_dir.resolve()) + + # Simulate gate approval by patching the gate step + from unittest.mock import patch + from specify_cli.workflows.base import StepResult + + with patch( + "specify_cli.workflows.steps.gate.GateStep.execute", + return_value=StepResult(output={"approved": True}), + ): + state = engine.resume(state.run_id) + + assert state.status == RunStatus.COMPLETED + stdout = state.step_results["after-gate"]["output"]["stdout"] + assert stdout.strip() == f"DIR={wf_dir.resolve()}" + + def test_workflow_dir_persisted_in_state(self, project_dir): + """workflow_dir is persisted in state.json and survives load/save.""" + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine, RunState + + wf_dir = project_dir / "persist-test" + wf_dir.mkdir() + wf_file = wf_dir / "workflow.yml" + wf_file.write_text(""" +schema_version: "1.0" +workflow: + id: "persist-wfdir" + name: "Persist WfDir" + version: "1.0.0" +steps: + - id: noop + type: shell + run: "echo ok" +""") + definition = WorkflowDefinition.from_yaml(wf_file) + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + + # Reload state from disk and verify workflow_dir survived + loaded = RunState.load(state.run_id, project_dir) + assert loaded.workflow_dir == str(wf_dir.resolve()) + + def test_installed_workflow_has_workflow_dir(self, project_dir): + """Installed-by-ID workflows get workflow_dir pointing to the + installation directory (.specify/workflows//). + """ + from specify_cli.workflows.engine import WorkflowEngine + from specify_cli.workflows.base import RunStatus + + wf_id = "installed-wfdir" + install_dir = project_dir / ".specify" / "workflows" / wf_id + install_dir.mkdir(parents=True) + (install_dir / "workflow.yml").write_text(""" +schema_version: "1.0" +workflow: + id: "installed-wfdir" + name: "Installed WfDir" + version: "1.0.0" +steps: + - id: check-dir + type: shell + run: "echo DIR={{ context.workflow_dir }}" +""") + engine = WorkflowEngine(project_dir) + definition = engine.load_workflow(wf_id) + state = engine.execute(definition) + + assert state.status == RunStatus.COMPLETED + stdout = state.step_results["check-dir"]["output"]["stdout"] + assert stdout.strip() == f"DIR={install_dir.resolve()}" + + def test_workflow_dir_is_resolved_to_absolute(self, project_dir): + """workflow_dir is resolved to an absolute path even when the + source path is relative. + """ + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + import os + + wf_dir = project_dir / "rel-test" + wf_dir.mkdir() + wf_file = wf_dir / "workflow.yml" + wf_file.write_text(""" +schema_version: "1.0" +workflow: + id: "rel-path" + name: "Relative Path" + version: "1.0.0" +steps: + - id: check + type: shell + run: "echo ok" +""") + # Load via a relative path + saved_cwd = os.getcwd() + try: + os.chdir(project_dir) + rel_path = Path("rel-test/workflow.yml") + definition = WorkflowDefinition.from_yaml(rel_path) + engine = WorkflowEngine(project_dir) + state = engine.execute(definition) + finally: + os.chdir(saved_cwd) + + assert Path(state.workflow_dir).is_absolute() + assert state.workflow_dir == str(wf_dir.resolve()) + + # ===== continue_on_error Tests ===== # # Locks the contract documented in workflows/README.md "Error Handling"