diff --git a/tests/test_workflows.py b/tests/test_workflows.py index acfdaa0b3b..92e9080d06 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -9150,6 +9150,94 @@ def _raising_open(file, mode="r", *args, **kwargs): with pytest.raises(WorkflowValidationError, match="Failed to write catalog config"): catalog.remove_catalog(0) + def test_oversized_workflow_catalog_does_not_block_healthy_one(self, project_dir, monkeypatch): + """A healthy catalog still works after an oversized one was rejected.""" + from specify_cli.workflows.catalog import ( + WorkflowCatalog, + WorkflowCatalogEntry, + WorkflowCatalogError, + ) + from specify_cli.authentication import http as auth_http + from specify_cli.workflows import catalog as catalog_module + + monkeypatch.setattr(catalog_module, "MAX_JSON_CATALOG_BYTES", 512) + + call_count = [0] + + class _OversizedResponse: + def __init__(self): + self._data = b"x" * 1024 + self._pos = 0 + + def read(self, n=-1): + if n < 0: + chunk = self._data[self._pos:] + self._pos = len(self._data) + return chunk + chunk = self._data[self._pos : self._pos + n] + self._pos += len(chunk) + return chunk + + def geturl(self): + return "https://bad.example.com/catalog.json" + + def __enter__(self): + return self + + def __exit__(self, *a): + pass + + class _HealthyResponse: + def __init__(self): + self._data = b'{"workflows": {}}' + self._pos = 0 + + def read(self, n=-1): + if n < 0: + chunk = self._data[self._pos:] + self._pos = len(self._data) + return chunk + chunk = self._data[self._pos : self._pos + n] + self._pos += len(chunk) + return chunk + + def geturl(self): + return "https://good.example.com/catalog.json" + + def __enter__(self): + return self + + def __exit__(self, *a): + pass + + def fake_open(url, timeout=30, redirect_validator=None): + call_count[0] += 1 + if call_count[0] == 1: + return _OversizedResponse() + return _HealthyResponse() + + monkeypatch.setattr(auth_http, "open_url", fake_open) + + catalog = WorkflowCatalog(project_dir) + + bad_entry = WorkflowCatalogEntry( + url="https://bad.example.com/catalog.json", + name="bad", + priority=1, + install_allowed=True, + ) + with pytest.raises(WorkflowCatalogError, match="exceeds maximum size"): + catalog._fetch_single_catalog(bad_entry, force_refresh=True) + + good_entry = WorkflowCatalogEntry( + url="https://good.example.com/catalog.json", + name="good", + priority=1, + install_allowed=True, + ) + result = catalog._fetch_single_catalog(good_entry, force_refresh=True) + assert isinstance(result, dict) + # ===== Integration Test ===== @@ -9981,6 +10069,94 @@ def test_get_step_info_returns_entry_or_none(self, project_dir, monkeypatch): missing = catalog.get_step_info("nonexistent") assert missing is None + def test_oversized_step_catalog_does_not_block_healthy_one(self, project_dir, monkeypatch): + """A healthy step catalog still works after an oversized one was rejected.""" + from specify_cli.workflows.catalog import ( + StepCatalog, + StepCatalogEntry, + StepCatalogError, + ) + from specify_cli.authentication import http as auth_http + from specify_cli.workflows.step import catalog as step_catalog_module + + monkeypatch.setattr(step_catalog_module, "MAX_JSON_CATALOG_BYTES", 512) + + call_count = [0] + + class _OversizedResponse: + def __init__(self): + self._data = b"x" * 1024 + self._pos = 0 + + def read(self, n=-1): + if n < 0: + chunk = self._data[self._pos:] + self._pos = len(self._data) + return chunk + chunk = self._data[self._pos : self._pos + n] + self._pos += len(chunk) + return chunk + + def geturl(self): + return "https://bad.example.com/steps.json" + + def __enter__(self): + return self + + def __exit__(self, *a): + pass + + class _HealthyResponse: + def __init__(self): + self._data = b'{"steps": {}}' + self._pos = 0 + + def read(self, n=-1): + if n < 0: + chunk = self._data[self._pos:] + self._pos = len(self._data) + return chunk + chunk = self._data[self._pos : self._pos + n] + self._pos += len(chunk) + return chunk + + def geturl(self): + return "https://good.example.com/steps.json" + + def __enter__(self): + return self + + def __exit__(self, *a): + pass + + def fake_open(url, timeout=30, redirect_validator=None): + call_count[0] += 1 + if call_count[0] == 1: + return _OversizedResponse() + return _HealthyResponse() + + monkeypatch.setattr(auth_http, "open_url", fake_open) + + catalog = StepCatalog(project_dir) + + bad_entry = StepCatalogEntry( + url="https://bad.example.com/steps.json", + name="bad", + priority=1, + install_allowed=True, + ) + with pytest.raises(StepCatalogError, match="exceeds maximum size"): + catalog._fetch_single_catalog(bad_entry, force_refresh=True) + + good_entry = StepCatalogEntry( + url="https://good.example.com/steps.json", + name="good", + priority=1, + install_allowed=True, + ) + result = catalog._fetch_single_catalog(good_entry, force_refresh=True) + assert isinstance(result, dict) + # ===== Load Custom Steps Tests =====