fix(workflow): preserve prior catalog install on reinstall registry-save failure

_install_workflow_from_catalog's final registry.add() failure handler
unconditionally rmtree'd workflow_dir. That's safe for a brand-new
install, but plain `workflow add <catalog-id>` also allows re-adding an
already-installed workflow, downloading the new version over the
existing directory first. If registry.add() then failed to save, the
unconditional rmtree deleted the prior working install while the
registry (after its own rollback) still reported it installed -- data
loss with no way back. workflow_update already avoids this via an outer
backup/restore around this function, but plain add has no such caller.

Fix mirrors _validate_and_install_local's existed-before/backup-aware
handling: capture whether workflow_dir existed and back up its
workflow.yml bytes before any download write, then on a registry.add()
OSError, restore those bytes for a reinstall or rmtree only a
brand-new directory. Only one file (workflow.yml) is ever written by
this path, so no further per-file bookkeeping is needed.

Added a failing-first regression: install a catalog workflow, re-add it
with a simulated registry save OSError, and assert a clean error, the
original workflow.yml restored byte-for-byte, and the registry still
reporting the original version installed. Confirmed red (prior file
deleted) before the fix, green after.

Assisted-by: GitHub Copilot (model: Claude Sonnet 5, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-11 02:28:48 +02:00
parent 86de599d3e
commit fb64ddd48c
2 changed files with 83 additions and 2 deletions

View File

@@ -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