diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 062b85116..5b48aed0a 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -789,11 +789,14 @@ def _install_workflow_from_catalog( registry: Any, workflows_dir: Path, workflow_id: str, + expected_version: str | None = None, ) -> None: """Download, validate, and register a catalog workflow. Shared by ``workflow add`` and ``workflow update``. Raises ``typer.Exit`` on any failure; the registry entry is only written on full success. + ``expected_version``, when given, rejects a downloaded workflow whose + version does not match the catalog version that triggered the install. """ from .catalog import WorkflowCatalog, WorkflowCatalogError from .engine import WorkflowDefinition @@ -928,6 +931,25 @@ def _install_workflow_from_catalog( ) raise typer.Exit(1) + # A stale or misconfigured URL can serve a different version than the + # catalog advertised; without this check `update` would report success + # while leaving the old version installed (or even downgrading). + if expected_version is not None: + from packaging import version as pkg_version + try: + version_matches = pkg_version.Version(str(definition.version)) == pkg_version.Version(expected_version) + except pkg_version.InvalidVersion: + version_matches = str(definition.version) == expected_version + if not version_matches: + import shutil + shutil.rmtree(workflow_dir, ignore_errors=True) + console.print( + f"[red]Error:[/red] Downloaded workflow version ({_escape_markup(str(definition.version))}) " + f"does not match the catalog version ({_escape_markup(expected_version)}). " + f"The catalog entry may be stale or misconfigured." + ) + raise typer.Exit(1) + entry = { "name": definition.name or info.get("name", workflow_id), "version": definition.version or info.get("version", "0.0.0"), @@ -1109,7 +1131,10 @@ def workflow_update( wf_dir = _safe_workflow_id_dir(workflows_dir, update["id"]) wf_file = wf_dir / "workflow.yml" backup = wf_file.read_bytes() if wf_file.is_file() else None - _install_workflow_from_catalog(project_root, registry, workflows_dir, update["id"]) + _install_workflow_from_catalog( + project_root, registry, workflows_dir, update["id"], + expected_version=update["available"], + ) except (typer.Exit, OSError) as exc: if backup is not None and wf_dir is not None and wf_file is not None: try: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 96c2449d5..0725db8d9 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -7660,6 +7660,54 @@ steps: # The previously installed workflow must survive. assert "1.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8") + def test_update_rejects_version_mismatch_from_stale_url(self, project_dir, monkeypatch): + """A URL serving a different version than the catalog advertised must fail the update.""" + from unittest.mock import patch + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry + + monkeypatch.chdir(project_dir) + WorkflowRegistry(project_dir).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", + }, + ) + # The URL still serves the old 1.0.0 payload. + data = self.WORKFLOW_YAML.format(version="1.0.0").encode() + runner = CliRunner() + with patch( + "specify_cli.authentication.http.open_url", + side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url), + ): + result = runner.invoke(app, ["workflow", "update"], input="y\n") + assert "does not match the catalog version" in result.output + assert "Failed to update" in result.output + meta = WorkflowRegistry(project_dir).get("align-wf") + assert meta["version"] == "1.0.0" + assert "1.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8") + def test_update_preserves_disabled_state(self, project_dir, monkeypatch): from unittest.mock import patch from typer.testing import CliRunner