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: Integration catalog — discovery, versioning, and community distribution (#2130)

Browse Source 
* Initial plan

* feat: add integration catalog system with catalog files, IntegrationCatalog class, list --catalog flag, upgrade command, integration.yml descriptor, and tests

Agent-Logs-Url: https://github.com/github/spec-kit/sessions/bbcd44e8-c69c-4735-adc1-bdf1ce109184

Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>

* fix: address PR review feedback

- Replace empty except with cache cleanup in _fetch_single_catalog
- Log teardown failure warning instead of silent pass in upgrade
- Validate catalog_data and integrations are dicts before use
- Catch OSError/UnicodeError in IntegrationDescriptor._load
- Add isinstance checks for integration/requires/provides/commands
- Enforce semver (X.Y.Z) instead of PEP 440 for descriptor versions
- Fix docstring and CONTRIBUTING.md to match actual block-on-modified behavior
- Restore old manifest on upgrade failure for transactional safety

* refactor: address second round of PR review feedback

- Remove dead cache_file/cache_metadata_file attributes from IntegrationCatalog
- Deduplicate non-default catalog warning (show once per process)
- Anchor version regex to reject partial matches like 1.0.0beta
- Fix 'Preserved modified' message to 'Skipped' for accuracy
- Make upgrade transactional: install new files first, then remove stale
  old-only files, so a failed setup leaves old integration intact
- Update CONTRIBUTING.md: speckit_version validates presence only

* Potential fix for pull request finding 'Empty except'

Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

* fix: address third round of PR review feedback

- Fix CONTRIBUTING.md JSON examples to show full catalog structure with
  schema_version and integrations wrapper
- Wrap cache writes in try/except OSError for read-only project dirs
- Validate _load_catalog_config YAML root is a dict
- Skip non-dict integ_data entries in merged catalog
- Normalize tags to list-of-strings before filtering/searching
- Add path traversal containment check for stale file deletion
- Clarify docstring: lower numeric priority = higher precedence

* fix: address fourth round of PR review feedback

- Remove unused _write_catalog helper from test file
- Fix comment: tests use monkeypatched urlopen, not file:// URLs
- Wrap cache unlink calls in OSError handler
- Add explicit encoding='utf-8' to all cache read_text/write_text calls
- Restore packaging.version.Version for descriptor version validation
  to align with extension/preset validators
- Add missing goose entry to integrations/catalog.json

* fix: remove unused Path import, add comment to empty except

* fix: validate descriptor root is dict, add shared infra to upgrade

- Add isinstance(self.data, dict) check at start of _validate() so
  non-mapping YAML roots raise IntegrationDescriptorError
- Run _install_shared_infra() and ensure_executable_scripts() in
  upgrade command to match install/switch behavior

* fix: address sixth round of PR review feedback

- Validate integration.id/name/version/description are strings
- Catch TypeError in pkg_version.Version() for non-string versions
- Swap validation order: check catalogs type before emptiness
- Isolate TestActiveCatalogs from user ~/.specify/ via monkeypatch

* fix: address seventh round of PR review feedback

- Update docs: version field uses PEP 440, not semver
- Harden search() against non-string author/name/description fields
- Validate requires.speckit_version is a non-empty string
- Validate command name/file are non-empty strings, file is safe relative path
- Handle stale symlinks in upgrade cleanup
- Document catalog configuration stack in README.md

* fix: validate script entries, remove destructive teardown from upgrade rollback

- Validate provides.scripts entries are non-empty strings with safe relative paths
- Remove teardown from upgrade rollback since setup overwrites in-place —
  teardown would delete files that were working before the upgrade

* fix: use consistent resolved root for stale-file cleanup paths

* fix: validate redirect URL and reject drive-qualified paths

- Validate final URL after redirects with _validate_catalog_url()
- Reject paths with Path.drive or Path.anchor for Windows safety
- Update FakeResponse mocks with geturl() method

* fix: fix docstring backticks, assert file modification in upgrade tests

* docs: clarify directory naming convention for hyphenated integration keys

* fix: correct key type hint, isolate all catalog tests from env

- Fix key parameter type to str | None (defaults to None)
- Add HOME/USERPROFILE monkeypatch and clear SPECKIT_INTEGRATION_CATALOG_URL
  in all TestCatalogFetch tests for full environment isolation

* fix: neutralize catalog table title, handle non-dict cache metadata

* fix: validate requires.tools entries in descriptor

* fix: show discovery-only status, clear metadata files in clear_cache

* fix: catch OSError/UnicodeError in cache read path

* refactor: reuse IntegrationManifest.uninstall for stale-file cleanup

* fix: normalize null tools to empty list in descriptor accessor

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: mnriem <15701806+mnriem@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
This commit is contained in:
Copilot
2026-04-16 08:59:32 -05:00
committed by GitHub
parent e0fd355dad
commit 282dd3da56
7 changed files with 1979 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
integrations/CONTRIBUTING.md Normal file
View File

@@ -0,0 +1,142 @@
# Contributing to the Integration Catalog
This guide covers adding integrations to both the **built-in** and **community** catalogs.
## Adding a Built-In Integration
Built-in integrations are maintained by the Spec Kit core team and ship with the CLI.
### Checklist
1. **Create the integration subpackage** under `src/specify_cli/integrations/<package_dir>/`
— `<package_dir>` matches the integration key when it contains no hyphens (e.g., `gemini`), or replaces hyphens with underscores when it does (e.g., key `cursor-agent` → directory `cursor_agent/`, key `kiro-cli` → directory `kiro_cli/`). Python package names cannot use hyphens.
2. **Implement the integration class** extending `MarkdownIntegration`, `TomlIntegration`, or `SkillsIntegration`
3. **Register the integration** in `src/specify_cli/integrations/__init__.py`
4. **Add tests** under `tests/integrations/test_integration_<package_dir>.py`
5. **Add a catalog entry** in `integrations/catalog.json`
6. **Update documentation** in `AGENTS.md` and `README.md`
### Catalog Entry Format
Add your integration under the top-level `integrations` key in `integrations/catalog.json`:
```json
{
"schema_version": "1.0",
"integrations": {
"my-agent": {
"id": "my-agent",
"name": "My Agent",
"version": "1.0.0",
"description": "Integration for My Agent",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
}
}
}
```
## Adding a Community Integration
Community integrations are contributed by external developers and listed in `integrations/catalog.community.json` for discovery.
### Prerequisites
1. **Working integration** — tested with `specify integration install`
2. **Public repository** — hosted on GitHub or similar
3. **`integration.yml` descriptor** — valid descriptor file (see below)
4. **Documentation** — README with usage instructions
5. **License** — open source license file
### `integration.yml` Descriptor
Every community integration must include an `integration.yml`:
```yaml
schema_version: "1.0"
integration:
id: "my-agent"
name: "My Agent"
version: "1.0.0"
description: "Integration for My Agent"
author: "your-name"
repository: "https://github.com/your-name/speckit-my-agent"
license: "MIT"
requires:
speckit_version: ">=0.6.0"
tools:
- name: "my-agent"
version: ">=1.0.0"
required: true
provides:
commands:
- name: "speckit.specify"
file: "templates/speckit.specify.md"
scripts:
- update-context.sh
```
### Descriptor Validation Rules
| Field | Rule |
|-------|------|
| `schema_version` | Must be `"1.0"` |
| `integration.id` | Lowercase alphanumeric + hyphens (`^[a-z0-9-]+$`) |
| `integration.version` | Valid PEP 440 version (parsed with `packaging.version.Version()`) |
| `requires.speckit_version` | Required field; specify a version constraint such as `>=0.6.0` (current validation checks presence only) |
| `provides` | Must include at least one command or script |
| `provides.commands[].name` | String identifier |
| `provides.commands[].file` | Relative path to template file |
### Submitting to the Community Catalog
1. **Fork** the [spec-kit repository](https://github.com/github/spec-kit)
2. **Add your entry** under the `integrations` key in `integrations/catalog.community.json`:
```json
{
"schema_version": "1.0",
"integrations": {
"my-agent": {
"id": "my-agent",
"name": "My Agent",
"version": "1.0.0",
"description": "Integration for My Agent",
"author": "your-name",
"repository": "https://github.com/your-name/speckit-my-agent",
"tags": ["cli"]
}
}
}
```
3. **Open a pull request** with:
- Your catalog entry
- Link to your integration repository
- Confirmation that `integration.yml` is valid
### Version Updates
To update your integration version in the catalog:
1. Release a new version of your integration
2. Open a PR updating the `version` field in `catalog.community.json`
3. Ensure backward compatibility or document breaking changes
## Upgrade Workflow
The `specify integration upgrade` command supports diff-aware upgrades:
1. **Hash comparison** — the manifest records SHA-256 hashes of all installed files
2. **Modified file detection** — files changed since installation are flagged
3. **Safe default** — the upgrade blocks if any installed files were modified since installation
4. **Forced reinstall** — passing `--force` overwrites modified files with the latest version
```bash
# Upgrade current integration (blocks if files are modified)
specify integration upgrade
# Force upgrade (overwrites modified files)
specify integration upgrade --force
```

129
integrations/README.md Normal file
View File

@@ -0,0 +1,129 @@
# Spec Kit Integration Catalog
The integration catalog enables discovery, versioning, and distribution of AI agent integrations for Spec Kit.
## Catalog Files
### Built-In Catalog (`catalog.json`)
Contains integrations that ship with Spec Kit. These are maintained by the core team and always installable.
### Community Catalog (`catalog.community.json`)
Community-contributed integrations. Listed for discovery only — users install from the source repositories.
## Catalog Configuration
The catalog stack is resolved in this order (first match wins):
1. **Environment variable** — `SPECKIT_INTEGRATION_CATALOG_URL` overrides all catalogs with a single URL
2. **Project config** — `.specify/integration-catalogs.yml` in the project root
3. **User config** — `~/.specify/integration-catalogs.yml` in the user home directory
4. **Built-in defaults** — `catalog.json` + `catalog.community.json`
Example `integration-catalogs.yml`:
```yaml
catalogs:
- url: "https://example.com/my-catalog.json"
name: "my-catalog"
priority: 1
install_allowed: true
```
## CLI Commands
```bash
# List built-in integrations (default)
specify integration list
# Browse full catalog (built-in + community)
specify integration list --catalog
# Install an integration
specify integration install copilot
# Upgrade the current integration (diff-aware)
specify integration upgrade
# Upgrade with force (overwrite modified files)
specify integration upgrade --force
```
## Integration Descriptor (`integration.yml`)
Each integration can include an `integration.yml` descriptor that documents its metadata, requirements, and provided commands/scripts:
```yaml
schema_version: "1.0"
integration:
id: "my-agent"
name: "My Agent"
version: "1.0.0"
description: "Integration for My Agent"
author: "my-org"
repository: "https://github.com/my-org/speckit-my-agent"
license: "MIT"
requires:
speckit_version: ">=0.6.0"
tools:
- name: "my-agent"
version: ">=1.0.0"
required: true
provides:
commands:
- name: "speckit.specify"
file: "templates/speckit.specify.md"
- name: "speckit.plan"
file: "templates/speckit.plan.md"
scripts:
- update-context.sh
- update-context.ps1
```
## Catalog Schema
Both catalog files follow the same JSON schema:
```json
{
"schema_version": "1.0",
"updated_at": "2026-04-08T00:00:00Z",
"catalog_url": "https://...",
"integrations": {
"my-agent": {
"id": "my-agent",
"name": "My Agent",
"version": "1.0.0",
"description": "Integration for My Agent",
"author": "my-org",
"repository": "https://github.com/my-org/speckit-my-agent",
"tags": ["cli"]
}
}
}
```
### Required Fields
| Field | Type | Description |
|-------|------|-------------|
| `schema_version` | string | Must be `"1.0"` |
| `updated_at` | string | ISO 8601 timestamp |
| `integrations` | object | Map of integration ID → metadata |
### Integration Entry Fields
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `id` | string | Yes | Unique ID (lowercase alphanumeric + hyphens) |
| `name` | string | Yes | Human-readable display name |
| `version` | string | Yes | PEP 440 version (e.g., `1.0.0`, `1.0.0a1`) |
| `description` | string | Yes | One-line description |
| `author` | string | No | Author name or organization |
| `repository` | string | No | Source repository URL |
| `tags` | array | No | Searchable tags (e.g., `["cli", "ide"]`) |
## Contributing
See [CONTRIBUTING.md](CONTRIBUTING.md) for how to add integrations to the community catalog.

6
integrations/catalog.community.json Normal file
View File

@@ -0,0 +1,6 @@
{
"schema_version": "1.0",
"updated_at": "2026-04-08T00:00:00Z",
"catalog_url": "https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.community.json",
"integrations": {}
}

259
integrations/catalog.json Normal file
View File

@@ -0,0 +1,259 @@
{
"schema_version": "1.0",
"updated_at": "2026-04-08T00:00:00Z",
"catalog_url": "https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.json",
"integrations": {
"claude": {
"id": "claude",
"name": "Claude Code",
"version": "1.0.0",
"description": "Anthropic Claude Code CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "anthropic"]
},
"copilot": {
"id": "copilot",
"name": "GitHub Copilot",
"version": "1.0.0",
"description": "GitHub Copilot IDE integration with agent commands and prompt files",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide", "github"]
},
"gemini": {
"id": "gemini",
"name": "Gemini CLI",
"version": "1.0.0",
"description": "Google Gemini CLI integration with TOML command format",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "google"]
},
"cursor-agent": {
"id": "cursor-agent",
"name": "Cursor",
"version": "1.0.0",
"description": "Cursor IDE integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide"]
},
"windsurf": {
"id": "windsurf",
"name": "Windsurf",
"version": "1.0.0",
"description": "Windsurf IDE workflow integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide"]
},
"amp": {
"id": "amp",
"name": "Amp",
"version": "1.0.0",
"description": "Amp CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"codex": {
"id": "codex",
"name": "Codex CLI",
"version": "1.0.0",
"description": "Codex CLI skills-based integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "skills"]
},
"qwen": {
"id": "qwen",
"name": "Qwen Code",
"version": "1.0.0",
"description": "Alibaba Qwen Code CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "alibaba"]
},
"opencode": {
"id": "opencode",
"name": "opencode",
"version": "1.0.0",
"description": "opencode CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"forge": {
"id": "forge",
"name": "Forge",
"version": "1.0.0",
"description": "Forge CLI integration with parameter-based commands",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"kiro-cli": {
"id": "kiro-cli",
"name": "Kiro CLI",
"version": "1.0.0",
"description": "Kiro CLI prompt-based integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"junie": {
"id": "junie",
"name": "Junie",
"version": "1.0.0",
"description": "Junie by JetBrains CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "jetbrains"]
},
"auggie": {
"id": "auggie",
"name": "Auggie CLI",
"version": "1.0.0",
"description": "Auggie CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"shai": {
"id": "shai",
"name": "SHAI",
"version": "1.0.0",
"description": "SHAI CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"tabnine": {
"id": "tabnine",
"name": "Tabnine CLI",
"version": "1.0.0",
"description": "Tabnine CLI integration with TOML command format",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"kilocode": {
"id": "kilocode",
"name": "Kilo Code",
"version": "1.0.0",
"description": "Kilo Code IDE workflow integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide"]
},
"roo": {
"id": "roo",
"name": "Roo Code",
"version": "1.0.0",
"description": "Roo Code IDE integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide"]
},
"bob": {
"id": "bob",
"name": "IBM Bob",
"version": "1.0.0",
"description": "IBM Bob IDE integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide", "ibm"]
},
"trae": {
"id": "trae",
"name": "Trae",
"version": "1.0.0",
"description": "Trae IDE rules-based integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide"]
},
"codebuddy": {
"id": "codebuddy",
"name": "CodeBuddy",
"version": "1.0.0",
"description": "CodeBuddy CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"qodercli": {
"id": "qodercli",
"name": "Qoder CLI",
"version": "1.0.0",
"description": "Qoder CLI integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"kimi": {
"id": "kimi",
"name": "Kimi Code",
"version": "1.0.0",
"description": "Kimi Code CLI skills-based integration by Moonshot AI",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "skills"]
},
"pi": {
"id": "pi",
"name": "Pi Coding Agent",
"version": "1.0.0",
"description": "Pi terminal coding agent prompt-based integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"iflow": {
"id": "iflow",
"name": "iFlow CLI",
"version": "1.0.0",
"description": "iFlow CLI integration by iflow-ai",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
},
"vibe": {
"id": "vibe",
"name": "Mistral Vibe",
"version": "1.0.0",
"description": "Mistral Vibe CLI prompt-based integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli", "mistral"]
},
"agy": {
"id": "agy",
"name": "Antigravity",
"version": "1.0.0",
"description": "Antigravity IDE skills-based integration",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["ide", "skills"]
},
"generic": {
"id": "generic",
"name": "Generic (bring your own agent)",
"version": "1.0.0",
"description": "Generic integration for any agent via --ai-commands-dir",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["generic"]
},
"goose": {
"id": "goose",
"name": "Goose",
"version": "1.0.0",
"description": "Goose CLI integration with YAML recipe format",
"author": "spec-kit-core",
"repository": "https://github.com/github/spec-kit",
"tags": ["cli"]
}
}
}

