diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 97bf58a04e..294dd8759a 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -76,8 +76,16 @@ def _load(self) -> dict[str, Any]: if self.registry_path.exists(): try: with open(self.registry_path, encoding="utf-8") as f: - return json.load(f) - except (json.JSONDecodeError, ValueError): + data = json.load(f) + # Validate shape: must be a dict with a dict "workflows" field, + # otherwise every method that indexes data["workflows"] crashes. + # Mirrors StepRegistry._load. + if not isinstance(data, dict): + return {"schema_version": self.SCHEMA_VERSION, "workflows": {}} + if not isinstance(data.get("workflows"), dict): + data["workflows"] = {} + return data + except (json.JSONDecodeError, ValueError, OSError, UnicodeError): # Corrupted registry file — reset to default return {"schema_version": self.SCHEMA_VERSION, "workflows": {}} return {"schema_version": self.SCHEMA_VERSION, "workflows": {}} @@ -438,9 +446,9 @@ def search( q = query.lower() searchable = " ".join( [ - wf_data.get("name", ""), - wf_data.get("description", ""), - wf_data.get("id", ""), + str(wf_data.get("name") or ""), + str(wf_data.get("description") or ""), + str(wf_data.get("id") or ""), ] ).lower() if q not in searchable: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 2fdbf887b3..8eb808e6d1 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -4318,12 +4318,59 @@ def test_persistence(self, project_dir): registry2 = WorkflowRegistry(project_dir) assert registry2.is_installed("test-wf") + @pytest.mark.parametrize("bad_content", ["[]", '{"schema_version": "1.0"}']) + def test_load_tolerates_misshaped_registry(self, project_dir, bad_content): + """A JSON-valid but mis-shaped registry file must not crash every method. + + A list root, or a dict lacking a 'workflows' mapping, previously made + is_installed/get/list/remove/add raise TypeError/KeyError. Mirrors the + shape guard StepRegistry._load already has. + """ + from specify_cli.workflows.catalog import WorkflowRegistry + + reg_path = project_dir / ".specify" / "workflows" / "workflow-registry.json" + reg_path.parent.mkdir(parents=True, exist_ok=True) + reg_path.write_text(bad_content, encoding="utf-8") + + registry = WorkflowRegistry(project_dir) + assert registry.data == { + "schema_version": WorkflowRegistry.SCHEMA_VERSION, + "workflows": {}, + } + # None of these should raise on the recovered-default shape. + assert registry.is_installed("x") is False + assert registry.get("x") is None + assert registry.list() == {} # list() always returns a dict + registry.remove("x") + registry.add("x", {"name": "X"}) + assert registry.is_installed("x") + # ===== Workflow Catalog Tests ===== class TestWorkflowCatalog: """Test WorkflowCatalog catalog resolution.""" + def test_search_with_non_string_fields(self, project_dir, monkeypatch): + """Non-string workflow fields (null/int name/description) must not + raise TypeError in search — StepCatalog.search already coerces these.""" + from specify_cli.workflows.catalog import WorkflowCatalog + + catalog = WorkflowCatalog(project_dir) + monkeypatch.setattr(catalog, "_get_merged_workflows", lambda **kw: { + "42": { + "id": 42, + "name": None, + "description": 99, + "_catalog_name": "test", + "_install_allowed": True, + }, + }) + + assert len(catalog.search()) == 1 + assert len(catalog.search(query="42")) == 1 + assert len(catalog.search(query="missing")) == 0 + def test_default_catalogs(self, project_dir, monkeypatch): from specify_cli.workflows.catalog import WorkflowCatalog