feat: add catalogs CLI commands

.github/workflows/test-pr.yml

@@ -21,6 +21,10 @@ on:
       - 'cortexapps_cli/**'
       - 'tests/**'
 
+concurrency:
+  group: test-pr-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
 env:
   AWS_ACCOUNT_ID: ${{ secrets.AWS_ACCOUNT_ID }}
   CORTEX_API_KEY: ${{ secrets.CORTEX_API_KEY }}

cortexapps_cli/cli.py

@@ -16,6 +16,7 @@
 import cortexapps_cli.commands.audit_logs as audit_logs
 import cortexapps_cli.commands.backup as backup
 import cortexapps_cli.commands.catalog as catalog
+import cortexapps_cli.commands.catalogs as catalogs
 import cortexapps_cli.commands.custom_data as custom_data
 import cortexapps_cli.commands.custom_events as custom_events
 import cortexapps_cli.commands.custom_metrics as custom_metrics
@@ -54,6 +55,7 @@
 app.add_typer(audit_logs.app, name="audit-logs")
 app.add_typer(backup.app, name="backup")
 app.add_typer(catalog.app, name="catalog")
+app.add_typer(catalogs.app, name="catalogs")
 app.add_typer(custom_data.app, name="custom-data")
 app.add_typer(custom_events.app, name="custom-events")
 app.add_typer(custom_metrics.app, name="custom-metrics")

cortexapps_cli/commands/catalogs.py

@@ -0,0 +1,117 @@
+import json
+from typing import Optional
+import typer
+from typing_extensions import Annotated
+from cortexapps_cli.command_options import CommandOptions
+from cortexapps_cli.command_options import ListCommandOptions
+from cortexapps_cli.utils import print_output_with_context
+
+app = typer.Typer(
+    help="Catalogs commands — manage catalog pages (distinct from 'catalog' which manages entities)",
+    no_args_is_help=True
+)
+
+@app.command(name="list")
+def catalogs_list(
+    ctx: typer.Context,
+    page: ListCommandOptions.page = None,
+    page_size: ListCommandOptions.page_size = 250,
+    _print: CommandOptions._print = True,
+    table_output: ListCommandOptions.table_output = False,
+    csv_output: ListCommandOptions.csv_output = False,
+    columns: ListCommandOptions.columns = [],
+    no_headers: ListCommandOptions.no_headers = False,
+    filters: ListCommandOptions.filters = [],
+    sort: ListCommandOptions.sort = [],
+):
+    """
+    List all catalogs. API key must have the View catalogs permission.
+    """
+    client = ctx.obj["client"]
+
+    if (table_output or csv_output) and not ctx.params.get('columns'):
+        ctx.params['columns'] = [
+            "Name=name",
+            "Slug=slug",
+            "Description=description",
+            "Icon=iconTag",
+            "IsDraft=isDraft",
+            "Type=catalogType",
+        ]
+
+    params = {k: v for k, v in {"page": page, "pageSize": page_size}.items() if v is not None}
+
+    result = client.fetch("api/v1/catalogs", params=params) if page is None else client.get("api/v1/catalogs", params=params)
+
+    if _print:
+        print_output_with_context(ctx, result)
+    else:
+        return result
+
+
+@app.command()
+def get(
+    ctx: typer.Context,
+    slug: str = typer.Option(..., "--slug", "-s", help="The slug of the catalog"),
+    _print: CommandOptions._print = True,
+):
+    """
+    Retrieve a catalog by its slug. API key must have the View catalogs permission.
+    """
+    client = ctx.obj["client"]
+
+    result = client.get("api/v1/catalogs/" + slug)
+
+    if _print:
+        print_output_with_context(ctx, result)
+    else:
+        return result
+
+
+@app.command()
+def create(
+    ctx: typer.Context,
+    file_input: Annotated[
+        typer.FileText,
+        typer.Option(..., "--file", "-f", help="File containing JSON catalog definition; use - for stdin, e.g. -f-"),
+    ],
+    mode: Optional[str] = typer.Option(
+        None,
+        "--mode",
+        "-m",
+        help="UPSERT (default): create or replace existing catalog. CREATE: fail if slug already exists.",
+    ),
+    _print: CommandOptions._print = True,
+):
+    """
+    Create or replace a catalog. API key must have the Edit catalogs permission.
+
+    JSON fields: name (required), slug (required), iconTag (required), description,
+    isDraft, filter, relationshipTypeTag, catalogType (FILTER|RELATIONSHIP_TYPE|DOMAIN).
+    """
+    client = ctx.obj["client"]
+    data = json.loads(file_input.read())
+
+    params = {}
+    if mode:
+        params["mode"] = mode.upper()
+
+    result = client.post("api/v1/catalogs", data=data, params=params if params else None)
+
+    if _print:
+        print_output_with_context(ctx, result)
+    else:
+        return result
+
+
+@app.command()
+def delete(
+    ctx: typer.Context,
+    slug: str = typer.Option(..., "--slug", "-s", help="The slug of the catalog to delete"),
+):
+    """
+    Delete a catalog by its slug. API key must have the Edit catalogs permission.
+    """
+    client = ctx.obj["client"]
+
+    client.delete("api/v1/catalogs/" + slug)