162
src/specify_cli/__init__.py
View File

@@ -1775,7 +1775,9 @@ def _resolve_script_type(project_root: Path, script_type: str | None) -> str:
@integration_app.command("list")
def integration_list():
def integration_list(
catalog: bool = typer.Option(False, "--catalog", help="Browse full catalog (built-in + community)"),
):
"""List available integrations and installed status."""
from .integrations import INTEGRATION_REGISTRY
@@ -1790,6 +1792,50 @@ def integration_list():
current = _read_integration_json(project_root)
installed_key = current.get("integration")
if catalog:
from .integrations.catalog import IntegrationCatalog, IntegrationCatalogError
ic = IntegrationCatalog(project_root)
try:
entries = ic.search()
except IntegrationCatalogError as exc:
console.print(f"[red]Error:[/red] {exc}")
raise typer.Exit(1)
if not entries:
console.print("[yellow]No integrations found in catalog.[/yellow]")
return
table = Table(title="Integration Catalog")
table.add_column("ID", style="cyan")
table.add_column("Name")
table.add_column("Version")
table.add_column("Source")
table.add_column("Status")
for entry in sorted(entries, key=lambda e: e["id"]):
eid = entry["id"]
cat_name = entry.get("_catalog_name", "")
install_allowed = entry.get("_install_allowed", True)
if eid == installed_key:
status = "[green]installed[/green]"
elif eid in INTEGRATION_REGISTRY:
status = "built-in"
elif install_allowed is False:
status = "discovery-only"
else:
status = ""
table.add_row(
eid,
entry.get("name", eid),
entry.get("version", ""),
cat_name,
status,
)
console.print(table)
return
table = Table(title="AI Agent Integrations")
table.add_column("Key", style="cyan")
table.add_column("Name")
@@ -2176,6 +2222,120 @@ def integration_switch(
console.print(f"\n[green]✓[/green] Switched to integration '{name}'")
@integration_app.command("upgrade")
def integration_upgrade(
key: str | None = typer.Argument(None, help="Integration key to upgrade (default: current integration)"),
force: bool = typer.Option(False, "--force", help="Force upgrade even if files are modified"),
script: str | None = typer.Option(None, "--script", help="Script type: sh or ps (default: from init-options.json or platform default)"),
integration_options: str | None = typer.Option(None, "--integration-options", help="Options for the integration"),
):
"""Upgrade an integration by reinstalling with diff-aware file handling.
Compares manifest hashes to detect locally modified files and
blocks the upgrade unless --force is used.
"""
from .integrations import get_integration
from .integrations.manifest import IntegrationManifest
project_root = Path.cwd()
specify_dir = project_root / ".specify"
if not specify_dir.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)
current = _read_integration_json(project_root)
installed_key = current.get("integration")
if key is None:
if not installed_key:
console.print("[yellow]No integration is currently installed.[/yellow]")
raise typer.Exit(0)
key = installed_key
if installed_key and installed_key != key:
console.print(
f"[red]Error:[/red] Integration '{key}' is not the currently installed integration ('{installed_key}')."
)
console.print(f"Use [cyan]specify integration switch {key}[/cyan] instead.")
raise typer.Exit(1)
integration = get_integration(key)
if integration is None:
console.print(f"[red]Error:[/red] Unknown integration '{key}'")
raise typer.Exit(1)
manifest_path = project_root / ".specify" / "integrations" / f"{key}.manifest.json"
if not manifest_path.exists():
console.print(f"[yellow]No manifest found for integration '{key}'. Nothing to upgrade.[/yellow]")
console.print(f"Run [cyan]specify integration install {key}[/cyan] to perform a fresh install.")
raise typer.Exit(0)
try:
old_manifest = IntegrationManifest.load(key, project_root)
except (ValueError, FileNotFoundError) as exc:
console.print(f"[red]Error:[/red] Integration manifest for '{key}' is unreadable: {exc}")
raise typer.Exit(1)
# Detect modified files via manifest hashes
modified = old_manifest.check_modified()
if modified and not force:
console.print(f"[yellow]⚠[/yellow] {len(modified)} file(s) have been modified since installation:")
for rel in modified:
console.print(f" {rel}")
console.print("\nUse [cyan]--force[/cyan] to overwrite modified files, or resolve manually.")
raise typer.Exit(1)
selected_script = _resolve_script_type(project_root, script)
# Ensure shared infrastructure is present (safe to run unconditionally;
# _install_shared_infra merges missing files without overwriting).
_install_shared_infra(project_root, selected_script)
if os.name != "nt":
ensure_executable_scripts(project_root)
# Phase 1: Install new files (overwrites existing; old-only files remain)
console.print(f"Upgrading integration: [cyan]{key}[/cyan]")
new_manifest = IntegrationManifest(key, project_root, version=get_speckit_version())
parsed_options: dict[str, Any] | None = None
if integration_options:
parsed_options = _parse_integration_options(integration, integration_options)
try:
integration.setup(
project_root,
new_manifest,
parsed_options=parsed_options,
script_type=selected_script,
raw_options=integration_options,
)
new_manifest.save()
_write_integration_json(project_root, key, selected_script)
_update_init_options_for_integration(project_root, integration, script_type=selected_script)
except Exception as exc:
# Don't teardown — setup overwrites in-place, so teardown would
# delete files that were working before the upgrade. Just report.
console.print(f"[red]Error:[/red] Failed to upgrade integration: {exc}")
console.print("[yellow]The previous integration files may still be in place.[/yellow]")
raise typer.Exit(1)
# Phase 2: Remove stale files from old manifest that are not in the new one
old_files = old_manifest.files
new_files = new_manifest.files
stale_keys = set(old_files) - set(new_files)
if stale_keys:
stale_manifest = IntegrationManifest(key, project_root, version="stale-cleanup")
stale_manifest._files = {k: old_files[k] for k in stale_keys}
stale_removed, _ = stale_manifest.uninstall(project_root, force=True)
if stale_removed:
console.print(f" Removed {len(stale_removed)} stale file(s) from previous install")
name = (integration.config or {}).get("name", key)
console.print(f"\n[green]✓[/green] Integration '{name}' upgraded successfully")
# ===== Preset Commands =====

