add tests

Author: jonhealy1

stac_fastapi/tests/api/test_api.py

@@ -50,6 +50,10 @@
     "POST /collections/{collection_id}/aggregate",
     "GET /collections-search",
     "POST /collections-search",
+    "GET /catalogs",
+    "POST /catalogs",
+    "GET /catalogs/{catalog_id}",
+    "GET /catalogs/{catalog_id}/collections",
 }
 
 

stac_fastapi/tests/conftest.py

@@ -39,6 +39,7 @@
 from stac_fastapi.types.config import Settings
 
 os.environ.setdefault("ENABLE_COLLECTIONS_SEARCH_ROUTE", "true")
+os.environ.setdefault("ENABLE_CATALOG_ROUTE", "false")
 
 if os.getenv("BACKEND", "elasticsearch").lower() == "opensearch":
     from stac_fastapi.opensearch.app import app_config
@@ -396,3 +397,63 @@ def build_test_app():
     # Create and return the app
     api = StacApi(**test_config)
     return api.app
+
+
+def build_test_app_with_catalogs():
+    """Build a test app with catalogs extension enabled."""
+    from stac_fastapi.core.extensions.catalogs import CatalogsExtension
+
+    # Get the base config
+    test_config = app_config.copy()
+
+    # Get database and settings (already imported above)
+    test_database = DatabaseLogic()
+    test_settings = AsyncSettings()
+
+    # Add catalogs extension
+    catalogs_extension = CatalogsExtension(
+        client=CoreClient(
+            database=test_database,
+            session=None,
+            landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
+        ),
+        settings=test_settings,
+        conformance_classes=[
+            "https://api.stacspec.org/v1.0.0-beta.1/catalogs",
+        ],
+    )
+
+    # Add to extensions if not already present
+    if not any(isinstance(ext, CatalogsExtension) for ext in test_config["extensions"]):
+        test_config["extensions"].append(catalogs_extension)
+
+    # Update client with new extensions
+    test_config["client"] = CoreClient(
+        database=test_database,
+        session=None,
+        extensions=test_config["extensions"],
+        post_request_model=test_config["search_post_request_model"],
+        landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
+    )
+
+    # Create and return the app
+    api = StacApi(**test_config)
+    return api.app
+
+
+@pytest_asyncio.fixture(scope="session")
+async def catalogs_app():
+    """Fixture to get the FastAPI app with catalogs extension enabled."""
+    return build_test_app_with_catalogs()
+
+
+@pytest_asyncio.fixture(scope="session")
+async def catalogs_app_client(catalogs_app):
+    """Fixture to get an async client for the app with catalogs extension enabled."""
+    await create_index_templates()
+    await create_collection_index()
+
+    async with AsyncClient(
+        transport=ASGITransport(app=catalogs_app), base_url="http://test-server"
+    ) as c:
+        yield c

stac_fastapi/tests/data/test_catalog.json

@@ -0,0 +1,37 @@
+{
+  "id": "test-catalog",
+  "type": "Catalog",
+  "stac_version": "1.0.0",
+  "title": "Test Catalog",
+  "description": "A test catalog for STAC API testing",
+  "links": [
+    {
+      "rel": "self",
+      "href": "http://test-server/catalogs/test-catalog",
+      "type": "application/json"
+    },
+    {
+      "rel": "root",
+      "href": "http://test-server/catalogs",
+      "type": "application/json"
+    },
+    {
+      "rel": "child",
+      "href": "http://test-server/collections/test-collection-1",
+      "type": "application/json",
+      "title": "Test Collection 1"
+    },
+    {
+      "rel": "child",
+      "href": "http://test-server/collections/test-collection-2",
+      "type": "application/json",
+      "title": "Test Collection 2"
+    },
+    {
+      "rel": "child",
+      "href": "http://test-server/collections/test-collection-3",
+      "type": "application/json",
+      "title": "Test Collection 3"
+    }
+  ]
+}

stac_fastapi/tests/extensions/test_catalogs.py

