diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index d5f5aba2d5..c50c17e0f7 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -1886,6 +1886,13 @@ def get_speckit_version() -> str: ) app.add_typer(integration_app, name="integration") +integration_catalog_app = typer.Typer( + name="catalog", + help="Manage integration catalog sources", + add_completion=False, +) +integration_app.add_typer(integration_catalog_app, name="catalog") + INTEGRATION_JSON = ".specify/integration.json" @@ -2535,6 +2542,310 @@ def integration_upgrade( console.print(f"\n[green]✓[/green] Integration '{name}' upgraded successfully") +# ===== Integration catalog discovery commands ===== +# +# These commands mirror the workflow catalog CLI shape: +# - `search` / `info` for discovery over the active catalog stack +# - `catalog list/add/remove` for managing catalog sources +# +# They deliberately do NOT add `integration add/remove/enable/disable/ +# set-priority`: integrations are single-active (install / uninstall / switch), +# not additive like extensions and presets. + + +def _require_specify_project() -> Path: + """Return the current project root if it is a spec-kit project, else exit.""" + project_root = Path.cwd() + if not (project_root / ".specify").exists(): + console.print("[red]Error:[/red] Not a spec-kit project (no .specify/ directory)") + console.print("Run this command from a spec-kit project root") + raise typer.Exit(1) + return project_root + + +@integration_app.command("search") +def integration_search( + query: Optional[str] = typer.Argument(None, help="Search query (optional)"), + tag: Optional[str] = typer.Option(None, "--tag", help="Filter by tag"), + author: Optional[str] = typer.Option(None, "--author", help="Filter by author"), +): + """Search for integrations in the active catalog stack.""" + from .integrations import INTEGRATION_REGISTRY + from .integrations.catalog import ( + IntegrationCatalog, + IntegrationCatalogError, + IntegrationValidationError, + ) + + project_root = _require_specify_project() + integration_config = _read_integration_json(project_root) + installed_key = integration_config.get("integration") + catalog = IntegrationCatalog(project_root) + + try: + results = catalog.search(query=query, tag=tag, author=author) + except IntegrationValidationError as exc: + console.print(f"[red]Error:[/red] {exc}") + console.print( + "\nTip: Check the configuration file path shown above for invalid catalog configuration " + "(for example, .specify/integration-catalogs.yml or ~/.specify/integration-catalogs.yml)." + ) + raise typer.Exit(1) + except IntegrationCatalogError as exc: + console.print(f"[red]Error:[/red] {exc}") + if os.environ.get("SPECKIT_INTEGRATION_CATALOG_URL", "").strip(): + console.print( + "\nTip: Check the SPECKIT_INTEGRATION_CATALOG_URL environment variable for an invalid " + "catalog URL, or unset it to use the configured catalog files " + "(.specify/integration-catalogs.yml or ~/.specify/integration-catalogs.yml)." + ) + else: + console.print("\nTip: The catalog may be temporarily unavailable. Try again later.") + raise typer.Exit(1) + + if not results: + console.print("\n[yellow]No integrations found matching criteria[/yellow]") + if query or tag or author: + console.print("\nTry:") + console.print(" • Broader search terms") + console.print(" • Remove filters") + console.print(" • specify integration search (show all)") + return + + console.print(f"\n[green]Found {len(results)} integration(s):[/green]\n") + for integ in sorted(results, key=lambda e: e.get("id", "")): + iid = integ.get("id", "?") + name = integ.get("name", iid) + version = integ.get("version", "?") + console.print(f"[bold]{name}[/bold] ({iid}) v{version}") + desc = integ.get("description", "") + if desc: + console.print(f" {desc}") + + console.print(f"\n [dim]Author:[/dim] {integ.get('author', 'Unknown')}") + tags = integ.get("tags", []) + if isinstance(tags, list) and tags: + console.print(f" [dim]Tags:[/dim] {', '.join(str(t) for t in tags)}") + + cat_name = integ.get("_catalog_name", "") + install_allowed = integ.get("_install_allowed", True) + if cat_name: + if install_allowed: + console.print(f" [dim]Catalog:[/dim] {cat_name}") + else: + console.print( + f" [dim]Catalog:[/dim] {cat_name} " + "[yellow](discovery only — not installable)[/yellow]" + ) + + if iid == installed_key: + console.print("\n [green]✓ Installed[/green] (currently active)") + elif iid in INTEGRATION_REGISTRY: + console.print(f"\n [cyan]Install:[/cyan] specify integration install {iid}") + elif install_allowed: + console.print( + "\n [yellow]Found in catalog.[/yellow] Only built-in integration IDs " + "can be installed with 'specify integration install'." + ) + else: + console.print( + f"\n [yellow]⚠[/yellow] Not directly installable from '{cat_name}'." + ) + console.print() + + +@integration_app.command("info") +def integration_info( + integration_id: str = typer.Argument(..., help="Integration ID"), +): + """Show catalog details for a single integration.""" + from .integrations import INTEGRATION_REGISTRY + from .integrations.catalog import ( + IntegrationCatalog, + IntegrationCatalogError, + IntegrationValidationError, + ) + + project_root = _require_specify_project() + catalog = IntegrationCatalog(project_root) + installed_key = _read_integration_json(project_root).get("integration") + + try: + info = catalog.get_integration_info(integration_id) + except IntegrationCatalogError as exc: + info = None + # Keep the live exception so the fallback branch below can give + # different guidance for local-config vs. network failures. + catalog_error: Optional[IntegrationCatalogError] = exc + else: + catalog_error = None + + if info: + name = info.get("name", integration_id) + version = info.get("version", "?") + console.print(f"\n[bold cyan]{name}[/bold cyan] ({integration_id}) v{version}") + if info.get("description"): + console.print(f" {info['description']}") + console.print() + + console.print(f" [dim]Author:[/dim] {info.get('author', 'Unknown')}") + if info.get("license"): + console.print(f" [dim]License:[/dim] {info['license']}") + + tags = info.get("tags", []) + if isinstance(tags, list) and tags: + console.print(f" [dim]Tags:[/dim] {', '.join(str(t) for t in tags)}") + + cat_name = info.get("_catalog_name", "") + install_allowed = info.get("_install_allowed", True) + if cat_name: + install_note = "" if install_allowed else " [yellow](discovery only)[/yellow]" + console.print(f" [dim]Source catalog:[/dim] {cat_name}{install_note}") + + if info.get("repository"): + console.print(f" [dim]Repository:[/dim] {info['repository']}") + + if integration_id == installed_key: + console.print("\n [green]✓ Installed[/green] (currently active)") + elif integration_id in INTEGRATION_REGISTRY: + console.print("\n [dim]Built-in integration (not currently active)[/dim]") + return + + if integration_id in INTEGRATION_REGISTRY: + integration = INTEGRATION_REGISTRY[integration_id] + cfg = integration.config or {} + name = cfg.get("name", integration_id) + console.print(f"\n[bold cyan]{name}[/bold cyan] ({integration_id})") + console.print(" [dim]Built-in integration (not listed in catalog)[/dim]") + if integration_id == installed_key: + console.print("\n [green]✓ Installed[/green] (currently active)") + if catalog_error: + console.print(f"\n[yellow]Catalog unavailable:[/yellow] {catalog_error}") + return + + if catalog_error: + console.print(f"[red]Error:[/red] Could not query integration catalog: {catalog_error}") + if isinstance(catalog_error, IntegrationValidationError): + console.print( + "\nCheck the configuration file path shown above " + "(.specify/integration-catalogs.yml or ~/.specify/integration-catalogs.yml), " + "or use a built-in integration ID directly." + ) + elif os.environ.get("SPECKIT_INTEGRATION_CATALOG_URL", "").strip(): + console.print( + "\nCheck whether SPECKIT_INTEGRATION_CATALOG_URL is set correctly and reachable, " + "or unset it to use the configured catalog files, or use a built-in integration ID directly." + ) + else: + console.print("\nTry again when online, or use a built-in integration ID directly.") + else: + console.print(f"[red]Error:[/red] Integration '{integration_id}' not found") + console.print("\nTry: specify integration search") + raise typer.Exit(1) + + +@integration_catalog_app.command("list") +def integration_catalog_list(): + """List configured integration catalog sources.""" + from .integrations.catalog import IntegrationCatalog, IntegrationCatalogError + + project_root = _require_specify_project() + catalog = IntegrationCatalog(project_root) + env_override = os.environ.get("SPECKIT_INTEGRATION_CATALOG_URL", "").strip() + + try: + if env_override: + project_configs = None + configs = catalog.get_catalog_configs() + else: + project_configs = catalog.get_project_catalog_configs() + configs = project_configs if project_configs is not None else catalog.get_catalog_configs() + except IntegrationCatalogError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + + console.print("\n[bold cyan]Integration Catalog Sources:[/bold cyan]\n") + if env_override: + console.print( + " SPECKIT_INTEGRATION_CATALOG_URL is set; it supersedes configured catalog files." + ) + console.print( + " Project/user catalog sources are not active while the env override is set.\n" + ) + console.print("[bold]Active catalog source from environment (non-removable here):[/bold]\n") + elif project_configs is None: + console.print(" No project-level catalog sources configured.\n") + console.print("[bold]Active catalog sources (non-removable here):[/bold]\n") + else: + console.print("[bold]Project catalog sources (removable):[/bold]\n") + + for i, cfg in enumerate(configs): + install_status = ( + "[green]install allowed[/green]" + if cfg.get("install_allowed") + else "[yellow]discovery only[/yellow]" + ) + if env_override or project_configs is None: + console.print(f" - [bold]{cfg.get('name', 'catalog')}[/bold] — {install_status}") + else: + console.print(f" [{i}] [bold]{cfg.get('name', f'catalog-{i + 1}')}[/bold] — {install_status}") + console.print(f" {cfg.get('url', '')}") + if cfg.get("description"): + console.print(f" [dim]{cfg['description']}[/dim]") + console.print() + + +@integration_catalog_app.command("add") +def integration_catalog_add( + url: str = typer.Argument( + ..., + help=( + "Catalog URL to add (HTTPS required, except http://localhost, " + "http://127.0.0.1, or http://[::1] for local testing)" + ), + ), + name: Optional[str] = typer.Option(None, "--name", help="Catalog name"), +): + """Add an integration catalog source to the project config.""" + from .integrations.catalog import IntegrationCatalog, IntegrationCatalogError + + project_root = _require_specify_project() + catalog = IntegrationCatalog(project_root) + + # Normalize once here so the success message reflects what was actually + # stored. ``IntegrationCatalog.add_catalog`` strips again defensively. + normalized_url = url.strip() + + try: + catalog.add_catalog(normalized_url, name) + except IntegrationCatalogError as exc: + # Covers both URL validation (base class) and config-file validation + # (IntegrationValidationError subclass). + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + + console.print(f"[green]✓[/green] Catalog source added: {normalized_url}") + + +@integration_catalog_app.command("remove") +def integration_catalog_remove( + index: int = typer.Argument(..., help="Catalog index to remove (from 'catalog list')"), +): + """Remove an integration catalog source by 0-based index.""" + from .integrations.catalog import IntegrationCatalog, IntegrationCatalogError + + project_root = _require_specify_project() + catalog = IntegrationCatalog(project_root) + + try: + removed_name = catalog.remove_catalog(index) + except IntegrationCatalogError as exc: + console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + + console.print(f"[green]✓[/green] Catalog source '{removed_name}' removed") + + # ===== Preset Commands ===== diff --git a/src/specify_cli/integrations/catalog.py b/src/specify_cli/integrations/catalog.py index 2faa69ae96..e8d2c370d6 100644 --- a/src/specify_cli/integrations/catalog.py +++ b/src/specify_cli/integrations/catalog.py @@ -16,7 +16,7 @@ from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple import yaml from packaging import version as pkg_version @@ -30,6 +30,10 @@ class IntegrationCatalogError(Exception): """Raised when a catalog operation fails.""" +class IntegrationValidationError(IntegrationCatalogError): + """Validation error for catalog config or catalog management operations.""" + + class IntegrationDescriptorError(Exception): """Raised when an integration.yml descriptor is invalid.""" @@ -96,28 +100,36 @@ def _load_catalog_config( Returns None when the file does not exist. Raises: - IntegrationCatalogError: on invalid content + IntegrationValidationError: on any local-config / YAML problem + (parse failures, wrong shape, missing/invalid fields, + invalid catalog URLs, etc.). This is a subclass of + :class:`IntegrationCatalogError`, so any caller that already + catches ``IntegrationCatalogError`` keeps working — but + callers that want to distinguish *local config* problems + from *remote/network* problems can match the subclass. """ if not config_path.exists(): return None try: - data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {} + data = yaml.safe_load(config_path.read_text(encoding="utf-8")) except (yaml.YAMLError, OSError, UnicodeError) as exc: - raise IntegrationCatalogError( + raise IntegrationValidationError( f"Failed to read catalog config {config_path}: {exc}" - ) + ) from exc + if data is None: + data = {} if not isinstance(data, dict): - raise IntegrationCatalogError( + raise IntegrationValidationError( f"Invalid catalog config {config_path}: expected a YAML mapping at the root" ) catalogs_data = data.get("catalogs", []) if not isinstance(catalogs_data, list): - raise IntegrationCatalogError( - f"Invalid catalog config: 'catalogs' must be a list, " + raise IntegrationValidationError( + f"Invalid catalog config {config_path}: 'catalogs' must be a list, " f"got {type(catalogs_data).__name__}" ) if not catalogs_data: - raise IntegrationCatalogError( + raise IntegrationValidationError( f"Catalog config {config_path} exists but contains no 'catalogs' entries. " f"Remove the file to use built-in defaults, or add valid catalog entries." ) @@ -125,21 +137,38 @@ def _load_catalog_config( skipped: List[int] = [] for idx, item in enumerate(catalogs_data): if not isinstance(item, dict): - raise IntegrationCatalogError( - f"Invalid catalog entry at index {idx}: " + raise IntegrationValidationError( + f"Invalid catalog config {config_path}: catalog entry at index {idx}: " f"expected a mapping, got {type(item).__name__}" ) url = str(item.get("url", "")).strip() if not url: skipped.append(idx) continue - self._validate_catalog_url(url) try: - priority = int(item.get("priority", idx + 1)) + self._validate_catalog_url(url) + except IntegrationCatalogError as exc: + # ``_validate_catalog_url`` raises the base class for direct + # callers (e.g. ``add_catalog`` validating user input); when + # the bad URL came from a local config file, surface it as a + # validation error so CLI handlers can route it accordingly. + raise IntegrationValidationError( + f"Invalid catalog URL in {config_path} at index {idx}: {exc}" + ) from exc + raw_priority = item.get("priority", idx + 1) + if isinstance(raw_priority, bool): + raise IntegrationValidationError( + f"Invalid catalog config {config_path}: " + f"Invalid priority for catalog '{item.get('name', idx + 1)}': " + f"expected integer, got {raw_priority!r}" + ) + try: + priority = int(raw_priority) except (TypeError, ValueError): - raise IntegrationCatalogError( + raise IntegrationValidationError( + f"Invalid catalog config {config_path}: " f"Invalid priority for catalog '{item.get('name', idx + 1)}': " - f"expected integer, got {item.get('priority')!r}" + f"expected integer, got {raw_priority!r}" ) raw_install = item.get("install_allowed", False) if isinstance(raw_install, str): @@ -157,7 +186,7 @@ def _load_catalog_config( ) entries.sort(key=lambda e: e.priority) if not entries: - raise IntegrationCatalogError( + raise IntegrationValidationError( f"Catalog config {config_path} contains {len(catalogs_data)} " f"entries but none have valid URLs (entries at indices {skipped} " f"were skipped). Each catalog entry must have a 'url' field." @@ -196,12 +225,12 @@ def get_active_catalogs(self) -> List[IntegrationCatalogEntry]: ) ] - project_cfg = self.project_root / ".specify" / "integration-catalogs.yml" + project_cfg = self.project_root / ".specify" / self.CONFIG_FILENAME catalogs = self._load_catalog_config(project_cfg) if catalogs is not None: return catalogs - user_cfg = Path.home() / ".specify" / "integration-catalogs.yml" + user_cfg = Path.home() / ".specify" / self.CONFIG_FILENAME catalogs = self._load_catalog_config(user_cfg) if catalogs is not None: return catalogs @@ -408,6 +437,279 @@ def clear_cache(self) -> None: for f in self.cache_dir.glob(pattern): f.unlink(missing_ok=True) + # -- Catalog-source management ---------------------------------------- + + CONFIG_FILENAME = "integration-catalogs.yml" + + def get_catalog_configs(self) -> List[Dict[str, Any]]: + """Return the active catalog stack as a list of dicts. + + Thin adapter over :meth:`get_active_catalogs` that yields plain dicts + suitable for CLI rendering and JSON-like consumers. + """ + return [ + { + "name": e.name, + "url": e.url, + "priority": e.priority, + "install_allowed": e.install_allowed, + "description": e.description, + } + for e in self.get_active_catalogs() + ] + + def get_project_catalog_configs(self) -> Optional[List[Dict[str, Any]]]: + """Return removable project-level catalog config entries, if configured.""" + config_path = self.project_root / ".specify" / self.CONFIG_FILENAME + entries = self._load_catalog_config(config_path) + if entries is None: + return None + return [ + { + "name": e.name, + "url": e.url, + "priority": e.priority, + "install_allowed": e.install_allowed, + "description": e.description, + } + for e in entries + ] + + def add_catalog(self, url: str, name: Optional[str] = None) -> None: + """Add a catalog source to the project-level config file. + + The URL is normalized (whitespace stripped) and validated before being + written. Duplicate URLs are rejected, including near-duplicates that + differ only by surrounding whitespace. Priority is derived as + ``max(existing) + 1`` so the new entry sorts last in the resolution + order unless the user edits the file manually. + """ + url = url.strip() + if not url: + raise IntegrationValidationError("Catalog URL must be non-empty.") + self._validate_catalog_url(url) + config_path = self.project_root / ".specify" / self.CONFIG_FILENAME + + data: Dict[str, Any] = {"catalogs": []} + if config_path.exists(): + try: + raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) + except (yaml.YAMLError, OSError, UnicodeError) as exc: + raise IntegrationValidationError( + f"Failed to read catalog config {config_path}: {exc}" + ) from exc + if raw is None: + raw = {} + if not isinstance(raw, dict): + raise IntegrationValidationError( + f"Catalog config file {config_path} is corrupted " + "(expected a mapping)." + ) + data = raw + + catalogs = data.get("catalogs", []) + if not isinstance(catalogs, list): + raise IntegrationValidationError( + f"Catalog config {config_path} has invalid 'catalogs' value: " + "must be a list." + ) + + # Validate each existing entry before mutating anything. Fail fast so + # we don't silently preserve a corrupt sibling entry or derive a new + # priority from a bogus value. + existing_priorities: List[int] = [] + for idx, cat in enumerate(catalogs): + if not isinstance(cat, dict): + raise IntegrationValidationError( + f"Invalid catalog entry at index {idx} in {config_path}: " + f"expected a mapping, got {type(cat).__name__}." + ) + existing_url = str(cat.get("url", "")).strip() + if not existing_url: + continue + # Re-run the same URL validation used when loading, so a corrupt + # entry surfaces here instead of at the next `integration` call. + try: + self._validate_catalog_url(existing_url) + except IntegrationCatalogError as exc: + raise IntegrationValidationError( + f"Invalid catalog entry at index {idx} in {config_path}: {exc}" + ) from exc + if existing_url == url: + raise IntegrationValidationError( + f"Catalog URL already configured: {url}" + ) + if "priority" in cat: + raw_priority = cat.get("priority") + if isinstance(raw_priority, bool): + raise IntegrationValidationError( + f"Invalid catalog entry at index {idx} in {config_path}: " + f"'priority' must be an integer, got " + f"{type(raw_priority).__name__}." + ) + try: + normalized_priority = int(raw_priority) + except (TypeError, ValueError): + raise IntegrationValidationError( + f"Invalid catalog entry at index {idx} in {config_path}: " + f"'priority' must be an integer, got " + f"{raw_priority!r}." + ) from None + existing_priorities.append(normalized_priority) + else: + # Match `_load_catalog_config()`'s defaulting rule so the new + # entry still sorts after implicit-priority siblings. + existing_priorities.append(idx + 1) + + max_priority = max(existing_priorities, default=0) + normalized_name = str(name).strip() if name is not None else "" + catalogs.append( + { + "name": normalized_name or f"catalog-{len(catalogs) + 1}", + "url": url, + "priority": max_priority + 1, + "install_allowed": True, + "description": "", + } + ) + data["catalogs"] = catalogs + + config_path.parent.mkdir(parents=True, exist_ok=True) + with open(config_path, "w", encoding="utf-8") as f: + yaml.dump( + data, + f, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + + def remove_catalog(self, index: int) -> str: + """Remove a catalog source by 0-based index. + + ``index`` is interpreted in the same display order shown by + ``integration catalog list`` (i.e. sorted ascending by priority, + with missing priority defaulting to ``yaml_index + 1``, matching + ``_load_catalog_config()``). This way, the index a user sees in + ``catalog list`` is the index they pass to ``catalog remove``, + even if the underlying YAML lists entries in a different order + from how they sort by priority. + + Returns the removed catalog's name. + """ + config_path = self.project_root / ".specify" / self.CONFIG_FILENAME + if not config_path.exists(): + raise IntegrationValidationError("No catalog config file found.") + + try: + data = yaml.safe_load(config_path.read_text(encoding="utf-8")) + except (yaml.YAMLError, OSError, UnicodeError) as exc: + raise IntegrationValidationError( + f"Failed to read catalog config {config_path}: {exc}" + ) from exc + if data is None: + data = {} + if not isinstance(data, dict): + raise IntegrationValidationError( + f"Catalog config file {config_path} is corrupted " + "(expected a mapping)." + ) + + catalogs = data.get("catalogs", []) + if not isinstance(catalogs, list): + raise IntegrationValidationError( + f"Catalog config {config_path} has invalid 'catalogs' value: " + "must be a list." + ) + + if not catalogs: + # An empty list is the kind of state that only happens if the + # user hand-edited the file; our own `remove_catalog` deletes + # the file when the last entry is popped. Surface a clear + # message instead of `out of range (0--1)`. + raise IntegrationValidationError( + "Catalog config contains no catalog entries." + ) + + # Map displayed index -> raw YAML index using the same priority + # defaulting as ``_load_catalog_config``. We deliberately stay + # tolerant here (no new validation errors) because the goal is + # only to mirror the order shown by ``catalog list``; entries + # that ``_load_catalog_config`` would have rejected outright + # would have failed ``catalog list`` already. + def _is_removable_catalog_entry(item: Any) -> bool: + if not isinstance(item, dict): + return False + raw_url = item.get("url") + if raw_url is None: + return False + return bool(str(raw_url).strip()) + + priority_pairs: List[Tuple[int, int]] = [] + for yaml_idx, item in enumerate(catalogs): + if not _is_removable_catalog_entry(item): + continue + + raw_priority = item.get("priority", yaml_idx + 1) + if isinstance(raw_priority, bool): + priority = yaml_idx + 1 + else: + try: + priority = int(raw_priority) + except (TypeError, ValueError): + priority = yaml_idx + 1 + priority_pairs.append((priority, yaml_idx)) + if not priority_pairs: + raise IntegrationValidationError( + "Catalog config contains no removable catalog entries." + ) + # Stable sort: ties keep their YAML order, matching list-view ordering. + priority_pairs.sort(key=lambda p: p[0]) + display_order: List[int] = [yaml_idx for _, yaml_idx in priority_pairs] + + if index < 0 or index >= len(display_order): + raise IntegrationValidationError( + f"Catalog index {index} out of range (0-{len(display_order) - 1})." + ) + + target_yaml_idx = display_order[index] + removed = catalogs.pop(target_yaml_idx) + + if any(_is_removable_catalog_entry(item) for item in catalogs): + data["catalogs"] = catalogs + with open(config_path, "w", encoding="utf-8") as f: + yaml.dump( + data, + f, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + else: + # Removing the final entry: delete the config file rather than + # leaving behind an empty `catalogs:` list. `_load_catalog_config` + # treats an empty list as an error, so leaving the file would + # break every subsequent `integration` command until the user + # manually deletes `.specify/integration-catalogs.yml`. + # Deleting the file lets the project fall back to built-in + # defaults, which matches the behavior before any + # `catalog add` was ever run. + try: + config_path.unlink(missing_ok=True) + except OSError as exc: + raise IntegrationValidationError( + f"Failed to delete catalog config {config_path}: {exc}" + ) from exc + + fallback_name = f"catalog-{target_yaml_idx + 1}" + if isinstance(removed, dict): + removed_name = removed.get("name") + if removed_name is not None: + normalized_name = str(removed_name).strip() + if normalized_name: + return normalized_name + return fallback_name + # --------------------------------------------------------------------------- # IntegrationDescriptor (integration.yml) diff --git a/tests/integrations/test_cli.py b/tests/integrations/test_cli.py index df48323ed2..50fee6aa6f 100644 --- a/tests/integrations/test_cli.py +++ b/tests/integrations/test_cli.py @@ -628,3 +628,516 @@ def test_full_init_copilot_skills_resolves_page_templates(self, tmp_path): assert "/speckit-plan" in content, "Copilot --skills should use /speckit-plan" assert "/speckit.plan" not in content, "dot-notation leaked into Copilot skills page template" assert "__SPECKIT_COMMAND_" not in content + + +class TestIntegrationCatalogDiscoveryCLI: + """End-to-end CLI tests for `integration search`, `info`, and `catalog …`. + + All tests patch `IntegrationCatalog._get_merged_integrations` so no network + or on-disk cache is touched. Adds #2344 coverage without affecting any + existing integration install/switch/uninstall/upgrade behavior. + """ + + FAKE_INTEGRATIONS = [ + { + "id": "acme-coder", + "name": "Acme Coder", + "version": "2.0.0", + "description": "Community integration for Acme Coder", + "author": "acme-org", + "tags": ["cli", "acme"], + "_catalog_name": "community", + "_install_allowed": False, + }, + { + "id": "stellar-agent", + "name": "Stellar Agent", + "version": "1.3.0", + "description": "First-party Stellar agent integration", + "author": "stellar-labs", + "tags": ["ide"], + "_catalog_name": "default", + "_install_allowed": True, + }, + ] + + def _make_project(self, tmp_path): + project = tmp_path / "proj" + project.mkdir() + (project / ".specify").mkdir() + return project + + def _patch_catalog(self, monkeypatch, integrations=None): + """Return a stubbed `_get_merged_integrations` that yields *integrations*.""" + from specify_cli.integrations.catalog import IntegrationCatalog + + data = list(integrations if integrations is not None else self.FAKE_INTEGRATIONS) + + def fake_merged(self, force_refresh=False): + return data + + monkeypatch.setattr(IntegrationCatalog, "_get_merged_integrations", fake_merged) + + def _invoke(self, argv, cwd): + from typer.testing import CliRunner + from specify_cli import app + + runner = CliRunner() + old = os.getcwd() + try: + os.chdir(cwd) + return runner.invoke(app, argv, catch_exceptions=False) + finally: + os.chdir(old) + + # -- Project guard ----------------------------------------------------- + + def test_search_requires_specify_project(self, tmp_path): + project = tmp_path / "bare" + project.mkdir() + result = self._invoke(["integration", "search"], project) + assert result.exit_code == 1 + assert "Not a spec-kit project" in result.output + + def test_catalog_list_requires_specify_project(self, tmp_path): + project = tmp_path / "bare" + project.mkdir() + result = self._invoke(["integration", "catalog", "list"], project) + assert result.exit_code == 1 + assert "Not a spec-kit project" in result.output + + # -- search ------------------------------------------------------------ + + def test_search_lists_all(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke(["integration", "search"], project) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 0, result.output + assert "Found 2 integration(s)" in result.output + assert "acme-coder" in result.output + assert "stellar-agent" in result.output + assert "specify integration install stellar-agent" not in normalized_output + assert "Only built-in integration IDs can be installed" in normalized_output + + def test_search_validates_integration_json_before_catalog_lookup( + self, tmp_path, monkeypatch + ): + project = self._make_project(tmp_path) + (project / ".specify" / "integration.json").write_text( + "{bad json\n", encoding="utf-8" + ) + + from specify_cli.integrations.catalog import IntegrationCatalog + + def fail_search(self, **kwargs): + raise AssertionError("catalog search should not be called") + + monkeypatch.setattr(IntegrationCatalog, "search", fail_search) + + result = self._invoke(["integration", "search"], project) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 1 + assert "contains invalid JSON" in normalized_output + assert "integration.json" in normalized_output + + def test_search_filters_by_tag(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke(["integration", "search", "--tag", "acme"], project) + assert result.exit_code == 0, result.output + assert "Found 1 integration(s)" in result.output + assert "acme-coder" in result.output + assert "stellar-agent" not in result.output + + def test_search_filters_by_author(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke( + ["integration", "search", "--author", "stellar-labs"], project + ) + assert result.exit_code == 0, result.output + assert "Found 1 integration(s)" in result.output + assert "stellar-agent" in result.output + + def test_search_no_match_hint(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke( + ["integration", "search", "--tag", "nope"], project + ) + assert result.exit_code == 0, result.output + assert "No integrations found" in result.output + assert "specify integration search" in result.output + + def test_search_marks_discovery_only_entry(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke(["integration", "search", "acme"], project) + assert result.exit_code == 0, result.output + # acme-coder is flagged _install_allowed=False, so we should warn + assert "Not directly installable" in result.output + + # -- info -------------------------------------------------------------- + + def test_info_found(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke( + ["integration", "info", "stellar-agent"], project + ) + assert result.exit_code == 0, result.output + assert "Stellar Agent" in result.output + assert "stellar-agent" in result.output + assert "v1.3.0" in result.output + + def test_info_not_found(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + self._patch_catalog(monkeypatch) + result = self._invoke( + ["integration", "info", "does-not-exist"], project + ) + assert result.exit_code == 1 + assert "not found" in result.output + + def test_info_builtin_not_in_catalog(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + # Empty catalog, but copilot is a registered built-in. + self._patch_catalog(monkeypatch, integrations=[]) + result = self._invoke(["integration", "info", "copilot"], project) + assert result.exit_code == 0, result.output + assert "Built-in integration" in result.output + + # -- validation vs network guidance ------------------------------------ + + def test_search_local_config_error_shows_local_config_tip( + self, tmp_path, monkeypatch + ): + """`integration search` must point at .specify/integration-catalogs.yml + for local-config errors (not the generic 'temporarily unavailable').""" + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + # Corrupt YAML to drive _load_catalog_config -> IntegrationValidationError. + cfg = project / ".specify" / "integration-catalogs.yml" + invalid_yaml = "catalogs:\n - [bad\n" + cfg.write_text(invalid_yaml, encoding="utf-8") + + result = self._invoke(["integration", "search"], project) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 1, result.output + assert "configuration file path shown above" in normalized_output + assert ".specify/integration-catalogs.yml" in normalized_output + assert "~/.specify/integration-catalogs.yml" in normalized_output + assert "temporarily unavailable" not in normalized_output + + def test_search_invalid_env_catalog_url_shows_env_tip( + self, tmp_path, monkeypatch + ): + project = self._make_project(tmp_path) + monkeypatch.setenv( + "SPECKIT_INTEGRATION_CATALOG_URL", + "http://insecure.example.com/catalog.json", + ) + + result = self._invoke(["integration", "search"], project) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 1, result.output + assert "SPECKIT_INTEGRATION_CATALOG_URL environment variable" in normalized_output + assert "unset it to use the configured catalog files" in normalized_output + assert ".specify/integration-catalogs.yml" in normalized_output + assert "~/.specify/integration-catalogs.yml" in normalized_output + assert "temporarily unavailable" not in normalized_output + + def test_search_whitespace_env_catalog_url_uses_generic_catalog_tip( + self, tmp_path, monkeypatch + ): + project = self._make_project(tmp_path) + monkeypatch.setenv("SPECKIT_INTEGRATION_CATALOG_URL", " ") + + from specify_cli.integrations.catalog import ( + IntegrationCatalog, + IntegrationCatalogError, + ) + + def fail_search(self, **kwargs): + raise IntegrationCatalogError("catalog offline") + + monkeypatch.setattr(IntegrationCatalog, "search", fail_search) + + result = self._invoke(["integration", "search"], project) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 1, result.output + assert "temporarily unavailable" in normalized_output + assert ( + "SPECKIT_INTEGRATION_CATALOG_URL environment variable" + not in normalized_output + ) + + def test_info_unknown_with_local_config_error_shows_local_config_tip( + self, tmp_path, monkeypatch + ): + """`integration info ` falls back to the catalog-error branch + and must show local-config guidance, not 'Try again when online'.""" + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + cfg = project / ".specify" / "integration-catalogs.yml" + invalid_yaml = "catalogs:\n - [bad\n" + cfg.write_text(invalid_yaml, encoding="utf-8") + + result = self._invoke( + ["integration", "info", "definitely-not-real"], project + ) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 1, result.output + assert "configuration file path shown above" in normalized_output + assert ".specify/integration-catalogs.yml" in normalized_output + assert "~/.specify/integration-catalogs.yml" in normalized_output + assert "Try again when online" not in normalized_output + + def test_info_unknown_with_invalid_env_catalog_url_shows_env_tip( + self, tmp_path, monkeypatch + ): + project = self._make_project(tmp_path) + monkeypatch.setenv( + "SPECKIT_INTEGRATION_CATALOG_URL", + "http://insecure.example.com/catalog.json", + ) + + result = self._invoke( + ["integration", "info", "definitely-not-real"], project + ) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 1, result.output + assert "SPECKIT_INTEGRATION_CATALOG_URL" in normalized_output + assert "unset it to use the configured catalog files" in normalized_output + assert "Try again when online" not in normalized_output + + # -- catalog list / add / remove --------------------------------------- + + def test_catalog_list_shows_builtin_defaults(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + result = self._invoke(["integration", "catalog", "list"], project) + assert result.exit_code == 0, result.output + assert "Integration Catalog Sources" in result.output + assert "No project-level catalog sources configured" in result.output + assert "Active catalog sources" in result.output + assert "non-removable" in result.output + assert "default" in result.output + assert "community" in result.output + # Built-in defaults are active, but not removable project entries. + assert "[0]" not in result.output + assert "[1]" not in result.output + + def test_catalog_add_then_remove_roundtrip(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + + add_result = self._invoke( + [ + "integration", + "catalog", + "add", + "https://new.example.com/catalog.json", + "--name", + "mine", + ], + project, + ) + assert add_result.exit_code == 0, add_result.output + assert "Catalog source added" in add_result.output + + cfg_path = project / ".specify" / "integration-catalogs.yml" + assert cfg_path.exists() + + list_result = self._invoke(["integration", "catalog", "list"], project) + assert list_result.exit_code == 0, list_result.output + assert "Project catalog sources" in list_result.output + assert "[0]" in list_result.output + assert "mine" in list_result.output + assert "default" not in list_result.output + assert "community" not in list_result.output + + remove_result = self._invoke( + ["integration", "catalog", "remove", "0"], project + ) + assert remove_result.exit_code == 0, remove_result.output + assert "'mine' removed" in remove_result.output + + def test_catalog_list_env_override_supersedes_project_config( + self, tmp_path, monkeypatch + ): + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.setenv( + "SPECKIT_INTEGRATION_CATALOG_URL", + "https://env.example.com/catalog.json", + ) + cfg_path = project / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + { + "url": "https://project.example.com/catalog.json", + "name": "project", + "priority": 1, + } + ] + } + ), + encoding="utf-8", + ) + + result = self._invoke(["integration", "catalog", "list"], project) + normalized_output = _normalize_cli_output(result.output) + assert result.exit_code == 0, result.output + assert "SPECKIT_INTEGRATION_CATALOG_URL is set" in normalized_output + assert "supersedes configured catalog files" in normalized_output + assert "non-removable" in normalized_output + assert "https://env.example.com/catalog.json" in normalized_output + assert "https://project.example.com/catalog.json" not in normalized_output + assert "[0]" not in normalized_output + + def test_catalog_add_strips_whitespace_in_success_output_and_storage( + self, tmp_path, monkeypatch + ): + """Surrounding whitespace in the URL must not appear in the success + message or be persisted to the YAML config.""" + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + + padded_url = " https://padded.example.com/catalog.json " + clean_url = "https://padded.example.com/catalog.json" + + add_result = self._invoke( + [ + "integration", + "catalog", + "add", + padded_url, + "--name", + "padded", + ], + project, + ) + assert add_result.exit_code == 0, add_result.output + assert clean_url in add_result.output + assert padded_url not in add_result.output + + cfg_path = project / ".specify" / "integration-catalogs.yml" + import yaml as _yaml + data = _yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + urls = [c["url"] for c in data["catalogs"]] + assert clean_url in urls + assert padded_url not in urls + + def test_catalog_add_rejects_invalid_url(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + result = self._invoke( + [ + "integration", + "catalog", + "add", + "http://insecure.example.com/catalog.json", + ], + project, + ) + assert result.exit_code == 1 + assert "HTTPS" in result.output + + def test_catalog_add_rejects_duplicate(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + url = "https://dup.example.com/catalog.json" + first = self._invoke( + ["integration", "catalog", "add", url], project + ) + assert first.exit_code == 0, first.output + second = self._invoke( + ["integration", "catalog", "add", url], project + ) + assert second.exit_code == 1 + assert "already configured" in second.output + + def test_catalog_remove_out_of_range(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + # Need a config file for remove to attempt an index lookup + self._invoke( + [ + "integration", + "catalog", + "add", + "https://only.example.com/catalog.json", + ], + project, + ) + result = self._invoke( + ["integration", "catalog", "remove", "9"], project + ) + assert result.exit_code == 1 + assert "out of range" in result.output + + def test_catalog_remove_without_config(self, tmp_path, monkeypatch): + project = self._make_project(tmp_path) + result = self._invoke( + ["integration", "catalog", "remove", "0"], project + ) + assert result.exit_code == 1 + assert "No catalog config" in result.output + + def test_catalog_remove_final_entry_restores_defaults( + self, tmp_path, monkeypatch + ): + """End-to-end: add → remove-last-entry → list should not error. + + Regression for the flow where a user adds a catalog, removes it, then + runs any follow-up integration command. Without the fix the config + file would be left as `catalogs: []` and every subsequent + `integration` call would fail with "contains no 'catalogs' entries". + """ + project = self._make_project(tmp_path) + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + + add = self._invoke( + [ + "integration", + "catalog", + "add", + "https://only.example.com/catalog.json", + "--name", + "only", + ], + project, + ) + assert add.exit_code == 0, add.output + + remove = self._invoke( + ["integration", "catalog", "remove", "0"], project + ) + assert remove.exit_code == 0, remove.output + assert "'only' removed" in remove.output + + cfg_path = project / ".specify" / "integration-catalogs.yml" + assert not cfg_path.exists(), ( + "config file should be deleted when the final catalog is removed" + ) + + # Follow-up command must succeed and show the built-in defaults, + # not error out on "contains no 'catalogs' entries". + listing = self._invoke(["integration", "catalog", "list"], project) + assert listing.exit_code == 0, listing.output + assert "default" in listing.output + assert "community" in listing.output diff --git a/tests/integrations/test_integration_catalog.py b/tests/integrations/test_integration_catalog.py index 6d82a6c390..fed2e0de65 100644 --- a/tests/integrations/test_integration_catalog.py +++ b/tests/integrations/test_integration_catalog.py @@ -12,6 +12,7 @@ IntegrationCatalogError, IntegrationDescriptor, IntegrationDescriptorError, + IntegrationValidationError, ) @@ -115,8 +116,45 @@ def test_empty_config_raises(self, tmp_path): cfg = specify / "integration-catalogs.yml" cfg.write_text(yaml.dump({"catalogs": []})) cat = IntegrationCatalog(tmp_path) - with pytest.raises(IntegrationCatalogError, match="no 'catalogs' entries"): + with pytest.raises(IntegrationCatalogError, match="no 'catalogs' entries") as exc_info: cat.get_active_catalogs() + assert str(cfg) in str(exc_info.value) + + def test_empty_config_file_raises_no_catalogs(self, tmp_path, monkeypatch): + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + specify = tmp_path / ".specify" + specify.mkdir() + cfg = specify / "integration-catalogs.yml" + cfg.write_text("", encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="no 'catalogs' entries" + ) as exc_info: + cat.get_active_catalogs() + assert str(cfg) in str(exc_info.value) + + @pytest.mark.parametrize("config_content", ["[]\n", "false\n", "0\n", "''\n"]) + def test_load_catalog_config_rejects_falsy_non_mapping_roots( + self, tmp_path, monkeypatch, config_content + ): + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + specify = tmp_path / ".specify" + specify.mkdir() + cfg = specify / "integration-catalogs.yml" + cfg.write_text(config_content, encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, + match="expected a YAML mapping at the root", + ) as exc_info: + cat.get_active_catalogs() + assert str(cfg) in str(exc_info.value) # --------------------------------------------------------------------------- @@ -654,3 +692,789 @@ def test_upgrade_no_manifest(self, tmp_path): os.chdir(old) assert result.exit_code == 0 assert "Nothing to upgrade" in result.output + + +# --------------------------------------------------------------------------- +# IntegrationCatalog — catalog source management (get_catalog_configs / add / remove) +# --------------------------------------------------------------------------- + + +class TestCatalogSourceManagement: + """Unit tests for add_catalog / remove_catalog / get_catalog_configs.""" + + def _isolate(self, tmp_path, monkeypatch): + """Point HOME at tmp_path and clear the env override so we read built-ins.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) + monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False) + (tmp_path / ".specify").mkdir() + + def test_get_catalog_configs_returns_builtin_stack(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + configs = cat.get_catalog_configs() + assert [c["name"] for c in configs] == ["default", "community"] + assert all(isinstance(c["url"], str) and c["url"] for c in configs) + assert configs[0]["install_allowed"] is True + assert configs[1]["install_allowed"] is False + + def test_add_catalog_creates_config_file(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://new.example.com/catalog.json", name="mine") + + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + assert cfg_path.exists() + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert data["catalogs"] == [ + { + "name": "mine", + "url": "https://new.example.com/catalog.json", + "priority": 1, + "install_allowed": True, + "description": "", + } + ] + # Round-trip: active catalogs should now come from the config file. + active = cat.get_active_catalogs() + assert [e.name for e in active] == ["mine"] + + def test_add_catalog_recovers_from_empty_config_file(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text("", encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://example.com/catalog.json") + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert data["catalogs"] == [ + { + "name": "catalog-1", + "url": "https://example.com/catalog.json", + "priority": 1, + "install_allowed": True, + "description": "", + } + ] + + @pytest.mark.parametrize("config_content", ["[]\n", "false\n", "0\n", "''\n"]) + def test_add_catalog_rejects_falsy_non_mapping_config_roots( + self, tmp_path, monkeypatch, config_content + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text(config_content, encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, + match="corrupted.*expected a mapping", + ) as exc_info: + cat.add_catalog("https://example.com/catalog.json") + assert str(cfg_path) in str(exc_info.value) + + def test_add_catalog_auto_derives_name_and_priority(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://a.example.com/catalog.json") + cat.add_catalog("https://b.example.com/catalog.json") + + data = yaml.safe_load( + (tmp_path / ".specify" / "integration-catalogs.yml").read_text(encoding="utf-8") + ) + entries = data["catalogs"] + assert [e["name"] for e in entries] == ["catalog-1", "catalog-2"] + assert [e["priority"] for e in entries] == [1, 2] + + def test_add_catalog_normalizes_name(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://a.example.com/catalog.json", name=" mine ") + cat.add_catalog("https://b.example.com/catalog.json", name=" ") + + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + entries = data["catalogs"] + assert [e["name"] for e in entries] == ["mine", "catalog-2"] + + def test_add_catalog_rejects_duplicate_url(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://dup.example.com/catalog.json") + with pytest.raises(IntegrationValidationError, match="already configured"): + cat.add_catalog("https://dup.example.com/catalog.json") + + def test_add_catalog_rejects_invalid_url(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + with pytest.raises(IntegrationCatalogError, match="HTTPS"): + cat.add_catalog("http://insecure.example.com/catalog.json") + assert not (tmp_path / ".specify" / "integration-catalogs.yml").exists() + + def test_add_catalog_rejects_empty_url(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + with pytest.raises(IntegrationValidationError, match="must be non-empty"): + cat.add_catalog(" ") + assert not (tmp_path / ".specify" / "integration-catalogs.yml").exists() + + def test_remove_catalog_without_config_errors(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + with pytest.raises(IntegrationValidationError, match="No catalog config"): + cat.remove_catalog(0) + + def test_remove_catalog_happy_path(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://a.example.com/catalog.json", name="a") + cat.add_catalog("https://b.example.com/catalog.json", name="b") + + removed = cat.remove_catalog(0) + assert removed == "a" + + data = yaml.safe_load( + (tmp_path / ".specify" / "integration-catalogs.yml").read_text(encoding="utf-8") + ) + assert [e["name"] for e in data["catalogs"]] == ["b"] + + def test_remove_catalog_index_out_of_range(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://a.example.com/catalog.json", name="a") + with pytest.raises(IntegrationValidationError, match="out of range"): + cat.remove_catalog(5) + with pytest.raises(IntegrationValidationError, match="out of range"): + cat.remove_catalog(-1) + + def test_corrupt_config_rejected_on_add(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text("- just\n- a\n- list\n", encoding="utf-8") + cat = IntegrationCatalog(tmp_path) + with pytest.raises(IntegrationValidationError, match="corrupted") as exc_info: + cat.add_catalog("https://new.example.com/catalog.json") + assert str(cfg_path) in str(exc_info.value) + + def test_add_catalog_rejects_non_list_catalogs_with_config_path( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump({"catalogs": "not-a-list"}), encoding="utf-8" + ) + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="invalid 'catalogs' value" + ) as exc_info: + cat.add_catalog("https://new.example.com/catalog.json") + assert str(cfg_path) in str(exc_info.value) + + def test_add_catalog_rejects_non_mapping_entry_with_config_path( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump({"catalogs": ["not-a-mapping"]}), encoding="utf-8" + ) + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="Invalid catalog entry at index 0" + ) as exc_info: + cat.add_catalog("https://new.example.com/catalog.json") + message = str(exc_info.value) + assert str(cfg_path) in message + assert "expected a mapping" in message + + def test_add_catalog_skips_blank_url_entries(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": " ", "name": "blank", "priority": 99}, + { + "url": "https://a.example.com/catalog.json", + "name": "a", + "priority": 5, + }, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://b.example.com/catalog.json", name="b") + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert data["catalogs"][-1]["name"] == "b" + assert data["catalogs"][-1]["priority"] == 6 + + def test_add_catalog_rejects_non_integer_priority(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + { + "url": "https://a.example.com/catalog.json", + "name": "a", + "priority": "first", + } + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, + match="'priority' must be an integer, got 'first'", + ): + cat.add_catalog("https://b.example.com/catalog.json") + + def test_add_catalog_accepts_numeric_string_priority(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + { + "url": "https://a.example.com/catalog.json", + "name": "a", + "priority": "10", + } + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://b.example.com/catalog.json", name="b") + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert data["catalogs"][-1]["name"] == "b" + assert data["catalogs"][-1]["priority"] == 11 + + @pytest.mark.parametrize( + ("bad_url", "reason"), + [ + ("http://insecure.example.com/catalog.json", "HTTPS"), + (123, "HTTPS"), + ], + ) + def test_add_catalog_rejects_existing_entry_with_bad_url( + self, tmp_path, monkeypatch, bad_url, reason + ): + """A sibling entry with an http:// URL should block a new add.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + { + "url": bad_url, + "name": "bad", + } + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + with pytest.raises(IntegrationValidationError) as exc_info: + cat.add_catalog("https://good.example.com/catalog.json") + message = str(exc_info.value) + assert str(cfg_path) in message + assert "index 0" in message + assert reason in message + + def test_add_catalog_wraps_yaml_parse_errors(self, tmp_path, monkeypatch): + """Invalid YAML on disk surfaces as IntegrationValidationError, not a raw YAMLError.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + invalid_yaml = "catalogs:\n - url: 'https://a.example.com/cat.json'\n - [bad\n" + cfg_path.write_text(invalid_yaml, encoding="utf-8") + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="Failed to read catalog config" + ): + cat.add_catalog("https://b.example.com/catalog.json") + + def test_remove_catalog_wraps_yaml_parse_errors(self, tmp_path, monkeypatch): + """Invalid YAML on disk surfaces as IntegrationValidationError from remove_catalog too.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + invalid_yaml = "catalogs:\n - url: 'https://a.example.com/cat.json'\n - [bad\n" + cfg_path.write_text(invalid_yaml, encoding="utf-8") + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="Failed to read catalog config" + ): + cat.remove_catalog(0) + + def test_add_catalog_defaults_missing_priority_to_index_plus_one( + self, tmp_path, monkeypatch + ): + """Existing entries without `priority` should be treated as idx + 1. + + Matches the rule in `_load_catalog_config()`: a valid catalog entry + without an explicit `priority` sorts at `idx + 1`, so the new entry + should get `max(...) + 1` from those derived values. + """ + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + # No explicit priority → should be treated as 1 + {"url": "https://a.example.com/cat.json", "name": "a"}, + # No explicit priority → should be treated as 2 + {"url": "https://b.example.com/cat.json", "name": "b"}, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://c.example.com/cat.json", name="c") + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + new_entry = data["catalogs"][-1] + assert new_entry["name"] == "c" + # max(implicit [1, 2]) + 1 == 3 + assert new_entry["priority"] == 3 + + def test_add_catalog_strips_whitespace_in_url(self, tmp_path, monkeypatch): + """Whitespace around the incoming URL should be normalized before write.""" + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog(" https://a.example.com/catalog.json\n", name="a") + + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert data["catalogs"][0]["url"] == "https://a.example.com/catalog.json" + + def test_add_catalog_rejects_whitespace_only_duplicate(self, tmp_path, monkeypatch): + """A second add with only whitespace differences must be rejected as a duplicate.""" + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://a.example.com/catalog.json", name="a") + with pytest.raises(IntegrationValidationError, match="already configured"): + cat.add_catalog(" https://a.example.com/catalog.json ") + + def test_remove_catalog_wraps_unlink_oserror(self, tmp_path, monkeypatch): + """An OSError from `Path.unlink` surfaces as IntegrationValidationError.""" + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://only.example.com/catalog.json", name="only") + + from pathlib import Path as _Path + + def boom(self, *args, **kwargs): + raise OSError("simulated unlink failure") + + monkeypatch.setattr(_Path, "unlink", boom) + + with pytest.raises( + IntegrationValidationError, match="Failed to delete catalog config" + ): + cat.remove_catalog(0) + + def test_remove_catalog_ignores_missing_final_config_during_unlink( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://only.example.com/catalog.json", name="only") + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + + from pathlib import Path as _Path + + original_unlink = _Path.unlink + + def delete_first_then_unlink(self, *args, **kwargs): + if self == cfg_path and self.exists(): + original_unlink(self) + return original_unlink(self, *args, **kwargs) + + monkeypatch.setattr(_Path, "unlink", delete_first_then_unlink) + + assert cat.remove_catalog(0) == "only" + assert not cfg_path.exists() + + def test_remove_catalog_empty_list_gives_clear_error(self, tmp_path, monkeypatch): + """Hand-edited empty `catalogs:` produces a clear error, not '0--1'.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text(yaml.dump({"catalogs": []}), encoding="utf-8") + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="contains no catalog entries" + ): + cat.remove_catalog(0) + + def test_remove_catalog_empty_config_file_gives_clear_error( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text("", encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="contains no catalog entries" + ): + cat.remove_catalog(0) + + def test_remove_catalog_rejects_non_list_catalogs_with_config_path( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text( + yaml.dump({"catalogs": "not-a-list"}), encoding="utf-8" + ) + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="invalid 'catalogs' value" + ) as exc_info: + cat.remove_catalog(0) + assert str(cfg_path) in str(exc_info.value) + + @pytest.mark.parametrize("config_content", ["[]\n", "false\n", "0\n", "''\n"]) + def test_remove_catalog_rejects_falsy_non_mapping_config_roots( + self, tmp_path, monkeypatch, config_content + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.write_text(config_content, encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, + match="corrupted.*expected a mapping", + ) as exc_info: + cat.remove_catalog(0) + assert str(cfg_path) in str(exc_info.value) + + def test_remove_last_catalog_deletes_file_and_restores_defaults( + self, tmp_path, monkeypatch + ): + """Removing the final catalog must not leave behind `catalogs: []`. + + `_load_catalog_config` treats an empty `catalogs` list as an error, + so writing that file would break every subsequent `integration` + command. Removing the last entry should delete the config file so the + project falls back to built-in defaults. + """ + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + + cat.add_catalog("https://only.example.com/catalog.json", name="only") + assert cfg_path.exists() + assert [e.name for e in cat.get_active_catalogs()] == ["only"] + + removed = cat.remove_catalog(0) + assert removed == "only" + + assert not cfg_path.exists(), ( + "remove_catalog should delete the config file when emptying it" + ) + # Follow-up loads fall back to built-in defaults, not an error. + active = cat.get_active_catalogs() + assert [e.name for e in active] == ["default", "community"] + + def test_load_catalog_config_raises_validation_error_for_invalid_yaml( + self, tmp_path, monkeypatch + ): + """Local-config problems must surface as IntegrationValidationError so + CLI handlers can route them to local-config (not network) guidance.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + invalid_yaml = "catalogs:\n - [bad\n" + cfg_path.write_text(invalid_yaml, encoding="utf-8") + + cat = IntegrationCatalog(tmp_path) + # Subclass match: IntegrationValidationError (specifically), not the + # bare IntegrationCatalogError parent that callers used previously. + with pytest.raises(IntegrationValidationError, match="Failed to read catalog config"): + cat.get_active_catalogs() + + def test_load_catalog_config_rejects_boolean_priority(self, tmp_path, monkeypatch): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + { + "url": "https://a.example.com/catalog.json", + "name": "a", + "priority": True, + } + ] + } + ), + encoding="utf-8", + ) + + cat = IntegrationCatalog(tmp_path) + with pytest.raises( + IntegrationValidationError, match="Invalid priority|expected integer" + ) as exc_info: + cat.get_active_catalogs() + assert str(cfg_path) in str(exc_info.value) + + @pytest.mark.parametrize( + ("raw_name", "expected"), + [ + (None, "catalog-1"), + (" ", "catalog-1"), + (123, "123"), + ], + ) + def test_remove_catalog_normalizes_removed_display_name( + self, tmp_path, monkeypatch, raw_name, expected + ): + self._isolate(tmp_path, monkeypatch) + cat = IntegrationCatalog(tmp_path) + cat.add_catalog("https://one.example.com/c.json", name="one") + + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + data["catalogs"][0]["name"] = raw_name + cfg_path.write_text(yaml.dump(data), encoding="utf-8") + + assert cat.remove_catalog(0) == expected + + def test_remove_catalog_uses_display_order_with_explicit_priorities( + self, tmp_path, monkeypatch + ): + """`remove_catalog(index)` must remove the entry shown at that index by + `catalog list`, not the entry at that raw YAML position.""" + self._isolate(tmp_path, monkeypatch) + # YAML order: alpha (priority=20), beta (priority=10), gamma (priority=15). + # Display (sorted by priority asc): beta (10), gamma (15), alpha (20). + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": "https://alpha.example.com/c.json", "name": "alpha", "priority": 20}, + {"url": "https://beta.example.com/c.json", "name": "beta", "priority": 10}, + {"url": "https://gamma.example.com/c.json", "name": "gamma", "priority": 15}, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + # Display index 0 = beta (lowest priority), not alpha (raw YAML idx 0). + removed = cat.remove_catalog(0) + assert removed == "beta" + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + remaining_names = [c["name"] for c in data["catalogs"]] + # YAML order is preserved for the survivors; only beta is gone. + assert remaining_names == ["alpha", "gamma"] + + def test_remove_catalog_display_order_with_missing_priorities( + self, tmp_path, monkeypatch + ): + """Entries without `priority` default to `idx + 1` (matching + `_load_catalog_config`), so display order tracks YAML order and the + first display entry is the first YAML entry.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": "https://one.example.com/c.json", "name": "one"}, + {"url": "https://two.example.com/c.json", "name": "two"}, + {"url": "https://three.example.com/c.json", "name": "three"}, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + # Implicit priorities: one=1, two=2, three=3 → display order matches YAML. + removed = cat.remove_catalog(0) + assert removed == "one" + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert [c["name"] for c in data["catalogs"]] == ["two", "three"] + + def test_remove_catalog_bool_priority_falls_back_to_yaml_index( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": "https://one.example.com/c.json", "name": "one"}, + { + "url": "https://bool.example.com/c.json", + "name": "bool", + "priority": False, + }, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + removed = cat.remove_catalog(0) + + assert removed == "one" + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert [c["name"] for c in data["catalogs"]] == ["bool"] + + def test_remove_catalog_display_order_skips_blank_url_entries( + self, tmp_path, monkeypatch + ): + """Blank-url entries are not shown by catalog list, so remove skips them too.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": " ", "name": "blank", "priority": 0}, + {"url": "https://one.example.com/c.json", "name": "one"}, + {"url": "https://two.example.com/c.json", "name": "two"}, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + removed = cat.remove_catalog(0) + assert removed == "one" + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert [c["name"] for c in data["catalogs"]] == ["blank", "two"] + + def test_remove_catalog_deletes_file_when_only_skipped_entries_remain( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": " ", "name": "blank", "priority": 0}, + {"url": "https://one.example.com/c.json", "name": "one"}, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + removed = cat.remove_catalog(0) + assert removed == "one" + assert not cfg_path.exists() + + active = cat.get_active_catalogs() + assert [e.name for e in active] == ["default", "community"] + + def test_remove_catalog_allows_numeric_url_entry_cleanup( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump({"catalogs": [{"name": "numeric-url", "url": 123}]}), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + removed = cat.remove_catalog(0) + + assert removed == "numeric-url" + assert not cfg_path.exists() + + def test_remove_catalog_errors_when_no_entries_are_removable( + self, tmp_path, monkeypatch + ): + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": "", "name": "empty"}, + {"name": "missing"}, + "not-a-mapping", + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + with pytest.raises( + IntegrationValidationError, + match="no removable catalog entries", + ): + cat.remove_catalog(0) + + def test_remove_catalog_display_order_mixes_explicit_and_default( + self, tmp_path, monkeypatch + ): + """An explicit low priority should sort ahead of default-priority + siblings, even if it appears later in the YAML.""" + self._isolate(tmp_path, monkeypatch) + cfg_path = tmp_path / ".specify" / "integration-catalogs.yml" + cfg_path.parent.mkdir(parents=True, exist_ok=True) + # Defaults: a=1, b=2 (implicit). Explicit c=0 → display: c, a, b. + cfg_path.write_text( + yaml.dump( + { + "catalogs": [ + {"url": "https://a.example.com/c.json", "name": "a"}, + {"url": "https://b.example.com/c.json", "name": "b"}, + {"url": "https://c.example.com/c.json", "name": "c", "priority": 0}, + ] + } + ), + encoding="utf-8", + ) + cat = IntegrationCatalog(tmp_path) + + removed = cat.remove_catalog(0) + assert removed == "c" + + data = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) + assert [c["name"] for c in data["catalogs"]] == ["a", "b"]