From e7bba7afa27a46e0f09d42fff31cf248bd9c1b26 Mon Sep 17 00:00:00 2001 From: WOLIKIMCHENG <35391914+WOLIKIMCHENG@users.noreply.github.com> Date: Mon, 28 Sep 2026 11:15:46 +0800 Subject: [PATCH] fix(workflows): load custom steps before update --- src/specify_cli/workflows/command_update.py | 2 + .../workflows/test_command_update.py | 170 ++++++++++++++++++ 2 files changed, 172 insertions(+) diff --git a/src/specify_cli/workflows/command_update.py b/src/specify_cli/workflows/command_update.py index 84ecc530e3..e95ef3c151 100644 --- a/src/specify_cli/workflows/command_update.py +++ b/src/specify_cli/workflows/command_update.py @@ -14,9 +14,11 @@ def workflow_update( """Update installed workflow(s) to the latest catalog version.""" from packaging import version as pkg_version + from . import load_custom_steps from .catalog import WorkflowCatalog, WorkflowCatalogError project_root = cli._require_specify_project() + load_custom_steps(project_root) registry = cli._open_workflow_registry(project_root) workflows_dir = project_root / ".specify" / "workflows" cli._reject_unsafe_dir(project_root / ".specify", ".specify") diff --git a/tests/specify_cli/workflows/test_command_update.py b/tests/specify_cli/workflows/test_command_update.py index b0f2d29e65..2ed38d82bd 100644 --- a/tests/specify_cli/workflows/test_command_update.py +++ b/tests/specify_cli/workflows/test_command_update.py @@ -5,6 +5,7 @@ import json import os import stat +import sys from pathlib import Path import pytest @@ -41,6 +42,49 @@ def _install_dev(self, runner, app, project_dir): assert result.exit_code == 0, result.output return src + @staticmethod + def _write_custom_step(project_dir, type_key): + step_dir = project_dir / ".specify" / "workflows" / "steps" / type_key + step_dir.mkdir(parents=True) + (step_dir / "step.yml").write_text( + "\n".join( + ( + 'schema_version: "1.0"', + "step:", + f' type_key: "{type_key}"', + ' name: "Test Update Step"', + ' version: "1.0.0"', + "", + ) + ), + encoding="utf-8", + ) + (step_dir / "__init__.py").write_text( + "\n".join( + ( + "from specify_cli.workflows.base import StepBase, StepResult", + "", + "", + "class TestUpdateStep(StepBase):", + f' type_key = "{type_key}"', + "", + " def execute(self, config, context):", + " return StepResult()", + "", + ) + ), + encoding="utf-8", + ) + + @staticmethod + def _custom_step_module_name(type_key): + import hashlib + import re + + safe_key = re.sub(r"[^A-Za-z0-9_]", "_", type_key) + key_hash = hashlib.sha256(type_key.encode()).hexdigest()[:8] + return f"_speckit_custom_step_{safe_key}_{key_hash}" + class _FakeResponse: def __init__(self, data, url="https://example.com/workflow.yml", headers=None): self._data = data @@ -202,6 +246,132 @@ def test_update_installs_newer_catalog_version(self, project_dir, monkeypatch): assert meta["version"] == "2.0.0" assert "2.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8") + def test_update_installs_newer_workflow_with_project_custom_step( + self, project_dir, monkeypatch + ): + """Updates validate against custom steps installed by this project.""" + from unittest.mock import patch + + from typer.testing import CliRunner + from specify_cli import app + import specify_cli.workflows as workflows + from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry + + custom_type = "test-update-step" + module_name = self._custom_step_module_name(custom_type) + self._write_custom_step(project_dir, custom_type) + monkeypatch.chdir(project_dir) + monkeypatch.setattr(workflows, "STEP_REGISTRY", dict(workflows.STEP_REGISTRY)) + try: + registry = WorkflowRegistry(project_dir) + registry.add("align-wf", { + "name": "Align Workflow", + "version": "1.0.0", + "description": "CLI alignment test workflow", + "source": "catalog", + "catalog_name": "test-catalog", + "url": "https://example.com/workflow.yml", + }) + wf_dir = project_dir / ".specify" / "workflows" / "align-wf" + wf_dir.mkdir(parents=True) + (wf_dir / "workflow.yml").write_text( + self.WORKFLOW_YAML.format(version="1.0.0"), encoding="utf-8" + ) + monkeypatch.setattr( + WorkflowCatalog, + "get_workflow_info", + lambda self, wid: { + "id": wid, + "name": "Align Workflow", + "version": "2.0.0", + "url": "https://example.com/workflow.yml", + "_install_allowed": True, + "_catalog_name": "test-catalog", + }, + ) + data = self.WORKFLOW_YAML.format(version="2.0.0").replace( + "type: shell", f"type: {custom_type}" + ).encode() + with patch( + "specify_cli.authentication.http.open_url", + side_effect=lambda url, timeout=None, extra_headers=None, + redirect_validator=None: self._FakeResponse(data, url), + ): + result = CliRunner().invoke(app, ["workflow", "update"], input="y\n") + + assert result.exit_code == 0, result.output + assert custom_type in workflows.STEP_REGISTRY + assert WorkflowRegistry(project_dir).get("align-wf")["version"] == "2.0.0" + assert (wf_dir / "workflow.yml").read_bytes() == data + finally: + sys.modules.pop(module_name, None) + + def test_update_rejects_unknown_step_without_changing_installed_workflow( + self, project_dir, monkeypatch + ): + """Loading a project custom step must not admit unrelated types.""" + from unittest.mock import patch + + from typer.testing import CliRunner + from specify_cli import app + import specify_cli.workflows as workflows + from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry + + custom_type = "test-update-step" + module_name = self._custom_step_module_name(custom_type) + self._write_custom_step(project_dir, custom_type) + monkeypatch.chdir(project_dir) + monkeypatch.setattr(workflows, "STEP_REGISTRY", dict(workflows.STEP_REGISTRY)) + try: + registry = WorkflowRegistry(project_dir) + registry.add("align-wf", { + "name": "Align Workflow", + "version": "1.0.0", + "description": "CLI alignment test workflow", + "source": "catalog", + "catalog_name": "test-catalog", + "url": "https://example.com/workflow.yml", + }) + wf_dir = project_dir / ".specify" / "workflows" / "align-wf" + wf_dir.mkdir(parents=True) + workflow_file = wf_dir / "workflow.yml" + original_workflow = self.WORKFLOW_YAML.format(version="1.0.0").encode() + workflow_file.write_bytes(original_workflow) + original_metadata = dict(WorkflowRegistry(project_dir).get("align-wf")) + registry_file = registry.registry_path + original_registry = registry_file.read_bytes() + + monkeypatch.setattr( + WorkflowCatalog, + "get_workflow_info", + lambda self, wid: { + "id": wid, + "name": "Align Workflow", + "version": "2.0.0", + "url": "https://example.com/workflow.yml", + "_install_allowed": True, + "_catalog_name": "test-catalog", + }, + ) + data = self.WORKFLOW_YAML.format(version="2.0.0").replace( + "type: shell", "type: unknown-update-step" + ).encode() + with patch( + "specify_cli.authentication.http.open_url", + side_effect=lambda url, timeout=None, extra_headers=None, + redirect_validator=None: self._FakeResponse(data, url), + ): + result = CliRunner().invoke(app, ["workflow", "update"], input="y\n") + + assert result.exit_code != 0 + assert "invalid type 'unknown-update-step'" in result.output + assert custom_type in workflows.STEP_REGISTRY + assert workflow_file.read_bytes() == original_workflow + assert registry_file.read_bytes() == original_registry + assert WorkflowRegistry(project_dir).get("align-wf") == original_metadata + finally: + sys.modules.pop(module_name, None) + def test_update_downloaded_invalid_yaml_escapes_rich_markup(self, project_dir, monkeypatch): """A malformed downloaded workflow can quote the offending line verbatim; escape it before printing.""" from unittest.mock import patch