626
src/specify_cli/integrations/catalog.py Normal file
View File

@@ -0,0 +1,626 @@
"""Integration catalog — discovery, validation, and upgrade support.
Provides:
- ``IntegrationCatalogEntry`` — single catalog source metadata.
- ``IntegrationCatalog`` — fetches, caches, and searches integration
catalogs (built-in + community).
- ``IntegrationDescriptor`` — loads and validates ``integration.yml``.
"""
from __future__ import annotations
import hashlib
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
import yaml
from packaging import version as pkg_version
# ---------------------------------------------------------------------------
# Errors
# ---------------------------------------------------------------------------
class IntegrationCatalogError(Exception):
"""Raised when a catalog operation fails."""
class IntegrationDescriptorError(Exception):
"""Raised when an integration.yml descriptor is invalid."""
# ---------------------------------------------------------------------------
# IntegrationCatalogEntry
# ---------------------------------------------------------------------------
@dataclass
class IntegrationCatalogEntry:
"""Represents a single catalog source in the catalog stack."""
url: str
name: str
priority: int
install_allowed: bool
description: str = ""
# ---------------------------------------------------------------------------
# IntegrationCatalog
# ---------------------------------------------------------------------------
class IntegrationCatalog:
"""Manages integration catalog fetching, caching, and searching."""
DEFAULT_CATALOG_URL = (
"https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.json"
)
COMMUNITY_CATALOG_URL = (
"https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.community.json"
)
CACHE_DURATION = 3600 # 1 hour
def __init__(self, project_root: Path) -> None:
self.project_root = project_root
self.cache_dir = project_root / ".specify" / "integrations" / ".cache"
# -- URL validation ---------------------------------------------------
@staticmethod
def _validate_catalog_url(url: str) -> None:
from urllib.parse import urlparse
parsed = urlparse(url)
is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost):
raise IntegrationCatalogError(
f"Catalog URL must use HTTPS (got {parsed.scheme}://). "
"HTTP is only allowed for localhost."
)
if not parsed.netloc:
raise IntegrationCatalogError(
"Catalog URL must be a valid URL with a host."
)
# -- Catalog stack ----------------------------------------------------
def _load_catalog_config(
self, config_path: Path
) -> Optional[List[IntegrationCatalogEntry]]:
"""Load catalog stack from a YAML file.
Returns None when the file does not exist.
Raises:
IntegrationCatalogError: on invalid content
"""
if not config_path.exists():
return None
try:
data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
except (yaml.YAMLError, OSError, UnicodeError) as exc:
raise IntegrationCatalogError(
f"Failed to read catalog config {config_path}: {exc}"
)
if not isinstance(data, dict):
raise IntegrationCatalogError(
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, "
f"got {type(catalogs_data).__name__}"
)
if not catalogs_data:
raise IntegrationCatalogError(
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."
)
entries: List[IntegrationCatalogEntry] = []
skipped: List[int] = []
for idx, item in enumerate(catalogs_data):
if not isinstance(item, dict):
raise IntegrationCatalogError(
f"Invalid 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(
f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
f"expected integer, got {item.get('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)
entries.append(
IntegrationCatalogEntry(
url=url,
name=str(item.get("name", f"catalog-{idx + 1}")),
priority=priority,
install_allowed=install_allowed,
description=str(item.get("description", "")),
)
)
entries.sort(key=lambda e: e.priority)
if not entries:
raise IntegrationCatalogError(
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."
)
return entries
def get_active_catalogs(self) -> List[IntegrationCatalogEntry]:
"""Return the ordered list of active integration catalogs.
Resolution:
1. ``SPECKIT_INTEGRATION_CATALOG_URL`` env var
2. Project ``.specify/integration-catalogs.yml``
3. User ``~/.specify/integration-catalogs.yml``
4. Built-in defaults (built-in + community)
"""
import sys
env_value = os.environ.get("SPECKIT_INTEGRATION_CATALOG_URL", "").strip()
if env_value:
self._validate_catalog_url(env_value)
if env_value != self.DEFAULT_CATALOG_URL:
if not getattr(self, "_non_default_catalog_warning_shown", False):
print(
"Warning: Using non-default integration catalog. "
"Only use catalogs from sources you trust.",
file=sys.stderr,
)
self._non_default_catalog_warning_shown = True
return [
IntegrationCatalogEntry(
url=env_value,
name="custom",
priority=1,
install_allowed=True,
description="Custom catalog via SPECKIT_INTEGRATION_CATALOG_URL",
)
]
project_cfg = self.project_root / ".specify" / "integration-catalogs.yml"
catalogs = self._load_catalog_config(project_cfg)
if catalogs is not None:
return catalogs
user_cfg = Path.home() / ".specify" / "integration-catalogs.yml"
catalogs = self._load_catalog_config(user_cfg)
if catalogs is not None:
return catalogs
return [
IntegrationCatalogEntry(
url=self.DEFAULT_CATALOG_URL,
name="default",
priority=1,
install_allowed=True,
description="Built-in catalog of installable integrations",
),
IntegrationCatalogEntry(
url=self.COMMUNITY_CATALOG_URL,
name="community",
priority=2,
install_allowed=False,
description="Community-contributed integrations (discovery only)",
),
]
# -- Fetching ---------------------------------------------------------
def _fetch_single_catalog(
self,
entry: IntegrationCatalogEntry,
force_refresh: bool = False,
) -> Dict[str, Any]:
"""Fetch one catalog, with per-URL caching."""
import urllib.error
import urllib.request
url_hash = hashlib.sha256(entry.url.encode()).hexdigest()[:16]
cache_file = self.cache_dir / f"catalog-{url_hash}.json"
cache_meta = self.cache_dir / f"catalog-{url_hash}-metadata.json"
if not force_refresh and cache_file.exists() and cache_meta.exists():
try:
meta = json.loads(cache_meta.read_text(encoding="utf-8"))
cached_at = datetime.fromisoformat(meta.get("cached_at", ""))
if cached_at.tzinfo is None:
cached_at = cached_at.replace(tzinfo=timezone.utc)
age = (datetime.now(timezone.utc) - cached_at).total_seconds()
if age < self.CACHE_DURATION:
return json.loads(cache_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, ValueError, KeyError, TypeError, AttributeError, OSError, UnicodeError):
# Cache is invalid or stale metadata; delete and refetch from source.
try:
cache_file.unlink(missing_ok=True)
cache_meta.unlink(missing_ok=True)
except OSError:
pass # Cache cleanup is best-effort; ignore deletion failures.
try:
with urllib.request.urlopen(entry.url, timeout=10) as resp:
# Validate final URL after redirects
final_url = resp.geturl()
if final_url != entry.url:
self._validate_catalog_url(final_url)
catalog_data = json.loads(resp.read())
if not isinstance(catalog_data, dict):
raise IntegrationCatalogError(
f"Invalid catalog format from {entry.url}: expected a JSON object"
)
if (
"schema_version" not in catalog_data
or "integrations" not in catalog_data
):
raise IntegrationCatalogError(
f"Invalid catalog format from {entry.url}"
)
if not isinstance(catalog_data.get("integrations"), dict):
raise IntegrationCatalogError(
f"Invalid catalog format from {entry.url}: 'integrations' must be a JSON object"
)
try:
self.cache_dir.mkdir(parents=True, exist_ok=True)
cache_file.write_text(json.dumps(catalog_data, indent=2), encoding="utf-8")
cache_meta.write_text(
json.dumps(
{
"cached_at": datetime.now(timezone.utc).isoformat(),
"catalog_url": entry.url,
},
indent=2,
),
encoding="utf-8",
)
except OSError:
pass # Cache is best-effort; proceed with fetched data
return catalog_data
except urllib.error.URLError as exc:
raise IntegrationCatalogError(
f"Failed to fetch catalog from {entry.url}: {exc}"
)
except json.JSONDecodeError as exc:
raise IntegrationCatalogError(
f"Invalid JSON in catalog from {entry.url}: {exc}"
)
def _get_merged_integrations(
self, force_refresh: bool = False
) -> List[Dict[str, Any]]:
"""Fetch and merge integrations from all active catalogs.
Catalogs are processed in the order returned by
:meth:`get_active_catalogs`. On conflicts, the first catalog in that
order wins (lower numeric priority = higher precedence). Each dict is
annotated with ``_catalog_name`` and ``_install_allowed``.
"""
import sys
active = self.get_active_catalogs()
merged: Dict[str, Dict[str, Any]] = {}
any_success = False
for entry in active:
try:
data = self._fetch_single_catalog(entry, force_refresh)
any_success = True
except IntegrationCatalogError as exc:
print(
f"Warning: Could not fetch catalog '{entry.name}': {exc}",
file=sys.stderr,
)
continue
for integ_id, integ_data in data.get("integrations", {}).items():
if not isinstance(integ_data, dict):
continue
if integ_id not in merged:
merged[integ_id] = {
**integ_data,
"id": integ_id,
"_catalog_name": entry.name,
"_install_allowed": entry.install_allowed,
}
if not any_success and active:
raise IntegrationCatalogError(
"Failed to fetch any integration catalog"
)
return list(merged.values())
# -- Search / info ----------------------------------------------------
def search(
self,
query: Optional[str] = None,
tag: Optional[str] = None,
author: Optional[str] = None,
) -> List[Dict[str, Any]]:
"""Search catalogs for integrations matching the given filters."""
results: List[Dict[str, Any]] = []
for item in self._get_merged_integrations():
author_val = item.get("author", "")
if not isinstance(author_val, str):
author_val = str(author_val) if author_val is not None else ""
if author and author_val.lower() != author.lower():
continue
if tag:
raw_tags = item.get("tags", [])
tags_list = raw_tags if isinstance(raw_tags, list) else []
if tag.lower() not in [t.lower() for t in tags_list if isinstance(t, str)]:
continue
if query:
raw_tags = item.get("tags", [])
tags_list = raw_tags if isinstance(raw_tags, list) else []
name_val = item.get("name", "")
desc_val = item.get("description", "")
id_val = item.get("id", "")
haystack = " ".join(
[
str(name_val) if name_val else "",
str(desc_val) if desc_val else "",
str(id_val) if id_val else "",
]
+ [t for t in tags_list if isinstance(t, str)]
).lower()
if query.lower() not in haystack:
continue
results.append(item)
return results
def get_integration_info(
self, integration_id: str
) -> Optional[Dict[str, Any]]:
"""Return catalog metadata for a single integration, or None."""
for item in self._get_merged_integrations():
if item["id"] == integration_id:
return item
return None
# -- Cache management -------------------------------------------------
def clear_cache(self) -> None:
"""Remove all cached catalog files."""
if self.cache_dir.exists():
for pattern in ("catalog-*.json", "catalog-*-metadata.json"):
for f in self.cache_dir.glob(pattern):
f.unlink(missing_ok=True)
# ---------------------------------------------------------------------------
# IntegrationDescriptor (integration.yml)
# ---------------------------------------------------------------------------
class IntegrationDescriptor:
"""Loads and validates an ``integration.yml`` descriptor.
The descriptor mirrors ``extension.yml`` and ``preset.yml``::
schema_version: "1.0"
integration:
id: "my-agent"
name: "My Agent"
version: "1.0.0"
description: "Integration for My Agent"
author: "my-org"
requires:
speckit_version: ">=0.6.0"
tools: [...]
provides:
commands: [...]
scripts: [...]
"""
SCHEMA_VERSION = "1.0"
REQUIRED_TOP_LEVEL = ["schema_version", "integration", "requires", "provides"]
def __init__(self, descriptor_path: Path) -> None:
self.path = descriptor_path
self.data = self._load(descriptor_path)
self._validate()
# -- Loading ----------------------------------------------------------
@staticmethod
def _load(path: Path) -> dict:
try:
with open(path, "r", encoding="utf-8") as fh:
return yaml.safe_load(fh) or {}
except yaml.YAMLError as exc:
raise IntegrationDescriptorError(f"Invalid YAML in {path}: {exc}")
except FileNotFoundError:
raise IntegrationDescriptorError(f"Descriptor not found: {path}")
except (OSError, UnicodeError) as exc:
raise IntegrationDescriptorError(
f"Unable to read descriptor {path}: {exc}"
)
# -- Validation -------------------------------------------------------
def _validate(self) -> None:
if not isinstance(self.data, dict):
raise IntegrationDescriptorError(
f"Descriptor root must be a YAML mapping, got {type(self.data).__name__}"
)
for field in self.REQUIRED_TOP_LEVEL:
if field not in self.data:
raise IntegrationDescriptorError(
f"Missing required field: {field}"
)
if self.data["schema_version"] != self.SCHEMA_VERSION:
raise IntegrationDescriptorError(
f"Unsupported schema version: {self.data['schema_version']} "
f"(expected {self.SCHEMA_VERSION})"
)
integ = self.data["integration"]
if not isinstance(integ, dict):
raise IntegrationDescriptorError(
"'integration' must be a mapping"
)
for field in ("id", "name", "version", "description"):
if field not in integ:
raise IntegrationDescriptorError(
f"Missing integration.{field}"
)
if not isinstance(integ[field], str):
raise IntegrationDescriptorError(
f"integration.{field} must be a string, got {type(integ[field]).__name__}"
)
if not re.match(r"^[a-z0-9-]+$", integ["id"]):
raise IntegrationDescriptorError(
f"Invalid integration ID '{integ['id']}': "
"must be lowercase alphanumeric with hyphens only"
)
try:
pkg_version.Version(integ["version"])
except (pkg_version.InvalidVersion, TypeError):
raise IntegrationDescriptorError(
f"Invalid version '{integ['version']}'"
)
requires = self.data["requires"]
if not isinstance(requires, dict):
raise IntegrationDescriptorError(
"'requires' must be a mapping"
)
if "speckit_version" not in requires:
raise IntegrationDescriptorError(
"Missing requires.speckit_version"
)
if not isinstance(requires["speckit_version"], str) or not requires["speckit_version"].strip():
raise IntegrationDescriptorError(
"requires.speckit_version must be a non-empty string"
)
tools = requires.get("tools")
if tools is not None:
if not isinstance(tools, list):
raise IntegrationDescriptorError(
"requires.tools must be a list"
)
for tool in tools:
if not isinstance(tool, dict):
raise IntegrationDescriptorError(
"Each requires.tools entry must be a mapping"
)
tool_name = tool.get("name")
if not isinstance(tool_name, str) or not tool_name.strip():
raise IntegrationDescriptorError(
"requires.tools entry 'name' must be a non-empty string"
)
provides = self.data["provides"]
if not isinstance(provides, dict):
raise IntegrationDescriptorError(
"'provides' must be a mapping"
)
commands = provides.get("commands", [])
scripts = provides.get("scripts", [])
if "commands" in provides and not isinstance(commands, list):
raise IntegrationDescriptorError(
"Invalid provides.commands: expected a list"
)
if "scripts" in provides and not isinstance(scripts, list):
raise IntegrationDescriptorError(
"Invalid provides.scripts: expected a list"
)
if not commands and not scripts:
raise IntegrationDescriptorError(
"Integration must provide at least one command or script"
)
for cmd in commands:
if not isinstance(cmd, dict):
raise IntegrationDescriptorError(
"Each command entry must be a mapping"
)
if "name" not in cmd or "file" not in cmd:
raise IntegrationDescriptorError(
"Command entry missing 'name' or 'file'"
)
cmd_name = cmd["name"]
cmd_file = cmd["file"]
if not isinstance(cmd_name, str) or not cmd_name.strip():
raise IntegrationDescriptorError(
"Command entry 'name' must be a non-empty string"
)
if not isinstance(cmd_file, str) or not cmd_file.strip():
raise IntegrationDescriptorError(
"Command entry 'file' must be a non-empty string"
)
if os.path.isabs(cmd_file) or ".." in Path(cmd_file).parts or Path(cmd_file).drive or Path(cmd_file).anchor:
raise IntegrationDescriptorError(
f"Command entry 'file' must be a relative path without '..': {cmd_file}"
)
for script_entry in scripts:
if not isinstance(script_entry, str) or not script_entry.strip():
raise IntegrationDescriptorError(
"Script entry must be a non-empty string"
)
if os.path.isabs(script_entry) or ".." in Path(script_entry).parts or Path(script_entry).drive or Path(script_entry).anchor:
raise IntegrationDescriptorError(
f"Script entry must be a relative path without '..': {script_entry}"
)
# -- Property accessors -----------------------------------------------
@property
def id(self) -> str:
return self.data["integration"]["id"]
@property
def name(self) -> str:
return self.data["integration"]["name"]
@property
def version(self) -> str:
return self.data["integration"]["version"]
@property
def description(self) -> str:
return self.data["integration"]["description"]
@property
def requires_speckit_version(self) -> str:
return self.data["requires"]["speckit_version"]
@property
def commands(self) -> List[Dict[str, Any]]:
return self.data.get("provides", {}).get("commands", [])
@property
def scripts(self) -> List[str]:
return self.data.get("provides", {}).get("scripts", [])
@property
def tools(self) -> List[Dict[str, Any]]:
return self.data.get("requires", {}).get("tools") or []
def get_hash(self) -> str:
"""SHA-256 hash of the descriptor file."""
with open(self.path, "rb") as fh:
return f"sha256:{hashlib.sha256(fh.read()).hexdigest()}"

656
tests/integrations/test_integration_catalog.py Normal file
View File

@@ -0,0 +1,656 @@
"""Tests for the integration catalog system (catalog.py)."""
import json
import os
import pytest
import yaml
from specify_cli.integrations.catalog import (
IntegrationCatalog,
IntegrationCatalogEntry,
IntegrationCatalogError,
IntegrationDescriptor,
IntegrationDescriptorError,
)
# ---------------------------------------------------------------------------
# IntegrationCatalogEntry
# ---------------------------------------------------------------------------
class TestIntegrationCatalogEntry:
def test_create_entry(self):
entry = IntegrationCatalogEntry(
url="https://example.com/catalog.json",
name="test",
priority=1,
install_allowed=True,
description="Test catalog",
)
assert entry.url == "https://example.com/catalog.json"
assert entry.name == "test"
assert entry.priority == 1
assert entry.install_allowed is True
assert entry.description == "Test catalog"
def test_default_description(self):
entry = IntegrationCatalogEntry(
url="https://example.com/catalog.json",
name="test",
priority=1,
install_allowed=False,
)
assert entry.description == ""
# ---------------------------------------------------------------------------
# IntegrationCatalog — URL validation
# ---------------------------------------------------------------------------
class TestCatalogURLValidation:
def test_https_allowed(self):
IntegrationCatalog._validate_catalog_url("https://example.com/catalog.json")
def test_http_rejected(self):
with pytest.raises(IntegrationCatalogError, match="HTTPS"):
IntegrationCatalog._validate_catalog_url("http://example.com/catalog.json")
def test_http_localhost_allowed(self):
IntegrationCatalog._validate_catalog_url("http://localhost:8080/catalog.json")
IntegrationCatalog._validate_catalog_url("http://127.0.0.1/catalog.json")
def test_missing_host_rejected(self):
with pytest.raises(IntegrationCatalogError, match="valid URL"):
IntegrationCatalog._validate_catalog_url("https:///no-host")
# ---------------------------------------------------------------------------
# IntegrationCatalog — active catalogs
# ---------------------------------------------------------------------------
class TestActiveCatalogs:
def test_defaults_when_no_config(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
active = cat.get_active_catalogs()
assert len(active) == 2
assert active[0].name == "default"
assert active[1].name == "community"
def test_env_var_override(self, tmp_path, monkeypatch):
(tmp_path / ".specify").mkdir()
monkeypatch.setenv(
"SPECKIT_INTEGRATION_CATALOG_URL",
"https://custom.example.com/catalog.json",
)
cat = IntegrationCatalog(tmp_path)
active = cat.get_active_catalogs()
assert len(active) == 1
assert active[0].name == "custom"
def test_project_config_overrides_defaults(self, tmp_path):
specify = tmp_path / ".specify"
specify.mkdir()
cfg = specify / "integration-catalogs.yml"
cfg.write_text(yaml.dump({
"catalogs": [
{"url": "https://my.example.com/cat.json", "name": "mine", "priority": 1, "install_allowed": True},
]
}))
cat = IntegrationCatalog(tmp_path)
active = cat.get_active_catalogs()
assert len(active) == 1
assert active[0].name == "mine"
def test_empty_config_raises(self, tmp_path):
specify = tmp_path / ".specify"
specify.mkdir()
cfg = specify / "integration-catalogs.yml"
cfg.write_text(yaml.dump({"catalogs": []}))
cat = IntegrationCatalog(tmp_path)
with pytest.raises(IntegrationCatalogError, match="no 'catalogs' entries"):
cat.get_active_catalogs()
# ---------------------------------------------------------------------------
# IntegrationCatalog — fetch & search (using monkeypatched urlopen responses)
# ---------------------------------------------------------------------------
class TestCatalogFetch:
"""Tests that use a local HTTP server stub via monkeypatch."""
def _patch_urlopen(self, monkeypatch, catalog_data):
"""Patch urllib.request.urlopen to return *catalog_data*."""
class FakeResponse:
def __init__(self, data, url=""):
self._data = json.dumps(data).encode()
self._url = url
def read(self):
return self._data
def geturl(self):
return self._url
def __enter__(self):
return self
def __exit__(self, *a):
pass
def fake_urlopen(url, timeout=10):
return FakeResponse(catalog_data, url)
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
def test_fetch_and_search_all(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
catalog = {
"schema_version": "1.0",
"updated_at": "2026-01-01T00:00:00Z",
"integrations": {
"acme-coder": {
"id": "acme-coder",
"name": "Acme Coder",
"version": "2.0.0",
"description": "Community integration for Acme Coder",
"author": "acme-org",
"tags": ["cli"],
},
},
}
self._patch_urlopen(monkeypatch, catalog)
results = cat.search()
assert len(results) >= 1
ids = [r["id"] for r in results]
assert "acme-coder" in ids
def test_search_by_tag(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
catalog = {
"schema_version": "1.0",
"updated_at": "2026-01-01T00:00:00Z",
"integrations": {
"a": {"id": "a", "name": "A", "version": "1.0.0", "tags": ["cli"]},
"b": {"id": "b", "name": "B", "version": "1.0.0", "tags": ["ide"]},
},
}
self._patch_urlopen(monkeypatch, catalog)
results = cat.search(tag="cli")
assert all("cli" in r.get("tags", []) for r in results)
def test_search_by_query(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
catalog = {
"schema_version": "1.0",
"updated_at": "2026-01-01T00:00:00Z",
"integrations": {
"claude": {"id": "claude", "name": "Claude Code", "version": "1.0.0", "description": "Anthropic", "tags": []},
"gemini": {"id": "gemini", "name": "Gemini CLI", "version": "1.0.0", "description": "Google", "tags": []},
},
}
self._patch_urlopen(monkeypatch, catalog)
results = cat.search(query="claude")
assert len(results) == 1
assert results[0]["id"] == "claude"
def test_get_integration_info(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
catalog = {
"schema_version": "1.0",
"updated_at": "2026-01-01T00:00:00Z",
"integrations": {
"claude": {"id": "claude", "name": "Claude Code", "version": "1.0.0"},
},
}
self._patch_urlopen(monkeypatch, catalog)
info = cat.get_integration_info("claude")
assert info is not None
assert info["name"] == "Claude Code"
assert cat.get_integration_info("nonexistent") is None
def test_invalid_catalog_format(self, tmp_path, monkeypatch):
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path))
monkeypatch.delenv("SPECKIT_INTEGRATION_CATALOG_URL", raising=False)
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
self._patch_urlopen(monkeypatch, {"schema_version": "1.0"}) # missing "integrations"
with pytest.raises(IntegrationCatalogError, match="Failed to fetch any integration catalog"):
cat.search()
def test_clear_cache(self, tmp_path):
(tmp_path / ".specify").mkdir()
cat = IntegrationCatalog(tmp_path)
cat.cache_dir.mkdir(parents=True, exist_ok=True)
(cat.cache_dir / "catalog-abc123.json").write_text("{}")
cat.clear_cache()
assert not list(cat.cache_dir.glob("catalog-*.json"))
# ---------------------------------------------------------------------------
# IntegrationDescriptor (integration.yml)
# ---------------------------------------------------------------------------
VALID_DESCRIPTOR = {
"schema_version": "1.0",
"integration": {
"id": "my-agent",
"name": "My Agent",
"version": "1.0.0",
"description": "Integration for My Agent",
"author": "my-org",
},
"requires": {
"speckit_version": ">=0.6.0",
},
"provides": {
"commands": [
{"name": "speckit.specify", "file": "templates/speckit.specify.md"},
],
"scripts": ["update-context.sh"],
},
}
class TestIntegrationDescriptor:
def _write(self, tmp_path, data):
p = tmp_path / "integration.yml"
p.write_text(yaml.dump(data))
return p
def test_valid_descriptor(self, tmp_path):
p = self._write(tmp_path, VALID_DESCRIPTOR)
desc = IntegrationDescriptor(p)
assert desc.id == "my-agent"
assert desc.name == "My Agent"
assert desc.version == "1.0.0"
assert desc.description == "Integration for My Agent"
assert desc.requires_speckit_version == ">=0.6.0"
assert len(desc.commands) == 1
assert desc.scripts == ["update-context.sh"]
def test_missing_schema_version(self, tmp_path):
data = {**VALID_DESCRIPTOR}
del data["schema_version"]
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="Missing required field: schema_version"):
IntegrationDescriptor(p)
def test_unsupported_schema_version(self, tmp_path):
data = {**VALID_DESCRIPTOR, "schema_version": "99.0"}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="Unsupported schema version"):
IntegrationDescriptor(p)
def test_missing_integration_id(self, tmp_path):
data = {**VALID_DESCRIPTOR, "integration": {"name": "X", "version": "1.0.0", "description": "Y"}}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="Missing integration.id"):
IntegrationDescriptor(p)
def test_invalid_id_format(self, tmp_path):
integ = {**VALID_DESCRIPTOR["integration"], "id": "BAD_ID"}
data = {**VALID_DESCRIPTOR, "integration": integ}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="Invalid integration ID"):
IntegrationDescriptor(p)
def test_invalid_version(self, tmp_path):
integ = {**VALID_DESCRIPTOR["integration"], "version": "not-semver"}
data = {**VALID_DESCRIPTOR, "integration": integ}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="Invalid version"):
IntegrationDescriptor(p)
def test_missing_speckit_version(self, tmp_path):
data = {**VALID_DESCRIPTOR, "requires": {}}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="requires.speckit_version"):
IntegrationDescriptor(p)
def test_no_commands_or_scripts(self, tmp_path):
data = {**VALID_DESCRIPTOR, "provides": {}}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="at least one command or script"):
IntegrationDescriptor(p)
def test_command_missing_name(self, tmp_path):
data = {**VALID_DESCRIPTOR, "provides": {"commands": [{"file": "x.md"}]}}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="missing 'name' or 'file'"):
IntegrationDescriptor(p)
def test_commands_not_a_list(self, tmp_path):
data = {**VALID_DESCRIPTOR, "provides": {"commands": "not-a-list", "scripts": ["a.sh"]}}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="expected a list"):
IntegrationDescriptor(p)
def test_scripts_not_a_list(self, tmp_path):
data = {**VALID_DESCRIPTOR, "provides": {"commands": [{"name": "a", "file": "b"}], "scripts": "not-a-list"}}
p = self._write(tmp_path, data)
with pytest.raises(IntegrationDescriptorError, match="expected a list"):
IntegrationDescriptor(p)
def test_file_not_found(self, tmp_path):
with pytest.raises(IntegrationDescriptorError, match="Descriptor not found"):
IntegrationDescriptor(tmp_path / "nonexistent.yml")
def test_invalid_yaml(self, tmp_path):
p = tmp_path / "integration.yml"
p.write_text(": : :")
with pytest.raises(IntegrationDescriptorError, match="Invalid YAML"):
IntegrationDescriptor(p)
def test_get_hash(self, tmp_path):
p = self._write(tmp_path, VALID_DESCRIPTOR)
desc = IntegrationDescriptor(p)
h = desc.get_hash()
assert h.startswith("sha256:")
def test_tools_accessor(self, tmp_path):
data = {**VALID_DESCRIPTOR, "requires": {
"speckit_version": ">=0.6.0",
"tools": [{"name": "my-agent", "version": ">=1.0.0", "required": True}],
}}
p = self._write(tmp_path, data)
desc = IntegrationDescriptor(p)
assert len(desc.tools) == 1
assert desc.tools[0]["name"] == "my-agent"
# ---------------------------------------------------------------------------
# CLI: integration list --catalog
# ---------------------------------------------------------------------------
class TestIntegrationListCatalog:
"""Test ``specify integration list --catalog``."""
def _init_project(self, tmp_path):
"""Create a minimal spec-kit project."""
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = tmp_path / "proj"
project.mkdir()
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, [
"init", "--here",
"--integration", "copilot",
"--script", "sh",
"--no-git",
"--ignore-agent-tools",
], catch_exceptions=False)
finally:
os.chdir(old)
assert result.exit_code == 0, result.output
return project
def test_list_catalog_flag(self, tmp_path, monkeypatch):
"""--catalog should show catalog entries."""
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path)
catalog = {
"schema_version": "1.0",
"updated_at": "2026-01-01T00:00:00Z",
"integrations": {
"test-agent": {
"id": "test-agent",
"name": "Test Agent",
"version": "1.0.0",
"description": "A test agent",
"tags": ["cli"],
},
},
}
import urllib.request
class FakeResponse:
def __init__(self, data, url=""):
self._data = json.dumps(data).encode()
self._url = url
def read(self):
return self._data
def geturl(self):
return self._url
def __enter__(self):
return self
def __exit__(self, *a):
pass
monkeypatch.setattr(urllib.request, "urlopen", lambda url, timeout=10: FakeResponse(catalog, url))
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "list", "--catalog"])
finally:
os.chdir(old)
assert result.exit_code == 0
assert "test-agent" in result.output
assert "Test Agent" in result.output
def test_list_without_catalog_still_works(self, tmp_path):
"""Default list (no --catalog) works as before."""
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path)
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "list"])
finally:
os.chdir(old)
assert result.exit_code == 0
assert "copilot" in result.output
assert "installed" in result.output
# ---------------------------------------------------------------------------
# CLI: integration upgrade
# ---------------------------------------------------------------------------
class TestIntegrationUpgrade:
"""Test ``specify integration upgrade``."""
def _init_project(self, tmp_path, integration="copilot"):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = tmp_path / "proj"
project.mkdir()
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, [
"init", "--here",
"--integration", integration,
"--script", "sh",
"--no-git",
"--ignore-agent-tools",
], catch_exceptions=False)
finally:
os.chdir(old)
assert result.exit_code == 0, result.output
return project
def test_upgrade_requires_speckit_project(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
old = os.getcwd()
try:
os.chdir(tmp_path)
result = runner.invoke(app, ["integration", "upgrade"])
finally:
os.chdir(old)
assert result.exit_code != 0
assert "Not a spec-kit project" in result.output
def test_upgrade_no_integration_installed(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = tmp_path / "proj"
project.mkdir()
(project / ".specify").mkdir()
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "upgrade"])
finally:
os.chdir(old)
assert result.exit_code == 0
assert "No integration is currently installed" in result.output
def test_upgrade_succeeds(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path, "copilot")
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "upgrade"], catch_exceptions=False)
finally:
os.chdir(old)
assert result.exit_code == 0
assert "upgraded successfully" in result.output
def test_upgrade_blocks_on_modified_files(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path, "copilot")
# Modify a tracked file so the manifest hash won't match
manifest_path = project / ".specify" / "integrations" / "copilot.manifest.json"
assert manifest_path.exists(), "Manifest should exist after init"
manifest_data = json.loads(manifest_path.read_text())
tracked_files = manifest_data.get("files", {})
assert tracked_files, "Manifest should track at least one file"
first_rel = next(iter(tracked_files))
target_file = project / first_rel
assert target_file.exists(), f"Tracked file {first_rel} should exist"
target_file.write_text("MODIFIED CONTENT\n")
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "upgrade"])
finally:
os.chdir(old)
assert result.exit_code != 0
assert "modified" in result.output.lower()
def test_upgrade_force_overwrites_modified(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path, "copilot")
# Modify a tracked file
manifest_path = project / ".specify" / "integrations" / "copilot.manifest.json"
manifest_data = json.loads(manifest_path.read_text())
tracked_files = manifest_data.get("files", {})
assert tracked_files, "Manifest should track at least one file"
first_rel = next(iter(tracked_files))
target_file = project / first_rel
assert target_file.exists(), f"Tracked file {first_rel} should exist"
target_file.write_text("MODIFIED CONTENT\n")
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "upgrade", "--force"], catch_exceptions=False)
finally:
os.chdir(old)
assert result.exit_code == 0
assert "upgraded successfully" in result.output
def test_upgrade_wrong_integration_key(self, tmp_path):
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path, "copilot")
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "upgrade", "claude"])
finally:
os.chdir(old)
assert result.exit_code != 0
assert "not the currently installed integration" in result.output
def test_upgrade_no_manifest(self, tmp_path):
"""Upgrade with missing manifest suggests fresh install."""
from typer.testing import CliRunner
from specify_cli import app
runner = CliRunner()
project = self._init_project(tmp_path, "copilot")
# Remove manifest
manifest_path = project / ".specify" / "integrations" / "copilot.manifest.json"
if manifest_path.exists():
manifest_path.unlink()
old = os.getcwd()
try:
os.chdir(project)
result = runner.invoke(app, ["integration", "upgrade"])
finally:
os.chdir(old)
assert result.exit_code == 0
assert "Nothing to upgrade" in result.output