diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index e206be07f..95dff473b 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -353,7 +353,7 @@ def workflow_run( from .catalog import WorkflowRegistry installed_meta = WorkflowRegistry(project_root).get(source) - if installed_meta is not None and installed_meta.get("enabled", True) is False: + if isinstance(installed_meta, dict) and installed_meta.get("enabled", True) is False: err.print( f"[red]Error:[/red] Workflow '{_escape_markup(source)}' is disabled. " f"Enable with: specify workflow enable {_escape_markup(source)}" @@ -903,14 +903,19 @@ def _install_workflow_from_catalog( ) raise typer.Exit(1) - registry.add(workflow_id, { + entry = { "name": definition.name or info.get("name", workflow_id), "version": definition.version or info.get("version", "0.0.0"), "description": definition.description or info.get("description", ""), "source": "catalog", "catalog_name": info.get("_catalog_name", ""), "url": workflow_url, - }) + } + # Preserve a prior disabled state across updates/reinstalls. + existing = registry.get(workflow_id) + if isinstance(existing, dict) and existing.get("enabled", True) is False: + entry["enabled"] = False + registry.add(workflow_id, entry) console.print(f"[green]✓[/green] Workflow '{info.get('name', workflow_id)}' installed from catalog") @@ -1009,7 +1014,10 @@ def workflow_update( updates_available: list[dict[str, str]] = [] for wf_id in targets: safe_id = _escape_markup(str(wf_id)) - metadata = installed.get(wf_id) or {} + metadata = installed.get(wf_id) + if not isinstance(metadata, dict): + console.print(f"⚠ {safe_id}: Registry entry is corrupted (skipping)") + continue if metadata.get("source") != "catalog": console.print(f"⚠ {safe_id}: Installed from a local path or URL — re-add to update (skipping)") continue @@ -1097,6 +1105,11 @@ def workflow_enable( if metadata is None: console.print(f"[red]Error:[/red] Workflow '{_escape_markup(workflow_id)}' is not installed") raise typer.Exit(1) + if not isinstance(metadata, dict): + console.print( + f"[red]Error:[/red] Registry entry for '{_escape_markup(workflow_id)}' is corrupted" + ) + raise typer.Exit(1) if metadata.get("enabled", True): console.print(f"[yellow]Workflow '{_escape_markup(workflow_id)}' is already enabled[/yellow]") raise typer.Exit(0) @@ -1118,6 +1131,11 @@ def workflow_disable( if metadata is None: console.print(f"[red]Error:[/red] Workflow '{_escape_markup(workflow_id)}' is not installed") raise typer.Exit(1) + if not isinstance(metadata, dict): + console.print( + f"[red]Error:[/red] Registry entry for '{_escape_markup(workflow_id)}' is corrupted" + ) + raise typer.Exit(1) if not metadata.get("enabled", True): console.print(f"[yellow]Workflow '{_escape_markup(workflow_id)}' is already disabled[/yellow]") raise typer.Exit(0) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index ab995113c..85b53810e 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -7435,6 +7435,80 @@ steps: assert meta["version"] == "2.0.0" assert "2.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 + 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": "", + "source": "catalog", + "url": "https://example.com/workflow.yml", + "enabled": False, + }) + monkeypatch.setattr( + WorkflowCatalog, + "get_workflow_info", + lambda self, wid: { + "id": wid, + "version": "2.0.0", + "url": "https://example.com/workflow.yml", + "_install_allowed": True, + }, + ) + data = self.WORKFLOW_YAML.format(version="2.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 result.exit_code == 0, result.output + meta = WorkflowRegistry(project_dir).get("align-wf") + assert meta["version"] == "2.0.0" + assert meta["enabled"] is False + + def test_update_skips_corrupted_registry_entry(self, project_dir, monkeypatch): + import json + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.catalog import WorkflowRegistry + + 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": {"broken": "not-a-dict"}}), + encoding="utf-8", + ) + runner = CliRunner() + result = runner.invoke(app, ["workflow", "update"]) + assert result.exit_code == 0, result.output + assert "corrupted" in result.output + + def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch): + import json + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.catalog import WorkflowRegistry + + 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": {"broken": "not-a-dict"}}), + encoding="utf-8", + ) + runner = CliRunner() + for cmd in ("enable", "disable"): + result = runner.invoke(app, ["workflow", cmd, "broken"]) + assert result.exit_code != 0 + assert "corrupted" in result.output + def test_update_up_to_date_reports_and_exits_zero(self, project_dir, monkeypatch): from typer.testing import CliRunner from specify_cli import app