Add missing catalog tests

Author: rambleraptor

tests/integration/test_catalog.py

@@ -41,6 +41,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.typedef import Identifier
 from pyiceberg.types import IntegerType, LongType, NestedField, TimestampType, UUIDType
 from tests.conftest import clean_up
 
@@ -635,3 +636,136 @@ def test_rest_custom_namespace_separator(rest_catalog: RestCatalog, table_schema
 
     loaded_table = rest_catalog.load_table(identifier=full_table_identifier_tuple)
     assert loaded_table.name() == full_table_identifier_tuple
+
+def _namespace_exists(catalog: Catalog, namespace: str | Identifier) -> bool:
+    try:
+        catalog.load_namespace_properties(namespace)
+        return True
+    except NoSuchNamespaceError:
+        return False
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_namespace_with_slash(test_catalog: Catalog) -> None:
+    if isinstance(test_catalog, (HiveCatalog, SqlCatalog, RestCatalog)):
+        pytest.skip(f"{type(test_catalog).__name__} does not support slash in namespace")
+
+    namespace = ("new/db",)
+
+    if _namespace_exists(test_catalog, namespace):
+        test_catalog.drop_namespace(namespace)
+
+    assert not _namespace_exists(test_catalog, namespace)
+
+    test_catalog.create_namespace(namespace)
+    assert _namespace_exists(test_catalog, namespace)
+
+    properties = test_catalog.load_namespace_properties(namespace)
+    assert properties is not None
+
+    test_catalog.drop_namespace(namespace)
+    assert not _namespace_exists(test_catalog, namespace)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_namespace_with_dot(test_catalog: Catalog) -> None:
+    if isinstance(test_catalog, (HiveCatalog, SqlCatalog, RestCatalog)):
+        pytest.skip(f"{type(test_catalog).__name__} does not support dot in namespace")
+
+    namespace = ("new.db",)
+
+    if _namespace_exists(test_catalog, namespace):
+        test_catalog.drop_namespace(namespace)
+
+    assert not _namespace_exists(test_catalog, namespace)
+
+    test_catalog.create_namespace(namespace)
+    assert _namespace_exists(test_catalog, namespace)
+
+    # list_namespaces returns a list of tuples
+    assert namespace in test_catalog.list_namespaces()
+
+    properties = test_catalog.load_namespace_properties(namespace)
+    assert properties is not None
+
+    test_catalog.drop_namespace(namespace)
+    assert not _namespace_exists(test_catalog, namespace)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_table_name_with_slash(test_catalog: Catalog, table_schema_simple: Schema) -> None:
+    if isinstance(test_catalog, (HiveCatalog, SqlCatalog, RestCatalog)):
+        pytest.skip(f"{type(test_catalog).__name__} does not support slash in table name")
+
+    namespace = ("ns_slash",)
+    table_ident = ("ns_slash", "tab/le")
+
+    if not _namespace_exists(test_catalog, namespace):
+        test_catalog.create_namespace(namespace)
+
+    if test_catalog.table_exists(table_ident):
+        test_catalog.drop_table(table_ident)
+
+    assert not test_catalog.table_exists(table_ident)
+
+    test_catalog.create_table(table_ident, table_schema_simple)
+    assert test_catalog.table_exists(table_ident)
+
+    table = test_catalog.load_table(table_ident)
+    assert table.schema().as_struct() == table_schema_simple.as_struct()
+
+    test_catalog.drop_table(table_ident)
+    assert not test_catalog.table_exists(table_ident)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_table_name_with_dot(test_catalog: Catalog, table_schema_simple: Schema) -> None:
+    if isinstance(test_catalog, (HiveCatalog, SqlCatalog, RestCatalog)):
+        pytest.skip(f"{type(test_catalog).__name__} does not support dot in table name")
+
+    namespace = ("ns_dot",)
+    table_ident = ("ns_dot", "ta.ble")
+
+    if not _namespace_exists(test_catalog, namespace):
+        test_catalog.create_namespace(namespace)
+
+    if test_catalog.table_exists(table_ident):
+        test_catalog.drop_table(table_ident)
+
+    assert not test_catalog.table_exists(table_ident)
+
+    test_catalog.create_table(table_ident, table_schema_simple)
+    assert test_catalog.table_exists(table_ident)
+
+    assert table_ident in test_catalog.list_tables(namespace)
+
+    table = test_catalog.load_table(table_ident)
+    assert table.schema().as_struct() == table_schema_simple.as_struct()
+
+    test_catalog.drop_table(table_ident)
+    assert not test_catalog.table_exists(table_ident)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_drop_missing_table(test_catalog: Catalog, database_name: str) -> None:
+    test_catalog.create_namespace_if_not_exists(database_name)
+    table_ident = (database_name, "missing_table")
+    assert not test_catalog.table_exists(table_ident)
+    with pytest.raises(NoSuchTableError):
+        test_catalog.drop_table(table_ident)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("test_catalog", CATALOGS)
+def test_drop_nonexistent_namespace(test_catalog: Catalog) -> None:
+    if isinstance(test_catalog, HiveCatalog):
+        pytest.skip("HiveCatalog raises NoSuchObjectException instead of NoSuchNamespaceError")
+
+    namespace = ("non_existent_namespace",)
+    with pytest.raises(NoSuchNamespaceError):
+        test_catalog.drop_namespace(namespace)