Fix(fabric): Fix failing janitor test

Author: erindru

sqlmesh/core/engine_adapter/fabric.py

@@ -172,7 +172,7 @@ def __init__(self, tenant_id: str, workspace_id: str, client_id: str, client_sec
         self.client_secret = client_secret
         self.workspace_id = workspace_id
 
-    def create_warehouse(self, warehouse_name: str) -> None:
+    def create_warehouse(self, warehouse_name: str, if_not_exists: bool = True) -> None:
         """Create a catalog (warehouse) in Microsoft Fabric via REST API."""
         logger.info(f"Creating Fabric warehouse: {warehouse_name}")
 
@@ -182,6 +182,15 @@ def create_warehouse(self, warehouse_name: str) -> None:
         }
 
         response = self.session.post(self._endpoint_url("warehouses"), json=request_data)
+
+        if (
+            if_not_exists
+            and response.status_code == 400
+            and response.json().get("errorCode", None) == "ItemDisplayNameAlreadyInUse"
+        ):
+            logger.warning(f"Fabric warehouse {warehouse_name} already exists")
+            return
+
         response.raise_for_status()
 
         # Handle direct success (201) or async creation (202)
@@ -197,11 +206,12 @@ def create_warehouse(self, warehouse_name: str) -> None:
             logger.error(f"Unexpected response from Fabric API: {response}\n{response.text}")
             raise SQLMeshError(f"Unable to create warehouse: {response}")
 
-    def delete_warehouse(self, warehouse_name: str) -> None:
+    def delete_warehouse(self, warehouse_name: str, if_exists: bool = True) -> None:
         """Drop a catalog (warehouse) in Microsoft Fabric via REST API."""
         logger.info(f"Deleting Fabric warehouse: {warehouse_name}")
 
         # Get the warehouse ID by listing warehouses
+        # TODO: handle continuationUri for pagination, ref: https://learn.microsoft.com/en-us/rest/api/fabric/warehouse/items/list-warehouses?tabs=HTTP#warehouses
         response = self.session.get(self._endpoint_url("warehouses"))
         response.raise_for_status()
 
@@ -213,9 +223,12 @@ def delete_warehouse(self, warehouse_name: str) -> None:
         warehouse_id = warehouse_name_to_id.get(warehouse_name, None)
 
         if not warehouse_id:
-            logger.error(
+            logger.warning(
                 f"Fabric warehouse does not exist: {warehouse_name}\n(available warehouses: {', '.join(warehouse_name_to_id)})"
             )
+            if if_exists:
+                return
+
             raise SQLMeshError(
                 f"Unable to delete Fabric warehouse {warehouse_name} as it doesnt exist"
             )

tests/core/engine_adapter/integration/test_integration_fabric.py

@@ -0,0 +1,41 @@
+import typing as t
+import pytest
+from pytest import FixtureRequest
+from sqlmesh.core.engine_adapter import FabricEngineAdapter
+from tests.core.engine_adapter.integration import TestContext
+
+from tests.core.engine_adapter.integration import (
+    TestContext,
+    generate_pytest_params,
+    ENGINES_BY_NAME,
+    IntegrationTestEngine,
+)
+
+
+@pytest.fixture(
+    params=list(generate_pytest_params(ENGINES_BY_NAME["fabric"], show_variant_in_test_id=False))
+)
+def ctx(
+    request: FixtureRequest,
+    create_test_context: t.Callable[[IntegrationTestEngine, str, str], t.Iterable[TestContext]],
+) -> t.Iterable[TestContext]:
+    yield from create_test_context(*request.param)
+
+
+@pytest.fixture
+def engine_adapter(ctx: TestContext) -> FabricEngineAdapter:
+    assert isinstance(ctx.engine_adapter, FabricEngineAdapter)
+    return ctx.engine_adapter
+
+
+def test_create_drop_catalog(ctx: TestContext, engine_adapter: FabricEngineAdapter):
+    catalog_name = ctx.add_test_suffix("test_catalog")
+
+    try:
+        ctx.create_catalog(catalog_name)
+        # if already exists, should be no-op, not error
+        ctx.create_catalog(catalog_name)
+        ctx.drop_catalog(catalog_name)
+    finally:
+        # if doesnt exist, should be no-op, not error
+        ctx.drop_catalog(catalog_name)