mirror of
https://github.com/github/spec-kit.git
synced 2026-08-03 06:26:30 +08:00
feat(workflows): expose workflow source directory to steps (#3469)
* feat(workflows): expose workflow source directory to steps (#3467) Propagate WorkflowDefinition.source_path to steps via {{ context.workflow_dir }} in template expressions and SPECKIT_WORKFLOW_DIR env var for shell steps. The original source directory is persisted in state.json so resume restores the correct value instead of the run-directory copy path. Closes #3467 Assisted-By: 🤖 Claude Code * fix: apply bot review suggestions (#2) Applied fixes from bot review comments: - Comment #3563319058: prevent stale SPECKIT_WORKFLOW_DIR leak from parent env - Comment #3563319094: use cross-platform Python one-liner instead of printenv - Comment #3563319103: add monkeypatch.delenv for deterministic env var test - Comment #3563319116: same env leak fix as #3563319058 Assisted-By: 🤖 Claude Code * fix: use YAML single-quotes and forward-slash paths for Windows CI (#2) sys.executable on Windows returns backslash paths (D:\a\...) which YAML double-quoted strings interpret as escape sequences. Switch to single-quoted YAML strings and normalize paths with replace("\\", "/"). Assisted-By: 🤖 Claude Code * fix: resolve workflow_dir to absolute path and add installed-by-ID test (#3469) Applied fixes from bot review comments: - Comment #3563382853: resolve source_path before taking parent to ensure absolute paths - Comment #3563382864: add test for installed-by-ID workflow_dir semantics Assisted-By: 🤖 Claude Code * docs: document context.workflow_dir and SPECKIT_WORKFLOW_DIR Add reference documentation for the new workflow_dir runtime value in both workflows/README.md and docs/reference/workflows.md so workflow authors can discover the feature and its semantics. Assisted-By: 🤖 Claude Code * fix: clarify installed workflow_dir is an absolute path (#3469) The documentation for context.workflow_dir described the installed-by-ID case as ".specify/workflows/<id>/" which appears relative, contradicting the "resolved absolute path" semantics. Clarified that it is the absolute path to the installation directory. Assisted-By: 🤖 Claude Code * fix: apply bot review suggestions (#3469) Applied fixes from bot review comments: - Comment #3580005128: Quote sys.executable in shell step env var test - Comment #3580005174: Quote sys.executable in no-env-var test Assisted-By: 🤖 Claude Code * fix: apply bot review suggestions (#3469) Applied fixes from bot review comments: - Comment #3587146944: Quote interpolated workflow_dir path in example Assisted-By: 🤖 Claude Code
This commit is contained in:
Roland Huss
@@ -4516,6 +4516,286 @@ steps:
|
||||
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/<id>/).
|
||||
"""
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user