Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions src/specify_cli/workflows/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -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": {}}
Expand Down Expand Up @@ -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:
Expand Down
47 changes: 47 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down