diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index cd24b22bc..fdbc9c504 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -934,6 +934,15 @@ def _install_workflow_from_catalog( workflow_dir = _safe_workflow_id_dir(workflows_dir, workflow_id) workflow_file = workflow_dir / "workflow.yml" + # Captured before any mkdir/download writes so a registry.add() failure + # at the end of this function can tell a fresh install from a + # reinstall-over-an-existing-one, mirroring _validate_and_install_local's + # existed-before/backup-aware rollback. + existed_before = workflow_dir.is_dir() + prior_workflow_bytes = ( + workflow_file.read_bytes() if existed_before and workflow_file.is_file() else None + ) + try: from specify_cli.authentication.http import open_url as _open_url from specify_cli.authentication.http import github_provider_hosts as _github_provider_hosts @@ -1047,8 +1056,15 @@ def _install_workflow_from_catalog( try: registry.add(workflow_id, entry) except OSError as exc: - import shutil - shutil.rmtree(workflow_dir, ignore_errors=True) + # Don't destroy a prior working install on a reinstall: only a + # brand-new directory is safe to remove wholesale; an existing one + # gets its previous workflow.yml restored instead. + if existed_before: + if prior_workflow_bytes is not None: + workflow_file.write_bytes(prior_workflow_bytes) + else: + import shutil + shutil.rmtree(workflow_dir, ignore_errors=True) console.print( f"[red]Error:[/red] Failed to update workflow registry for " f"'{_escape_markup(workflow_id)}': {_escape_markup(str(exc))}" diff --git a/tests/test_workflows.py b/tests/test_workflows.py index eb59c50d4..cf76d49c0 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -7782,6 +7782,71 @@ steps: assert not dest_dir.exists() assert not WorkflowRegistry(project_dir).is_installed("align-wf") + def test_add_catalog_reinstall_save_failure_restores_prior_file(self, project_dir, monkeypatch): + """Re-adding an already-installed catalog workflow downloads the new + version over the existing install directory. If registry.add() then + fails to save, the prior working workflow.yml must be restored + byte-for-byte (not left overwritten with the new download, and not + deleted like a fresh install) and the registry must remain valid and + still point at the original version -- the update path's caller has + an outer backup/restore for this, but plain `workflow add` does not, + so _install_workflow_from_catalog must handle it itself.""" + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry + + monkeypatch.chdir(project_dir) + monkeypatch.setattr( + WorkflowCatalog, + "get_workflow_info", + lambda self, wid: { + "id": wid, + "name": "Align Workflow", + "version": "1.0.0", + "url": "https://example.com/workflow.yml", + "_install_allowed": True, + "_catalog_name": "test-catalog", + }, + ) + original_data = self.WORKFLOW_YAML.format(version="1.0.0").encode() + runner = CliRunner() + with pytest.MonkeyPatch.context() as mp: + mp.setattr( + "specify_cli.authentication.http.open_url", + lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse( + original_data, url + ), + ) + result = runner.invoke(app, ["workflow", "add", "align-wf"]) + assert result.exit_code == 0, result.output + + dest_file = project_dir / ".specify" / "workflows" / "align-wf" / "workflow.yml" + assert dest_file.read_bytes() == original_data + + new_data = self.WORKFLOW_YAML.format(version="2.0.0").encode() + + def boom(self): + raise OSError("disk full") + + with pytest.MonkeyPatch.context() as mp: + mp.setattr( + "specify_cli.authentication.http.open_url", + lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse( + new_data, url + ), + ) + mp.setattr(WorkflowRegistry, "save", boom) + result = runner.invoke(app, ["workflow", "add", "align-wf"]) + + assert result.exit_code != 0 + assert result.exception is None or isinstance(result.exception, SystemExit) + assert result.output.strip() != "" + # The prior working install must survive untouched, byte-for-byte. + assert dest_file.read_bytes() == original_data + registry = WorkflowRegistry(project_dir) + assert registry.is_installed("align-wf") + assert registry.get("align-wf")["version"] == "1.0.0" + def test_download_redirect_validator_rejects_http_before_follow(self): import urllib.error