Fix test isolation issues in integration tests (#144)

- Enhanced state store cleanup in test_data_modeling_replicator fixture
  to prevent test interference by clearing cursors between tests
- Improved delete_delta_table_data with retry logic and verification
- Added aggressive pre-test cleanup in instance_table_paths fixture
- Fixed test_data_model_sync_service_delete failing when run in sequence
- Fixed test_events_service[10-100] isolation issues
- Tests now pass consistently both individually and in sequence
- Applied code formatting via pre-commit hooks

Author: muradsater

tests/integration/integration_steps/fabric_steps.py

@@ -8,6 +8,7 @@
 from deltalake.writer import write_deltalake
 from deltalake.exceptions import TableNotFoundError
 from azure.storage.filedatalake import DataLakeServiceClient
+import time
 
 TIMESTAMP_COLUMN = "timestamp"
 DATA_MODEL_TIMESTAMP_COLUMNS = ["lastUpdatedTime", "createdTime"]
@@ -30,11 +31,40 @@ def get_ts_delta_table(
 
 
 def delete_delta_table_data(credential: DefaultAzureCredential, path: str):
-    try:
-        delta_table = get_ts_delta_table(credential, path)
-        delta_table.delete()
-    except TableNotFoundError:
-        print(f"Table not found {path}")
+    max_retries = 3
+    retry_delay = 1  # seconds
+
+    for attempt in range(max_retries):
+        try:
+            delta_table = get_ts_delta_table(credential, path)
+            delta_table.delete()
+
+            # Verify deletion by checking if table is empty
+            time.sleep(0.5)  # Small delay to ensure operation completes
+            try:
+                verification_table = get_ts_delta_table(credential, path)
+                df = verification_table.to_pandas()
+                if len(df) == 0:
+                    return  # Successfully deleted all data
+                else:
+                    print(
+                        f"Warning: Table {path} still contains {len(df)} rows after deletion attempt {attempt + 1}"
+                    )
+            except TableNotFoundError:
+                return  # Table doesn't exist, which is fine
+
+        except TableNotFoundError:
+            print(f"Table not found {path}")
+            return
+        except Exception as e:
+            print(
+                f"Attempt {attempt + 1}/{max_retries}: Error deleting data from {path}: {e}"
+            )
+            if attempt < max_retries - 1:
+                time.sleep(retry_delay)
+            else:
+                print(f"Failed to delete data from {path} after {max_retries} attempts")
+                raise
 
 
 def read_deltalake_timeseries(timeseries_path: str, credential: DefaultAzureCredential):

tests/integration/test_data_model_replicator.py

@@ -19,6 +19,7 @@
     lakehouse_table_name,
     assert_data_model_instances_in_fabric,
     assert_data_model_instances_update,
+    get_ts_delta_table,
 )
 from integration_steps.data_model_generation import Node, Edge, create_node, create_edge
 
@@ -38,6 +39,31 @@ def test_data_modeling_replicator():
     replicator._load_state_store()
     replicator.logger = Mock()
     yield replicator
+
+    # Clean up state store after each test to prevent test interference
+    try:
+        # Clear all state store entries that might interfere with subsequent tests
+        for data_model_config in replicator.config.data_modeling or []:
+            try:
+                all_views = replicator.cognite_client.data_modeling.views.list(
+                    space=data_model_config.space, limit=-1
+                )
+                views_dict = all_views.dump()
+                for view in views_dict:
+                    state_id = f"state_{data_model_config.space}_{view['externalId']}_{view['version']}"
+                    replicator.state_store.delete_state(state_id)
+                # Also clean edge state
+                edge_state_id = f"state_{data_model_config.space}_edges"
+                replicator.state_store.delete_state(edge_state_id)
+            except Exception as e:
+                print(
+                    f"Warning: Could not clean up state for space {data_model_config.space}: {e}"
+                )
+
+        replicator.state_store.synchronize()
+    except Exception as e:
+        print(f"Warning: Could not clean up state store: {e}")
+
     try:
         os.remove("states.json")
     except FileNotFoundError:
@@ -116,10 +142,34 @@ def instance_table_paths(
         instance_table_paths.append(
             lakehouse_table_name(test_model.space + "_" + view.external_id)
         )
+        # Clean up before the test
         delete_delta_table_data(azure_credential, instance_table_paths[-1])
+
     edge_table_path = lakehouse_table_name(test_model.space + "_edges")
     instance_table_paths.append(edge_table_path)
+    # Clean up edge table before the test
+    delete_delta_table_data(azure_credential, edge_table_path)
+
+    # Double-check cleanup - make sure tables are completely empty before proceeding
+    import time
+
+    time.sleep(1)  # Give some time for cleanup operations to complete
+
+    for path in instance_table_paths:
+        try:
+            delta_table = get_ts_delta_table(azure_credential, path)
+            df = delta_table.to_pandas()
+            if len(df) > 0:
+                print(
+                    f"Warning: Table {path} still contains {len(df)} rows before test, attempting additional cleanup"
+                )
+                delete_delta_table_data(azure_credential, path)
+        except Exception as e:
+            print(f"Could not verify cleanup for {path}: {e}")
+
     yield instance_table_paths
+
+    # Clean up after the test
     for path in instance_table_paths:
         delete_delta_table_data(azure_credential, path)