ensure parent ids are removed

jonhealy1 · jonhealy1 · commit cfdcfc1dd346 · 2025-12-09T14:45:39.000+08:00
diff --git a/stac_fastapi/core/stac_fastapi/core/extensions/catalogs.py b/stac_fastapi/core/stac_fastapi/core/extensions/catalogs.py
@@ -360,52 +360,44 @@ async def delete_catalog(
                                 f"Failed to delete collection {coll_id}: {e}"
                             )
             else:
-                # Remove catalog link from each collection (orphan them)
+                # Remove catalog from each collection's parent_ids and links (orphan them)
                 for coll_id in collection_ids:
                     try:
-                        collection = await self.client.get_collection(
-                            coll_id, request=request
+                        # Get the collection from database to access parent_ids
+                        collection_db = await self.client.database.find_collection(
+                            coll_id
                         )
-                        # Remove the catalog link from the collection
-                        if hasattr(collection, "links"):
-                            collection.links = [
-                                link
-                                for link in collection.links
-                                if not (
-                                    getattr(link, "rel", None) == "catalog"
-                                    and catalog_id in getattr(link, "href", "")
-                                )
-                            ]
-                        elif isinstance(collection, dict):
-                            collection["links"] = [
-                                link
-                                for link in collection.get("links", [])
-                                if not (
-                                    link.get("rel") == "catalog"
-                                    and catalog_id in link.get("href", "")
-                                )
-                            ]
 
-                        # Update the collection in the database
-                        collection_dict = (
-                            collection.model_dump(mode="json")
-                            if hasattr(collection, "model_dump")
-                            else collection
-                        )
-                        collection_db = (
-                            self.client.database.collection_serializer.stac_to_db(
-                                collection_dict, request
+                        # Remove catalog_id from parent_ids
+                        parent_ids = collection_db.get("parent_ids", [])
+                        if catalog_id in parent_ids:
+                            parent_ids.remove(catalog_id)
+                            collection_db["parent_ids"] = parent_ids
+
+                            # Also remove the catalog link from the collection's links
+                            if "links" in collection_db:
+                                collection_db["links"] = [
+                                    link
+                                    for link in collection_db.get("links", [])
+                                    if not (
+                                        link.get("rel") == "catalog"
+                                        and catalog_id in link.get("href", "")
+                                    )
+                                ]
+
+                            # Update the collection in the database
+                            await self.client.database.update_collection(
+                                collection_id=coll_id,
+                                collection=collection_db,
+                                refresh=True,
+                            )
+                            logger.info(
+                                f"Removed catalog {catalog_id} from collection {coll_id} parent_ids and links"
+                            )
+                        else:
+                            logger.debug(
+                                f"Catalog {catalog_id} not in parent_ids for collection {coll_id}"
                             )
-                        )
-                        await self.client.database.client.index(
-                            index=COLLECTIONS_INDEX,
-                            id=coll_id,
-                            body=collection_db.model_dump()
-                            if hasattr(collection_db, "model_dump")
-                            else collection_db,
-                            refresh=True,
-                        )
-                        logger.info(f"Removed catalog link from collection {coll_id}")
                     except Exception as e:
                         error_msg = str(e)
                         if "not found" in error_msg.lower():
@@ -414,7 +406,7 @@ async def delete_catalog(
                             )
                         else:
                             logger.warning(
-                                f"Failed to remove catalog link from collection {coll_id}: {e}"
+                                f"Failed to remove catalog {catalog_id} from collection {coll_id}: {e}"
                             )
 
             # Delete the catalog
diff --git a/stac_fastapi/tests/extensions/test_catalogs.py b/stac_fastapi/tests/extensions/test_catalogs.py
@@ -605,6 +605,64 @@ async def test_delete_catalog_no_cascade(catalogs_app_client, load_test_data):
     ), "Collection should not have catalog link after catalog deletion"
 
 
+@pytest.mark.asyncio
+async def test_delete_catalog_removes_parent_ids_from_collections(
+    catalogs_app_client, load_test_data
+):
+    """Test that deleting a catalog removes its ID from child collections' parent_ids."""
+    # Create a catalog
+    test_catalog = load_test_data("test_catalog.json")
+    catalog_id = f"test-catalog-{uuid.uuid4()}"
+    test_catalog["id"] = catalog_id
+    test_catalog["links"] = [
+        link for link in test_catalog.get("links", []) if link.get("rel") != "child"
+    ]
+
+    catalog_resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+    assert catalog_resp.status_code == 201
+
+    # Create 3 collections in the catalog
+    collection_ids = []
+    for i in range(3):
+        test_collection = load_test_data("test_collection.json")
+        collection_id = f"test-collection-{uuid.uuid4()}-{i}"
+        test_collection["id"] = collection_id
+
+        create_resp = await catalogs_app_client.post(
+            f"/catalogs/{catalog_id}/collections", json=test_collection
+        )
+        assert create_resp.status_code == 201
+        collection_ids.append(collection_id)
+
+    # Verify all collections have the catalog in their parent_ids
+    # (indirectly verified by checking they're accessible via the catalog endpoint)
+    get_collections_resp = await catalogs_app_client.get(
+        f"/catalogs/{catalog_id}/collections"
+    )
+    assert get_collections_resp.status_code == 200
+    collections_response = get_collections_resp.json()
+    returned_ids = [col["id"] for col in collections_response["collections"]]
+    for collection_id in collection_ids:
+        assert collection_id in returned_ids
+
+    # Delete the catalog without cascade
+    delete_resp = await catalogs_app_client.delete(f"/catalogs/{catalog_id}")
+    assert delete_resp.status_code == 204
+
+    # Verify all collections still exist
+    for collection_id in collection_ids:
+        get_resp = await catalogs_app_client.get(f"/collections/{collection_id}")
+        assert get_resp.status_code == 200
+
+    # Verify collections are no longer accessible via the deleted catalog
+    # (This indirectly verifies parent_ids was updated)
+    for collection_id in collection_ids:
+        get_from_catalog_resp = await catalogs_app_client.get(
+            f"/catalogs/{catalog_id}/collections/{collection_id}"
+        )
+        assert get_from_catalog_resp.status_code == 404
+
+
 @pytest.mark.asyncio
 async def test_create_catalog_collection_adds_parent_id(
     catalogs_app_client, load_test_data
@@ -880,6 +938,60 @@ async def test_delete_collection_not_in_catalog_returns_404(
     assert delete_resp.status_code == 404
 
 
+@pytest.mark.asyncio
+async def test_catalog_links_contain_all_collections(
+    catalogs_app_client, load_test_data
+):
+    """Test that a catalog's links contain all 3 collections added to it."""
+    # Create a catalog
+    test_catalog = load_test_data("test_catalog.json")
+    catalog_id = f"test-catalog-{uuid.uuid4()}"
+    test_catalog["id"] = catalog_id
+    # Remove any placeholder child links
+    test_catalog["links"] = [
+        link for link in test_catalog.get("links", []) if link.get("rel") != "child"
+    ]
+
+    catalog_resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+    assert catalog_resp.status_code == 201
+
+    # Create 3 collections in the catalog
+    collection_ids = []
+    for i in range(3):
+        test_collection = load_test_data("test_collection.json")
+        collection_id = f"test-collection-{uuid.uuid4()}-{i}"
+        test_collection["id"] = collection_id
+
+        create_resp = await catalogs_app_client.post(
+            f"/catalogs/{catalog_id}/collections", json=test_collection
+        )
+        assert create_resp.status_code == 201
+        collection_ids.append(collection_id)
+
+    # Get the catalog and verify all 3 collections are in its links
+    catalog_get_resp = await catalogs_app_client.get(f"/catalogs/{catalog_id}")
+    assert catalog_get_resp.status_code == 200
+
+    catalog_data = catalog_get_resp.json()
+    catalog_links = catalog_data.get("links", [])
+
+    # Extract all child links (collection links)
+    child_links = [link for link in catalog_links if link.get("rel") == "child"]
+
+    # Verify we have exactly 3 child links
+    assert (
+        len(child_links) == 3
+    ), f"Catalog should have 3 child links, but has {len(child_links)}"
+
+    # Verify each collection ID is in the child links
+    child_hrefs = [link.get("href", "") for link in child_links]
+    for collection_id in collection_ids:
+        collection_href = f"/collections/{collection_id}"
+        assert any(
+            collection_href in href for href in child_hrefs
+        ), f"Collection {collection_id} missing from catalog links. Found links: {child_hrefs}"
+
+
 @pytest.mark.asyncio
 async def test_parent_ids_not_exposed_to_client(catalogs_app_client, load_test_data):
     """Test that parent_ids field is not exposed in API responses."""
