github/github-spec-kit
1
0
Fork 0
mirror of https://github.com/github/spec-kit.git synced 2026-07-03 12:28:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add catalog discovery CLI commands (#2360)

Browse Source 
* feat: add catalog discovery CLI commands

* fix: address second Copilot review

* fix: address third Copilot review

* fix: align catalog remove with displayed order

* fix: route local catalog config errors to local guidance

* fix: address integration catalog review feedback

* fix: accept numeric string catalog priorities

* fix: align catalog remove with visible entries

* fix: preserve invalid catalog root validation

* fix: include invalid catalog priority value

* fix: preserve falsy catalog root validation

* fix: clarify integration catalog guidance

* fix: align integration catalog list and remove

* fix: align integration catalog edge cases

* fix: clarify catalog error guidance tests

* fix: clarify integration catalog edge cases

* fix: harden integration catalog removal

* fix: validate integration state before catalog search

* fix: reject empty integration catalog URL

* fix: allow catalog remove to clean non-string URLs

* fix: address catalog env and priority review

* fix: align catalog source display names

* fix: align catalog fallback names
This commit is contained in:
Adrian Osorio Blanchard
2026-04-29 08:24:30 -04:00
committed by GitHub
parent 9cf3151a72
commit 1049e17a43
4 changed files with 2071 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

315
src/specify_cli/__init__.py
View File

@@ -1886,6 +1886,13 @@ integration_app = typer.Typer(
)
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,314 @@ 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]"
)
raw_name = cfg.get("name")
display_name = str(raw_name).strip() if raw_name is not None else ""
if not display_name:
display_name = f"catalog-{i + 1}"
if env_override or project_configs is None:
console.print(f" - [bold]{display_name}[/bold] — {install_status}")
else:
console.print(f" [{i}] [bold]{display_name}[/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 =====

355
src/specify_cli/integrations/catalog.py
View File

@@ -16,7 +16,7 @@ import re
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 @@ class IntegrationCatalog:
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,31 +137,52 @@ class IntegrationCatalog:
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))
except (TypeError, ValueError):
raise IntegrationCatalogError(
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 {item.get('priority')!r}"
f"expected integer, got {raw_priority!r}"
)
try:
priority = int(raw_priority)
except (TypeError, ValueError):
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}"
)
raw_install = item.get("install_allowed", False)
if isinstance(raw_install, str):
install_allowed = raw_install.strip().lower() in ("true", "yes", "1")
else:
install_allowed = bool(raw_install)
raw_name = item.get("name")
name = str(raw_name).strip() if raw_name is not None else ""
if not name:
name = f"catalog-{len(entries) + 1}"
entries.append(
IntegrationCatalogEntry(
url=url,
name=str(item.get("name", f"catalog-{idx + 1}")),
name=name,
priority=priority,
install_allowed=install_allowed,
description=str(item.get("description", "")),
@@ -157,7 +190,7 @@ class IntegrationCatalog:
)
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 +229,12 @@ class IntegrationCatalog:
)
]
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 +441,288 @@ class IntegrationCatalog:
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] = []
valid_catalog_count = 0
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}"
)
valid_catalog_count += 1
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 ""
generated_name = f"catalog-{valid_catalog_count + 1}"
catalogs.append(
{
"name": normalized_name or generated_name,
"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-{index + 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
removed_url = removed.get("url")
if removed_url is not None:
normalized_url = str(removed_url).strip()
if normalized_url:
return normalized_url
return fallback_name
# ---------------------------------------------------------------------------
# IntegrationDescriptor (integration.yml)

547
tests/integrations/test_cli.py
View File

@@ -628,3 +628,550 @@ class TestSharedInfraCommandRefs:
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 <unknown>` 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_normalizes_blank_project_catalog_names(
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)
cfg_path = project / ".specify" / "integration-catalogs.yml"
cfg_path.write_text(
yaml.dump(
{
"catalogs": [
{
"url": "https://null-name.example.com/catalog.json",
"name": None,
},
{
"url": "https://blank-name.example.com/catalog.json",
"name": " ",
},
]
}
),
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 "[0] catalog-1" in normalized_output
assert "[1] catalog-2" in normalized_output
assert "None" not in normalized_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

875
tests/integrations/test_integration_catalog.py
View File

@@ -12,6 +12,7 @@ from specify_cli.integrations.catalog import (
IntegrationCatalogError,
IntegrationDescriptor,
IntegrationDescriptorError,
IntegrationValidationError,
)
@@ -115,8 +116,45 @@ class TestActiveCatalogs:
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,838 @@ class TestIntegrationUpgrade:
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_default_name_ignores_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"}]}),
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"][-1]["name"] == "catalog-1"
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", [None, " "])
def test_load_catalog_config_defaults_blank_names(
self, tmp_path, monkeypatch, raw_name
):
self._isolate(tmp_path, monkeypatch)
cfg_path = tmp_path / ".specify" / "integration-catalogs.yml"
cfg_path.write_text(
yaml.dump(
{
"catalogs": [
{
"url": " ",
"name": "skipped",
},
{
"url": "https://example.com/catalog.json",
"name": raw_name,
}
]
}
),
encoding="utf-8",
)
cat = IntegrationCatalog(tmp_path)
assert [entry.name for entry in cat.get_active_catalogs()] == ["catalog-1"]
@pytest.mark.parametrize(
("raw_name", "expected"),
[
(None, "https://one.example.com/c.json"),
(" ", "https://one.example.com/c.json"),
(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.
# The blank name should fall back to the removed URL, not raw YAML idx.
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": " ",
"priority": 0,
},
]
}
),
encoding="utf-8",
)
cat = IntegrationCatalog(tmp_path)
removed = cat.remove_catalog(0)
assert removed == "https://c.example.com/c.json"
data = yaml.safe_load(cfg_path.read_text(encoding="utf-8"))
assert [c["name"] for c in data["catalogs"]] == ["a", "b"]