Address PR feedback and improve install flow

Author: DyanGalih

src/specify_cli/__init__.py

@@ -1086,7 +1086,12 @@ def extension_add(
                         console.print(f"  {REINSTALL_COMMAND}")
                         raise typer.Exit(1)
 
-                    # Enforce install_allowed policy
+                    # If a different approved source exists, use it instead of prompting.
+                    installable_info = catalog.get_installable_extension_info(resolved_id)
+                    if installable_info is not None:
+                        ext_info = installable_info
+
+                    # Enforce install_allowed policy only when no approved source exists.
                     if not ext_info.get("_install_allowed", True):
                         catalog_name = ext_info.get("_catalog_name", "community")
                         console.print()

src/specify_cli/extensions.py

@@ -2112,11 +2112,19 @@ def _catalog_entry_to_dict(self, entry: CatalogEntry) -> Dict[str, Any]:
 
     def approve_catalog_install(self, catalog_name: str) -> CatalogEntry:
         """Persist install permission for a catalog while preserving the stack."""
-        active_catalogs = self.get_active_catalogs()
+        config_path = self.project_root / ".specify" / self.CONFIG_FILENAME
+
+        # Base the update on the project-level config if it exists
+        if config_path.exists():
+            base_catalogs = self._load_catalog_config(config_path) or []
+        else:
+            # Otherwise, preserve the currently active stack so user-level catalogs remain available.
+            base_catalogs = self.get_active_catalogs()
+
         updated_catalogs: List[Dict[str, Any]] = []
         approved_entry: Optional[CatalogEntry] = None
 
-        for entry in active_catalogs:
+        for entry in base_catalogs:
             if entry.name == catalog_name:
                 entry = self._entry(
                     url=entry.url,
@@ -2128,6 +2136,22 @@ def approve_catalog_install(self, catalog_name: str) -> CatalogEntry:
                 approved_entry = entry
             updated_catalogs.append(self._catalog_entry_to_dict(entry))
 
+        # If the catalog wasn't found in the base (e.g., a custom user-level catalog),
+        # we pull it from the active catalogs and append it to the project stack.
+        if approved_entry is None:
+            for entry in self.get_active_catalogs():
+                if entry.name == catalog_name:
+                    entry = self._entry(
+                        url=entry.url,
+                        name=entry.name,
+                        priority=entry.priority,
+                        install_allowed=True,
+                        description=entry.description,
+                    )
+                    approved_entry = entry
+                    updated_catalogs.append(self._catalog_entry_to_dict(entry))
+                    break
+
         if approved_entry is None:
             raise ValidationError(
                 f"Catalog '{catalog_name}' is not active and cannot be approved"
@@ -2542,6 +2566,36 @@ def get_extension_info(self, extension_id: str) -> Optional[Dict[str, Any]]:
                 return ext_data
         return None
 
+    def get_installable_extension_info(self, extension_id: str) -> Optional[Dict[str, Any]]:
+        """Return the first installable source for an extension, if any.
+
+        This checks the active catalogs in priority order and returns the
+        highest-priority source that is actually allowed to install. It is
+        used by the add flow to avoid prompting for approval when a usable
+        approved source already exists.
+        """
+        for catalog_entry in self.get_active_catalogs():
+            try:
+                catalog_data = self._fetch_single_catalog(catalog_entry, force_refresh=False)
+            except ExtensionError:
+                continue
+
+            ext_data = catalog_data.get("extensions", {}).get(extension_id)
+            if not isinstance(ext_data, dict):
+                continue
+
+            if not catalog_entry.install_allowed:
+                continue
+
+            return {
+                **ext_data,
+                "id": extension_id,
+                "_catalog_name": catalog_entry.name,
+                "_install_allowed": catalog_entry.install_allowed,
+            }
+
+        return None
+
     def download_extension(
         self, extension_id: str, target_dir: Optional[Path] = None
     ) -> Path:
@@ -2559,8 +2613,8 @@ def download_extension(
         """
         import urllib.error
 
-        # Get extension info from catalog
-        ext_info = self.get_extension_info(extension_id)
+        # Get the best installable source first, then fall back to the merged view.
+        ext_info = self.get_installable_extension_info(extension_id) or self.get_extension_info(extension_id)
         if not ext_info:
             raise ExtensionError(f"Extension '{extension_id}' not found in catalog")
 

tests/test_extensions.py

@@ -3623,6 +3623,7 @@ def fake_open(req, timeout=None):
         }
 
         with patch.object(catalog, "get_extension_info", return_value=ext_info), \
+             patch.object(catalog, "get_installable_extension_info", return_value=ext_info), \
              patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
             catalog.download_extension("test-ext", target_dir=temp_dir)
 
@@ -3669,6 +3670,7 @@ def fake_open(req, timeout=None):
         }
 
         with patch.object(catalog, "get_extension_info", return_value=ext_info), \
+             patch.object(catalog, "get_installable_extension_info", return_value=ext_info), \
              patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
             catalog.download_extension("test-ext", target_dir=temp_dir)
 
@@ -4191,6 +4193,58 @@ def test_approve_catalog_install_preserves_active_stack(self, temp_dir):
         assert parsed["catalogs"][1]["name"] == "community"
         assert parsed["catalogs"][1]["install_allowed"] is True
 
+    def test_approve_catalog_install_preserves_user_level_active_catalogs(self, temp_dir):
+        """Approving a catalog should preserve the full active stack when no project config exists."""
+        import yaml as yaml_module
+        from unittest.mock import patch
+
+        project_dir = self._make_project(temp_dir)
+        home_dir = temp_dir / "home"
+        specify_home = home_dir / ".specify"
+        specify_home.mkdir(parents=True)
+        with (specify_home / "extension-catalogs.yml").open("w", encoding="utf-8") as f:
+            yaml_module.safe_dump(
+                {
+                    "catalogs": [
+                        {
+                            "name": "alpha",
+                            "url": "https://alpha.example.com/catalog.json",
+                            "priority": 1,
+                            "install_allowed": False,
+                        },
+                        {
+                            "name": "community",
+                            "url": ExtensionCatalog.COMMUNITY_CATALOG_URL,
+                            "priority": 2,
+                            "install_allowed": False,
+                        },
+                        {
+                            "name": "beta",
+                            "url": "https://beta.example.com/catalog.json",
+                            "priority": 3,
+                            "install_allowed": True,
+                        },
+                    ]
+                },
+                f,
+                sort_keys=False,
+                allow_unicode=True,
+            )
+
+        catalog = ExtensionCatalog(project_dir)
+
+        with patch("specify_cli.extensions.Path.home", return_value=home_dir):
+            approved = catalog.approve_catalog_install("community")
+
+        config_path = project_dir / ".specify" / "extension-catalogs.yml"
+        parsed = yaml_module.safe_load(config_path.read_text(encoding="utf-8"))
+
+        assert approved.name == "community"
+        assert [entry["name"] for entry in parsed["catalogs"]] == ["alpha", "community", "beta"]
+        assert parsed["catalogs"][0]["install_allowed"] is False
+        assert parsed["catalogs"][1]["install_allowed"] is True
+        assert parsed["catalogs"][2]["install_allowed"] is True
+
     def test_approve_catalog_install_rejects_symlinked_specify_dir(self, temp_dir):
         """Approval writes fail closed when .specify resolves outside the project root."""
         project_dir = self._make_project(temp_dir)
@@ -4720,6 +4774,7 @@ def test_add_by_display_name_uses_resolved_id_for_download(self, tmp_path):
         # Mock catalog that returns extension by display name
         mock_catalog = MagicMock()
         mock_catalog.get_extension_info.return_value = None  # ID lookup fails
+        mock_catalog.get_installable_extension_info.return_value = None  # Installable lookup fails
         mock_catalog.search.return_value = [
             {
                 "id": "acme-jira-integration",
@@ -4822,14 +4877,20 @@ def test_add_blocked_extension_approval_updates_project_catalog_config(self, tmp
             commands=[],
         )
 
-        with patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
+        with (
+            patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
             "id": "security-review",
             "name": "Security Review",
             "version": "1.0.0",
             "description": "Security review extension",
             "_catalog_name": "community",
             "_install_allowed": False,
-        }),              patch("specify_cli.extensions.ExtensionCatalog.download_extension", return_value=zip_path),              patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=mock_manifest),              patch("typer.confirm", return_value=True),              patch.object(Path, "cwd", return_value=project_dir):
+        }),
+            patch("specify_cli.extensions.ExtensionCatalog.download_extension", return_value=zip_path),
+            patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=mock_manifest),
+            patch("typer.confirm", return_value=True),
+            patch.object(Path, "cwd", return_value=project_dir),
+        ):
             result = runner.invoke(
                 app,
                 ["extension", "add", "security-review"],
@@ -4864,14 +4925,19 @@ def record_status(*args, **kwargs):
             call_order.append("spinner")
             return MagicMock()
 
-        with patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
+        with (
+            patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
             "id": "security-review",
             "name": "Security Review",
             "version": "1.0.0",
             "description": "Security review extension",
             "_catalog_name": "community",
             "_install_allowed": False,
-        }),              patch("typer.confirm", side_effect=lambda *a, **kw: (call_order.append("confirm"), False)[-1]),              patch("specify_cli.console.status", side_effect=record_status),              patch.object(Path, "cwd", return_value=project_dir):
+        }),
+            patch("typer.confirm", side_effect=lambda *a, **kw: (call_order.append("confirm"), False)[-1]),
+            patch("specify_cli.console.status", side_effect=record_status),
+            patch.object(Path, "cwd", return_value=project_dir),
+        ):
             result = runner.invoke(
                 app,
                 ["extension", "add", "security-review"],
@@ -4894,14 +4960,18 @@ def test_add_blocked_extension_cancel_leaves_config_unchanged(self, tmp_path):
         project_dir.mkdir()
         (project_dir / ".specify").mkdir()
 
-        with patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
+        with (
+            patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
             "id": "security-review",
             "name": "Security Review",
             "version": "1.0.0",
             "description": "Security review extension",
             "_catalog_name": "community",
             "_install_allowed": False,
-        }),              patch("typer.confirm", return_value=False),              patch.object(Path, "cwd", return_value=project_dir):
+        }),
+            patch("typer.confirm", return_value=False),
+            patch.object(Path, "cwd", return_value=project_dir),
+        ):
             result = runner.invoke(
                 app,
                 ["extension", "add", "security-review"],
@@ -4939,14 +5009,80 @@ def test_add_approved_catalog_skips_approval_prompt(self, tmp_path):
         def unexpected_confirm(*args, **kwargs):
             raise AssertionError("Approval prompt should not run for approved catalogs")
 
-        with patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
+        with (
+            patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
             "id": "security-review",
             "name": "Security Review",
             "version": "1.0.0",
             "description": "Security review extension",
             "_catalog_name": "default",
             "_install_allowed": True,
-        }),              patch("specify_cli.extensions.ExtensionCatalog.download_extension", return_value=zip_path),              patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=mock_manifest),              patch("typer.confirm", side_effect=unexpected_confirm),              patch("specify_cli.console.status", return_value=contextlib.nullcontext()),              patch.object(Path, "cwd", return_value=project_dir):
+        }),
+            patch("specify_cli.extensions.ExtensionCatalog.download_extension", return_value=zip_path),
+            patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=mock_manifest),
+            patch("typer.confirm", side_effect=unexpected_confirm),
+            patch("specify_cli.console.status", return_value=contextlib.nullcontext()),
+            patch.object(Path, "cwd", return_value=project_dir),
+        ):
+            result = runner.invoke(
+                app,
+                ["extension", "add", "security-review"],
+                catch_exceptions=True,
+            )
+
+        assert result.exit_code == 0, result.output
+        assert "Catalog Approval Required" not in result.output
+
+    def test_add_prefers_approved_source_over_blocked_duplicate(self, tmp_path):
+        """If the same extension exists in approved and blocked catalogs, the add flow should skip approval."""
+        from typer.testing import CliRunner
+        from unittest.mock import patch
+        from types import SimpleNamespace
+        from specify_cli import app
+        import contextlib
+
+        runner = CliRunner()
+        project_dir = tmp_path / "test-project"
+        project_dir.mkdir()
+        (project_dir / ".specify").mkdir()
+
+        zip_path = tmp_path / "approved-duplicate.zip"
+        zip_path.write_bytes(b"fake-zip")
+        mock_manifest = SimpleNamespace(
+            id="security-review",
+            name="Security Review",
+            version="1.0.0",
+            description="Security review extension",
+            warnings=[],
+            commands=[],
+        )
+
+        def unexpected_confirm(*args, **kwargs):
+            raise AssertionError("Approval prompt should not run when an approved source exists")
+
+        with (
+            patch("specify_cli.extensions.ExtensionCatalog.get_extension_info", return_value={
+            "id": "security-review",
+            "name": "Security Review",
+            "version": "1.0.0",
+            "description": "Security review extension",
+            "_catalog_name": "community",
+            "_install_allowed": False,
+        }),
+            patch("specify_cli.extensions.ExtensionCatalog.get_installable_extension_info", return_value={
+            "id": "security-review",
+            "name": "Security Review",
+            "version": "1.0.0",
+            "description": "Security review extension",
+            "_catalog_name": "default",
+            "_install_allowed": True,
+        }),
+            patch("specify_cli.extensions.ExtensionCatalog.download_extension", return_value=zip_path),
+            patch("specify_cli.extensions.ExtensionManager.install_from_zip", return_value=mock_manifest),
+            patch("typer.confirm", side_effect=unexpected_confirm),
+            patch("specify_cli.console.status", return_value=contextlib.nullcontext()),
+            patch.object(Path, "cwd", return_value=project_dir),
+        ):
             result = runner.invoke(
                 app,
                 ["extension", "add", "security-review"],
@@ -4971,7 +5107,10 @@ def test_add_not_found_still_reports_missing_extension(self, tmp_path):
         mock_catalog.get_extension_info.return_value = None
         mock_catalog.search.return_value = []
 
-        with patch("specify_cli.extensions.ExtensionCatalog", return_value=mock_catalog),              patch.object(Path, "cwd", return_value=project_dir):
+        with (
+            patch("specify_cli.extensions.ExtensionCatalog", return_value=mock_catalog),
+            patch.object(Path, "cwd", return_value=project_dir),
+        ):
             result = runner.invoke(
                 app,
                 ["extension", "add", "does-not-exist"],