@@ -0,0 +1,127 @@
+import uuid
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_get_root_catalog(catalogs_app_client, load_test_data):
+    """Test getting the root catalog."""
+    resp = await catalogs_app_client.get("/catalogs")
+    assert resp.status_code == 200
+
+    catalog = resp.json()
+    assert catalog["type"] == "Catalog"
+    assert catalog["id"] == "root"
+    assert catalog["stac_version"] == "1.0.0"
+    assert "links" in catalog
+
+    # Check for required links
+    links = catalog["links"]
+    link_rels = [link["rel"] for link in links]
+    assert "self" in link_rels
+    assert "root" in link_rels
+    assert "parent" in link_rels
+
+
+@pytest.mark.asyncio
+async def test_create_catalog(catalogs_app_client, load_test_data):
+    """Test creating a new catalog."""
+    test_catalog = load_test_data("test_catalog.json")
+    test_catalog["id"] = f"test-catalog-{uuid.uuid4()}"
+
+    resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+    assert resp.status_code == 201
+
+    created_catalog = resp.json()
+    assert created_catalog["id"] == test_catalog["id"]
+    assert created_catalog["type"] == "Catalog"
+    assert created_catalog["title"] == test_catalog["title"]
+
+
+@pytest.mark.asyncio
+async def test_get_catalog(catalogs_app_client, load_test_data):
+    """Test getting a specific catalog."""
+    # First create a catalog
+    test_catalog = load_test_data("test_catalog.json")
+    test_catalog["id"] = f"test-catalog-{uuid.uuid4()}"
+
+    create_resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+    assert create_resp.status_code == 201
+
+    # Now get it back
+    resp = await catalogs_app_client.get(f"/catalogs/{test_catalog['id']}")
+    assert resp.status_code == 200
+
+    catalog = resp.json()
+    assert catalog["id"] == test_catalog["id"]
+    assert catalog["title"] == test_catalog["title"]
+    assert catalog["description"] == test_catalog["description"]
+
+
+@pytest.mark.asyncio
+async def test_get_nonexistent_catalog(catalogs_app_client):
+    """Test getting a catalog that doesn't exist."""
+    resp = await catalogs_app_client.get("/catalogs/nonexistent-catalog")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_get_catalog_collections(catalogs_app_client, load_test_data, ctx):
+    """Test getting collections linked from a catalog."""
+    # First create a catalog with a link to the test collection
+    test_catalog = load_test_data("test_catalog.json")
+    test_catalog["id"] = f"test-catalog-{uuid.uuid4()}"
+
+    # Update the catalog links to point to the actual test collection
+    for link in test_catalog["links"]:
+        if link["rel"] == "child":
+            link["href"] = f"http://test-server/collections/{ctx.collection['id']}"
+
+    create_resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+    assert create_resp.status_code == 201
+
+    # Now get collections from the catalog
+    resp = await catalogs_app_client.get(f"/catalogs/{test_catalog['id']}/collections")
+    assert resp.status_code == 200
+
+    collections_response = resp.json()
+    assert "collections" in collections_response
+    assert "links" in collections_response
+
+    # Should contain the test collection
+    collection_ids = [col["id"] for col in collections_response["collections"]]
+    assert ctx.collection["id"] in collection_ids
+
+
+@pytest.mark.asyncio
+async def test_get_catalog_collections_nonexistent_catalog(catalogs_app_client):
+    """Test getting collections from a catalog that doesn't exist."""
+    resp = await catalogs_app_client.get("/catalogs/nonexistent-catalog/collections")
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_root_catalog_with_multiple_catalogs(catalogs_app_client, load_test_data):
+    """Test that root catalog includes links to multiple catalogs."""
+    # Create multiple catalogs
+    catalog_ids = []
+    for i in range(3):
+        test_catalog = load_test_data("test_catalog.json")
+        test_catalog["id"] = f"test-catalog-{uuid.uuid4()}-{i}"
+        test_catalog["title"] = f"Test Catalog {i}"
+
+        resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+        assert resp.status_code == 201
+        catalog_ids.append(test_catalog["id"])
+
+    # Get root catalog
+    resp = await catalogs_app_client.get("/catalogs")
+    assert resp.status_code == 200
+
+    catalog = resp.json()
+    child_links = [link for link in catalog["links"] if link["rel"] == "child"]
+
+    # Should have child links for all created catalogs
+    child_hrefs = [link["href"] for link in child_links]
+    for catalog_id in catalog_ids:
+        assert any(catalog_id in href for href in child_hrefs)