Fix WorkflowCatalog YAML parsing error handling and isinstance checks

Author: Copilot

src/specify_cli/workflows/catalog.py

@@ -183,6 +183,11 @@ def _load_catalog_config(
             raise WorkflowValidationError(
                 f"Failed to read catalog config {config_path}: {exc}"
             )
+        if not isinstance(data, dict):
+            raise WorkflowValidationError(
+                f"Invalid catalog config: expected a mapping, "
+                f"got {type(data).__name__}"
+            )
         catalogs_data = data.get("catalogs", [])
         if not catalogs_data:
             # Empty catalogs list (e.g. after removing last entry)
@@ -476,7 +481,12 @@ def add_catalog(self, url: str, name: str | None = None) -> None:
 
         data: dict[str, Any] = {"catalogs": []}
         if config_path.exists():
-            raw = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+            try:
+                raw = yaml.safe_load(config_path.read_text(encoding="utf-8"))
+            except (yaml.YAMLError, OSError, UnicodeDecodeError) as exc:
+                raise WorkflowValidationError(
+                    f"Catalog config file is unreadable or malformed: {exc}"
+                ) from exc
             if not isinstance(raw, dict):
                 raise WorkflowValidationError(
                     "Catalog config file is corrupted (expected a mapping)."
@@ -523,17 +533,27 @@ def _coerce_priority(value: Any) -> int:
         )
         data["catalogs"] = catalogs
 
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        with open(config_path, "w", encoding="utf-8") as f:
-            yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
+        try:
+            config_path.parent.mkdir(parents=True, exist_ok=True)
+            with open(config_path, "w", encoding="utf-8") as f:
+                yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
+        except OSError as exc:
+            raise WorkflowValidationError(
+                f"Failed to write catalog config {config_path}: {exc}"
+            ) from exc
 
     def remove_catalog(self, index: int) -> str:
         """Remove a catalog source by index (0-based). Returns the removed name."""
         config_path = self.project_root / ".specify" / "workflow-catalogs.yml"
         if not config_path.exists():
             raise WorkflowValidationError("No catalog config file found.")
 
-        data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
+        try:
+            data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
+        except (yaml.YAMLError, OSError, UnicodeDecodeError) as exc:
+            raise WorkflowValidationError(
+                f"Catalog config file is unreadable or malformed: {exc}"
+            ) from exc
         if not isinstance(data, dict):
             raise WorkflowValidationError(
                 "Catalog config file is corrupted (expected a mapping)."
@@ -552,8 +572,13 @@ def remove_catalog(self, index: int) -> str:
         removed = catalogs.pop(index)
         data["catalogs"] = catalogs
 
-        with open(config_path, "w", encoding="utf-8") as f:
-            yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
+        try:
+            with open(config_path, "w", encoding="utf-8") as f:
+                yaml.dump(data, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
+        except OSError as exc:
+            raise WorkflowValidationError(
+                f"Failed to write catalog config {config_path}: {exc}"
+            ) from exc
 
         if isinstance(removed, dict):
             return removed.get("name", f"catalog-{index + 1}")

tests/test_workflows.py

@@ -1751,6 +1751,78 @@ def test_get_catalog_configs(self, project_dir):
         assert configs[0]["name"] == "default"
         assert isinstance(configs[0]["install_allowed"], bool)
 
+    def test_load_catalog_config_non_dict_yaml_raises(self, project_dir):
+        """A YAML catalog config that is a list (not a mapping) must raise WorkflowValidationError."""
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowValidationError
+
+        config_path = project_dir / ".specify" / "workflow-catalogs.yml"
+        config_path.write_text("- item1\n- item2\n", encoding="utf-8")
+
+        catalog = WorkflowCatalog(project_dir)
+        with pytest.raises(WorkflowValidationError, match="expected a mapping"):
+            catalog.get_active_catalogs()
+
+    def test_add_catalog_malformed_yaml_raises(self, project_dir):
+        """A malformed YAML config file must raise WorkflowValidationError when adding a catalog."""
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowValidationError
+
+        config_path = project_dir / ".specify" / "workflow-catalogs.yml"
+        config_path.write_text(": invalid: yaml: {\n", encoding="utf-8")
+
+        catalog = WorkflowCatalog(project_dir)
+        with pytest.raises(WorkflowValidationError, match="unreadable or malformed"):
+            catalog.add_catalog("https://example.com/new.json")
+
+    def test_remove_catalog_malformed_yaml_raises(self, project_dir):
+        """A malformed YAML config file must raise WorkflowValidationError when removing a catalog."""
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowValidationError
+
+        catalog = WorkflowCatalog(project_dir)
+        catalog.add_catalog("https://example.com/c1.json", "first")
+
+        config_path = project_dir / ".specify" / "workflow-catalogs.yml"
+        config_path.write_text(": bad: yaml: {\n", encoding="utf-8")
+
+        with pytest.raises(WorkflowValidationError, match="unreadable or malformed"):
+            catalog.remove_catalog(0)
+
+    def test_add_catalog_wraps_write_oserror(self, project_dir, monkeypatch):
+        """An OSError on write must be wrapped as WorkflowValidationError."""
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowValidationError
+        import builtins
+
+        catalog = WorkflowCatalog(project_dir)
+        config_path = project_dir / ".specify" / "workflow-catalogs.yml"
+        real_open = builtins.open
+
+        def _raising_open(file, mode="r", *args, **kwargs):
+            if Path(file) == config_path and "w" in mode:
+                raise OSError("simulated write failure")
+            return real_open(file, mode, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "open", _raising_open)
+        with pytest.raises(WorkflowValidationError, match="Failed to write catalog config"):
+            catalog.add_catalog("https://example.com/new-catalog.json", "my-catalog")
+
+    def test_remove_catalog_wraps_write_oserror(self, project_dir, monkeypatch):
+        """An OSError on write must be wrapped as WorkflowValidationError."""
+        from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowValidationError
+        import builtins
+
+        catalog = WorkflowCatalog(project_dir)
+        catalog.add_catalog("https://example.com/c1.json", "first")
+        config_path = project_dir / ".specify" / "workflow-catalogs.yml"
+        real_open = builtins.open
+
+        def _raising_open(file, mode="r", *args, **kwargs):
+            if Path(file) == config_path and "w" in mode:
+                raise OSError("simulated write failure")
+            return real_open(file, mode, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "open", _raising_open)
+        with pytest.raises(WorkflowValidationError, match="Failed to write catalog config"):
+            catalog.remove_catalog(0)
+
 
 # ===== Integration Test =====