data/import/catalogs/cli-test-catalog.json

@@ -0,0 +1,7 @@
+{
+  "name": "CLI Test Catalog",
+  "slug": "cli-test-catalog",
+  "description": "Catalog created by CLI integration tests",
+  "iconTag": "service",
+  "isDraft": false
+}

tests/test_catalogs.py

@@ -0,0 +1,53 @@
+from tests.helpers.utils import *
+
+
+def test_catalogs_crud():
+    """Test full lifecycle: create, list, get, delete for catalogs."""
+
+    # Create a catalog from a JSON file
+    cli(["catalogs", "create", "-f", "data/import/catalogs/cli-test-catalog.json"])
+
+    # List catalogs and verify the new one exists
+    response = cli(["catalogs", "list"])
+    assert any(catalog['slug'] == 'cli-test-catalog' for catalog in response['catalogs']), \
+        "Should find catalog with slug cli-test-catalog"
+
+    # Get the specific catalog by slug
+    response = cli(["catalogs", "get", "-s", "cli-test-catalog"])
+    assert response['name'] == "CLI Test Catalog", "Catalog name should match"
+    assert response['slug'] == "cli-test-catalog", "Catalog slug should match"
+
+    # Delete the catalog
+    cli(["catalogs", "delete", "-s", "cli-test-catalog"])
+
+    # Verify deletion - get should return 404
+    result = cli(["catalogs", "get", "-s", "cli-test-catalog"], ReturnType.RAW)
+    assert result.exit_code == 1, "Getting a deleted catalog should fail"
+
+
+def test_catalogs_upsert_replaces_existing():
+    """UPSERT mode (default) should replace an existing catalog without error."""
+
+    cli(["catalogs", "create", "-f", "data/import/catalogs/cli-test-catalog.json"])
+
+    try:
+        # Second create with same slug should succeed (upsert replaces)
+        response = cli(["catalogs", "create", "-f", "data/import/catalogs/cli-test-catalog.json"])
+        assert response['slug'] == 'cli-test-catalog', "Upserted catalog should be returned"
+    finally:
+        cli(["catalogs", "delete", "-s", "cli-test-catalog"])
+
+
+def test_catalogs_create_mode_fails_on_duplicate():
+    """mode=CREATE should fail with an error if the slug already exists."""
+
+    cli(["catalogs", "create", "-f", "data/import/catalogs/cli-test-catalog.json"])
+
+    try:
+        result = cli(
+            ["catalogs", "create", "-f", "data/import/catalogs/cli-test-catalog.json", "--mode", "CREATE"],
+            ReturnType.RAW,
+        )
+        assert result.exit_code == 1, "CREATE mode should fail if slug already exists"
+    finally:
+        cli(["catalogs", "delete", "-s", "cli-test-catalog"])