fix(workflows): catch OSError in per-workflow update loop and make restore best-effort

Transient FS errors (perms, disk full) from backup read or write no
longer abort the whole update run. The restore is wrapped in its own
try/except so a failed write only warns, and the offending workflow
is reported via 'Failed to update' like other per-workflow failures.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-10 08:26:37 +02:00
parent 8409bca244
commit 129362f29b
2 changed files with 64 additions and 3 deletions

View File

@@ -1096,10 +1096,21 @@ def workflow_update(
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"])
except typer.Exit:
except (typer.Exit, OSError) as exc:
if backup is not None and wf_dir is not None and wf_file is not None:
wf_dir.mkdir(parents=True, exist_ok=True)
wf_file.write_bytes(backup)
try:
wf_dir.mkdir(parents=True, exist_ok=True)
wf_file.write_bytes(backup)
except OSError as restore_exc:
console.print(
f"[yellow]Warning:[/yellow] Could not restore backup for "
f"'{_escape_markup(update['id'])}': {_escape_markup(str(restore_exc))}"
)
if isinstance(exc, OSError):
console.print(
f"[red]Error:[/red] Filesystem error updating "
f"'{_escape_markup(update['id'])}': {_escape_markup(str(exc))}"
)
failed.append(update["id"])
if failed:

View File

@@ -7608,6 +7608,56 @@ steps:
assert result.exit_code != 0
assert "Failed to update" in result.output
def test_update_survives_oserror_from_backup_read(self, project_dir, monkeypatch):
"""OSError while reading the backup for one workflow must not abort the whole update."""
import json
from pathlib import Path
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry, WorkflowCatalog
monkeypatch.chdir(project_dir)
registry_path = WorkflowRegistry(project_dir).registry_path
registry_path.parent.mkdir(parents=True, exist_ok=True)
registry_path.write_text(
json.dumps(
{
"schema_version": "1.0",
"workflows": {
"wobbly": {
"name": "Wobbly",
"version": "0.0.1",
"source": "catalog",
"url": "https://example.com/wobbly.yml",
},
},
}
),
encoding="utf-8",
)
# An existing installed file whose read_bytes will raise OSError.
wf_file = project_dir / ".specify" / "workflows" / "wobbly" / "workflow.yml"
wf_file.parent.mkdir(parents=True, exist_ok=True)
wf_file.write_bytes(b"schema_version: '1.0'\nworkflow:\n id: wobbly\n name: Wobbly\n version: 0.0.1\nsteps: []\n")
monkeypatch.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {"version": "9.9.9", "url": "https://example.com/wobbly.yml", "_install_allowed": True},
)
real_read = Path.read_bytes
def _boom(self, *args, **kwargs):
if self.name == "workflow.yml" and "wobbly" in str(self):
raise OSError("simulated permission denied")
return real_read(self, *args, **kwargs)
monkeypatch.setattr(Path, "read_bytes", _boom)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "update"], input="y\n")
assert result.exit_code != 0, result.output
assert "Filesystem error" in result.output
assert "Failed to update" in result.output
def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch):
import json
from typer.testing import CliRunner