Add catalog test suite to unify catalog's behavior.

Cargo.toml

@@ -107,6 +107,7 @@ rand = "0.8.5"
 regex = "1.11.3"
 reqwest = { version = "0.12.12", default-features = false, features = ["json"] }
 roaring = { version = "0.11" }
+rstest = "0.26"
 fastnum = { version = "0.7", default-features = false, features = ["std", "serde"] }
 serde = { version = "1.0.219", features = ["rc"] }
 serde_bytes = "0.11.17"

crates/catalog/glue/Cargo.toml

@@ -36,7 +36,6 @@ aws-sdk-glue = { workspace = true }
 iceberg = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true }
-tracing = { workspace = true }
 
 [dev-dependencies]
 iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }

crates/catalog/glue/src/catalog.rs

@@ -314,6 +314,13 @@ impl Catalog for GlueCatalog {
         namespace: &NamespaceIdent,
         properties: HashMap<String, String>,
     ) -> Result<Namespace> {
+        if self.namespace_exists(namespace).await? {
+            return Err(Error::new(
+                ErrorKind::NamespaceAlreadyExists,
+                format!("Namespace {namespace:?} already exists"),
+            ));
+        }
+
         let db_input = convert_to_database(namespace, &properties)?;
 
         let builder = self.client.0.create_database().database_input(db_input);
@@ -340,15 +347,27 @@ impl Catalog for GlueCatalog {
         let builder = self.client.0.get_database().name(&db_name);
         let builder = with_catalog_id!(builder, self.config);
 
-        let resp = builder.send().await.map_err(from_aws_sdk_error)?;
+        let resp = builder.send().await.map_err(|err| {
+            if err
+                .as_service_error()
+                .map(|e| e.is_entity_not_found_exception())
+                == Some(true)
+            {
+                return Error::new(
+                    ErrorKind::NamespaceNotFound,
+                    format!("Namespace {namespace:?} does not exist"),
+                );
+            }
+            from_aws_sdk_error(err)
+        })?;
 
         match resp.database() {
             Some(db) => {
                 let namespace = convert_to_namespace(db);
                 Ok(namespace)
             }
             None => Err(Error::new(
-                ErrorKind::DataInvalid,
+                ErrorKind::NamespaceNotFound,
                 format!("Database with name: {db_name} does not exist"),
             )),
         }
@@ -404,6 +423,13 @@ impl Catalog for GlueCatalog {
         namespace: &NamespaceIdent,
         properties: HashMap<String, String>,
     ) -> Result<()> {
+        if !self.namespace_exists(namespace).await? {
+            return Err(Error::new(
+                ErrorKind::NamespaceNotFound,
+                format!("Namespace {namespace:?} does not exist"),
+            ));
+        }
+
         let db_name = validate_namespace(namespace)?;
         let db_input = convert_to_database(namespace, &properties)?;
 
@@ -431,6 +457,13 @@ impl Catalog for GlueCatalog {
     /// - `Err(...)` signifies failure to drop the namespace due to validation
     /// errors, connectivity issues, or Glue Catalog constraints.
     async fn drop_namespace(&self, namespace: &NamespaceIdent) -> Result<()> {
+        if !self.namespace_exists(namespace).await? {
+            return Err(Error::new(
+                ErrorKind::NamespaceNotFound,
+                format!("Namespace {namespace:?} does not exist"),
+            ));
+        }
+
         let db_name = validate_namespace(namespace)?;
         let table_list = self.list_tables(namespace).await?;