Move replace integration tests to tests/integration/test_catalog.py

That file is the cross-catalog integration analog to Java's
CatalogTests.java; every other mutation (create/rename/drop/etc.)
already has a slot there parametrized over six catalog fixtures.
The two replace tests have no REST-specific code and don't belong
in test_rest_catalog.py.

Adopts the file's existing conventions: test_catalog parameter
name, database_name + table_name fixtures, no manual cleanup
guards (clean_up runs between tests).

Author: smaheshwar-pltr

tests/integration/test_catalog.py

@@ -20,6 +20,7 @@
 from collections.abc import Generator
 from pathlib import Path, PosixPath
 
+import pyarrow as pa
 import pytest
 from pytest_lazy_fixtures import lf
 
@@ -43,7 +44,7 @@
 from pyiceberg.table.metadata import INITIAL_SPEC_ID
 from pyiceberg.table.sorting import INITIAL_SORT_ORDER_ID, SortField, SortOrder
 from pyiceberg.transforms import BucketTransform, DayTransform, IdentityTransform
-from pyiceberg.types import IntegerType, LongType, NestedField, TimestampType, UUIDType
+from pyiceberg.types import BooleanType, IntegerType, LongType, NestedField, StringType, TimestampType, UUIDType
 from tests.conftest import (
     clean_up,
     does_support_atomic_concurrent_updates,
@@ -866,3 +867,63 @@ def test_load_missing_table(test_catalog: Catalog, database_name: str, table_nam
 
     with pytest.raises(NoSuchTableError):
         test_catalog.load_table(identifier)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_replace_table(test_catalog: Catalog, database_name: str, table_name: str) -> None:
+    test_catalog.create_namespace(database_name)
+    identifier = (database_name, table_name)
+
+    original_schema = Schema(
+        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
+        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
+    )
+    original = test_catalog.create_table(identifier, schema=original_schema)
+    original.append(
+        pa.Table.from_pydict(
+            {"id": [1, 2, 3], "data": ["a", "b", "c"]},
+            schema=pa.schema([pa.field("id", pa.int64()), pa.field("data", pa.large_string())]),
+        )
+    )
+    original.refresh()
+    original_snapshot_id = original.current_snapshot().snapshot_id  # type: ignore[union-attr]
+
+    new_schema = Schema(
+        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
+        NestedField(field_id=2, name="name", field_type=StringType(), required=False),
+        NestedField(field_id=3, name="active", field_type=BooleanType(), required=False),
+    )
+    replaced = test_catalog.replace_table(identifier, schema=new_schema)
+
+    assert replaced.metadata.table_uuid == original.metadata.table_uuid
+    assert replaced.current_snapshot() is None
+    assert any(s.snapshot_id == original_snapshot_id for s in replaced.metadata.snapshots)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_replace_table_transaction(test_catalog: Catalog, database_name: str, table_name: str) -> None:
+    test_catalog.create_namespace(database_name)
+    identifier = (database_name, table_name)
+
+    old_data = pa.Table.from_pydict(
+        {"id": [1], "data": ["old"]},
+        schema=pa.schema([pa.field("id", pa.int64()), pa.field("data", pa.large_string())]),
+    )
+    original = test_catalog.create_table(identifier, schema=old_data.schema)
+    original.append(old_data)
+    old_snapshot_id = test_catalog.load_table(identifier).current_snapshot().snapshot_id  # type: ignore[union-attr]
+
+    new_data = pa.Table.from_pydict(
+        {"id": [10, 20], "name": ["alice", "bob"]},
+        schema=pa.schema([pa.field("id", pa.int64()), pa.field("name", pa.large_string())]),
+    )
+    with test_catalog.replace_table_transaction(identifier, schema=new_data.schema) as txn:
+        txn.append(new_data)
+
+    replaced = test_catalog.load_table(identifier)
+    assert replaced.current_snapshot() is not None
+    assert replaced.current_snapshot().snapshot_id != old_snapshot_id  # type: ignore[union-attr]
+    assert any(s.snapshot_id == old_snapshot_id for s in replaced.metadata.snapshots)
+    assert replaced.scan().to_arrow().num_rows == 2

tests/integration/test_rest_catalog.py

@@ -18,15 +18,12 @@
 
 import time
 
-import pyarrow as pa
 import pytest
 from pytest_lazy_fixtures import lf
 
-from pyiceberg.catalog import Catalog
 from pyiceberg.catalog.rest import RestCatalog
 from pyiceberg.exceptions import NoSuchViewError
 from pyiceberg.schema import Schema
-from pyiceberg.types import BooleanType, LongType, NestedField, StringType
 from pyiceberg.view.metadata import SQLViewRepresentation, ViewVersion
 
 TEST_NAMESPACE_IDENTIFIER = "TEST NS"
@@ -72,74 +69,6 @@ def test_create_namespace_if_already_existing(catalog: RestCatalog) -> None:
     assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
 
 
-@pytest.mark.integration
-@pytest.mark.parametrize("catalog", [lf("session_catalog")])
-def test_replace_table(catalog: Catalog) -> None:
-    identifier = f"default.test_replace_table_e2e_{catalog.name}"
-    if not catalog.namespace_exists("default"):
-        catalog.create_namespace("default")
-    if catalog.table_exists(identifier):
-        catalog.drop_table(identifier)
-
-    original_schema = Schema(
-        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
-        NestedField(field_id=2, name="data", field_type=StringType(), required=False),
-    )
-    original = catalog.create_table(identifier, schema=original_schema)
-    original.append(
-        pa.Table.from_pydict(
-            {"id": [1, 2, 3], "data": ["a", "b", "c"]},
-            schema=pa.schema([pa.field("id", pa.int64()), pa.field("data", pa.large_string())]),
-        )
-    )
-    original.refresh()
-    original_snapshot_id = original.current_snapshot().snapshot_id  # type: ignore[union-attr]
-
-    new_schema = Schema(
-        NestedField(field_id=1, name="id", field_type=LongType(), required=False),
-        NestedField(field_id=2, name="name", field_type=StringType(), required=False),
-        NestedField(field_id=3, name="active", field_type=BooleanType(), required=False),
-    )
-    replaced = catalog.replace_table(identifier, schema=new_schema)
-
-    assert replaced.metadata.table_uuid == original.metadata.table_uuid
-    assert replaced.current_snapshot() is None
-    assert any(s.snapshot_id == original_snapshot_id for s in replaced.metadata.snapshots)
-    catalog.drop_table(identifier)
-
-
-@pytest.mark.integration
-@pytest.mark.parametrize("catalog", [lf("session_catalog")])
-def test_replace_table_transaction(catalog: Catalog) -> None:
-    identifier = f"default.test_replace_rtas_{catalog.name}"
-    if not catalog.namespace_exists("default"):
-        catalog.create_namespace("default")
-    if catalog.table_exists(identifier):
-        catalog.drop_table(identifier)
-
-    old_data = pa.Table.from_pydict(
-        {"id": [1], "data": ["old"]},
-        schema=pa.schema([pa.field("id", pa.int64()), pa.field("data", pa.large_string())]),
-    )
-    original = catalog.create_table(identifier, schema=old_data.schema)
-    original.append(old_data)
-    old_snapshot_id = catalog.load_table(identifier).current_snapshot().snapshot_id  # type: ignore[union-attr]
-
-    new_data = pa.Table.from_pydict(
-        {"id": [10, 20], "name": ["alice", "bob"]},
-        schema=pa.schema([pa.field("id", pa.int64()), pa.field("name", pa.large_string())]),
-    )
-    with catalog.replace_table_transaction(identifier, schema=new_data.schema) as txn:
-        txn.append(new_data)
-
-    replaced = catalog.load_table(identifier)
-    assert replaced.current_snapshot() is not None
-    assert replaced.current_snapshot().snapshot_id != old_snapshot_id  # type: ignore[union-attr]
-    assert any(s.snapshot_id == old_snapshot_id for s in replaced.metadata.snapshots)
-    assert replaced.scan().to_arrow().num_rows == 2
-    catalog.drop_table(identifier)
-
-
 @pytest.mark.integration
 @pytest.mark.parametrize("catalog", [lf("session_catalog")])
 def test_load_view(catalog: RestCatalog, table_schema_nested: Schema, database_name: str, view_name: str) -> None: