fix(workflows): refetch non-UTF-8 catalog caches (#3901)

Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Marsel Safin
2026-07-31 19:02:49 +02:00
committed by GitHub
parent 1831fffde6
commit 14e82353cb
2 changed files with 55 additions and 2 deletions

View File

@@ -514,7 +514,7 @@ class WorkflowCatalog:
cached = json.load(f)
if isinstance(cached, dict):
return cached
except (json.JSONDecodeError, OSError):
except (UnicodeDecodeError, json.JSONDecodeError, OSError):
# Ignore invalid/unreadable cache and fall back to fetching from source.
pass
@@ -1210,7 +1210,7 @@ class StepCatalog:
cached = json.load(f)
if isinstance(cached, dict):
return cached
except (json.JSONDecodeError, OSError):
except (UnicodeDecodeError, json.JSONDecodeError, OSError):
# Ignore invalid/unreadable cache and fall back to fetching from source.
pass

View File

@@ -7392,6 +7392,59 @@ class TestWorkflowCatalog:
assert catalog._fetch_single_catalog(entry) == payload
@pytest.mark.parametrize("catalog_type", ["workflow", "step"])
def test_non_utf8_cached_catalog_is_refetched(
self, project_dir, monkeypatch, catalog_type
):
import io
from specify_cli.authentication import http as auth_http
from specify_cli.workflows.catalog import (
StepCatalog,
StepCatalogEntry,
WorkflowCatalog,
WorkflowCatalogEntry,
)
catalog_cls = WorkflowCatalog if catalog_type == "workflow" else StepCatalog
entry_cls = (
WorkflowCatalogEntry
if catalog_type == "workflow"
else StepCatalogEntry
)
payload_key = "workflows" if catalog_type == "workflow" else "steps"
url = f"https://example.com/{catalog_type}.json"
catalog = catalog_cls(project_dir)
cache_path, metadata_path = catalog._get_cache_paths(url)
cache_path.parent.mkdir(parents=True, exist_ok=True)
cache_path.write_bytes(b"\xff\xfe")
metadata_path.write_text(
json.dumps({"fetched_at": 4_102_444_800}),
encoding="utf-8",
)
payload = {"schema_version": "1.0", payload_key: {}}
class _FakeResponse(io.BytesIO):
def geturl(self):
return url
monkeypatch.setattr(
auth_http,
"open_url",
lambda url, timeout=30, redirect_validator=None: _FakeResponse(
json.dumps(payload).encode("utf-8")
),
)
entry = entry_cls(
url=url,
name="test",
priority=1,
install_allowed=True,
)
assert catalog._fetch_single_catalog(entry) == payload
assert json.loads(cache_path.read_text(encoding="utf-8")) == payload
def test_non_mapping_stale_workflow_catalog_is_rejected(
self, project_dir, monkeypatch
):