diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index f275bfe09a..55bc7b0b33 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -3408,6 +3408,26 @@ def _safe_fetch(url: str) -> bytes: ) raise typer.Exit(1) + catalog_version = info.get("version") + downloaded_version = step_meta.get("version") + if "version" in info and "version" in step_meta: + from packaging import version as pkg_version + + try: + versions_match = pkg_version.Version( + str(downloaded_version) + ) == pkg_version.Version(str(catalog_version)) + except pkg_version.InvalidVersion: + versions_match = str(downloaded_version) == str(catalog_version) + if not versions_match: + console.print( + f"[red]Error:[/red] step.yml version " + f"({_escape_markup(repr(downloaded_version))}) does not match " + f"the catalog version ({_escape_markup(repr(catalog_version))}). " + "The catalog entry may be stale or misconfigured." + ) + raise typer.Exit(1) + # Write the two required files. try: (tmp_path / "step.yml").write_bytes(step_yml_content) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 980cc21a6e..a70877b811 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -10577,6 +10577,93 @@ def test_list_escapes_installed_metadata( class TestWorkflowStepAddCLI: + @pytest.mark.parametrize( + ("catalog_version", "downloaded_version"), + [ + ("1.0.0", "2.0.0"), + ("1.0.0", 0), + ("1.0.0", False), + ("1.0.0", ""), + ("1.0.0", None), + ("release-a", " release-a "), + ], + ) + def test_add_rejects_step_yml_version_mismatch( + self, project_dir, monkeypatch, catalog_version, downloaded_version + ): + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + from specify_cli.workflows.catalog import StepCatalog, StepRegistry + + monkeypatch.chdir(project_dir) + monkeypatch.setattr( + StepCatalog, + "get_step_info", + lambda self, step_id: { + "id": step_id, + "name": "Test Step", + "version": catalog_version, + "url": "https://example.com/step.yml", + "init_url": "https://example.com/__init__.py", + "_install_allowed": True, + }, + ) + bodies = { + "https://example.com/step.yml": yaml.safe_dump( + { + "step": { + "type_key": "my-step", + "version": downloaded_version, + } + } + ).encode(), + "https://example.com/__init__.py": b"# custom step\n", + } + + class _FakeResponse: + def __init__(self, url): + self.url = url + self.body = bodies[url] + self.offset = 0 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def getheader(self, name): + return None + + def geturl(self): + return self.url + + def read(self, size=-1): + if size < 0: + size = len(self.body) - self.offset + chunk = self.body[self.offset : self.offset + size] + self.offset += len(chunk) + return chunk + + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None: _FakeResponse(url), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step"] + ) + + assert result.exit_code != 0 + assert "does not match the catalog version" in result.output + assert not StepRegistry(project_dir).is_installed("my-step") + assert not ( + project_dir / ".specify" / "workflows" / "steps" / "my-step" + ).exists() + @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") def test_add_rejects_symlinked_steps_base_dir(self, project_dir, monkeypatch): from typer.testing import CliRunner