diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 6d78354809..aee112c332 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -12,6 +12,7 @@ import errno import hashlib import json +import logging import os import re import shutil @@ -45,6 +46,8 @@ from ..catalogs import CatalogStackBase from ..shared_infra import verify_archive_sha256 +logger = logging.getLogger(__name__) + _FALLBACK_CORE_COMMAND_NAMES = frozenset( { "analyze", @@ -3508,18 +3511,15 @@ def get_active_catalogs(self) -> List[CatalogEntry]: Raises: ValidationError: If a catalog URL is invalid """ - import sys - # 1. SPECKIT_CATALOG_URL env var replaces all defaults for backward compat if env_value := os.environ.get("SPECKIT_CATALOG_URL"): catalog_url = env_value.strip() self._validate_catalog_url(catalog_url) if catalog_url != self.DEFAULT_CATALOG_URL: if not getattr(self, "_non_default_catalog_warning_shown", False): - print( - "Warning: Using non-default extension catalog. " + logger.warning( + "Using non-default extension catalog. " "Only use catalogs from sources you trust.", - file=sys.stderr, ) self._non_default_catalog_warning_shown = True return [ @@ -3743,8 +3743,6 @@ def _get_merged_extensions( Raises: ExtensionError: If all catalogs fail to fetch """ - import sys - active_catalogs = self.get_active_catalogs() merged: Dict[str, Dict[str, Any]] = {} any_success = False @@ -3754,9 +3752,8 @@ def _get_merged_extensions( catalog_data = self._fetch_single_catalog(catalog_entry, force_refresh) any_success = True except ExtensionError as e: - print( - f"Warning: Could not fetch catalog '{catalog_entry.name}': {e}", - file=sys.stderr, + logger.warning( + "Could not fetch catalog '%s': %s", catalog_entry.name, e, ) continue diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 3d9146d52b..de7518e48a 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -5936,6 +5936,39 @@ def test_download_extension_preserves_tar_archive_format( assert archive_path.name == "test-ext-1.0.0.tar.gz" assert archive_path.read_bytes() == archive_bytes + def test_catalog_fetch_failure_logged_as_warning(self, temp_dir, caplog): + """When a catalog fetch fails, the error must be logged at WARNING + level instead of being silently swallowed or printed to stderr.""" + import logging + from pathlib import Path + from unittest.mock import patch + from specify_cli.extensions import ExtensionCatalog, ExtensionError + + project_dir = temp_dir / "project" + project_dir.mkdir() + (project_dir / ".specify").mkdir() + + catalog = ExtensionCatalog(project_dir) + + entry = CatalogEntry( + url="https://example.com/broken-catalog.json", + name="broken", + priority=1, + install_allowed=True, + ) + + # Force _fetch_single_catalog to raise for all catalogs + with patch.object(catalog, "get_active_catalogs", return_value=[entry]), \ + patch.object(catalog, "_fetch_single_catalog", side_effect=ExtensionError("network error")): + with caplog.at_level(logging.WARNING): + try: + catalog._get_merged_extensions() + except ExtensionError: + pass # may still raise if all catalogs fail + + assert any("broken" in record.message and "network error" in record.message + for record in caplog.records if record.levelno == logging.WARNING) + # ===== CatalogEntry Tests =====