test(bigquery): add integration tests for dataset admin operations

src/auth/integration-tests/Cargo.toml

@@ -28,6 +28,7 @@ gax           = { path = "../../../src/gax", package = "google-cloud-gax" }
 language      = { path = "../../../src/generated/cloud/language/v2", package = "google-cloud-language-v2" }
 scoped-env    = "2"
 secretmanager = { path = "../../../src/generated/cloud/secretmanager/v1", package = "google-cloud-secretmanager-v1" }
+bigquery = { path = "../../../src/generated/cloud/bigquery/v2", package = "google-cloud-bigquery-v2" }
 tempfile      = "3"
 
 [dev-dependencies]

src/integration-tests/Cargo.toml

@@ -41,6 +41,10 @@ wkt.workspace         = true
 package = "google-cloud-firestore"
 path    = "../../src/firestore"
 
+[dependencies.bigquery]
+package = "google-cloud-bigquery-v2"
+path    = "../../src/generated/cloud/bigquery/v2"
+
 [dependencies.sm]
 package = "google-cloud-secretmanager-v1"
 path    = "../../src/generated/cloud/secretmanager/v1"

src/integration-tests/src/bigquery.rs

@@ -0,0 +1,145 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::Result;
+use gax::error::Error;
+use rand::{Rng, distr::Alphanumeric};
+
+pub async fn dataset_admin(
+    builder: bigquery::builder::dataset_service::ClientBuilder,
+) -> Result<()> {
+    // Enable a basic subscriber. Useful to troubleshoot problems and visually
+    // verify tracing is doing something.
+    #[cfg(feature = "log-integration-tests")]
+    let _guard = {
+        use tracing_subscriber::fmt::format::FmtSpan;
+        let subscriber = tracing_subscriber::fmt()
+            .with_level(true)
+            .with_thread_ids(true)
+            .with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
+            .finish();
+
+        tracing::subscriber::set_default(subscriber)
+    };
+
+    let project_id = crate::project_id()?;
+    let client = builder.build().await?;
+    cleanup_stale_datasets(&client, &project_id).await?;
+
+    let rand_suffix: String = rand::rng()
+        .sample_iter(&Alphanumeric)
+        .take(8)
+        .map(char::from)
+        .collect();
+
+    let ds_name = format!("rust_bq_test_dataset_{rand_suffix}");
+
+    println!("CREATING DATASET WITH ID: {ds_name}");
+
+    let create = client
+        .insert_dataset(&project_id)
+        .set_dataset(bigquery::model::Dataset::new().set_dataset_reference(
+            bigquery::model::DatasetReference::new().set_dataset_id(&ds_name),
+        ))
+        .send()
+        .await?;
+    println!("CREATE DATASET = {create:?}");
+
+    assert!(create.dataset_reference.is_some());
+
+    let list = client.list_datasets(&project_id).send().await?;
+    println!("LIST DATASET = {} entries", list.datasets.len());
+
+    assert!(!list.datasets.is_empty());
+    assert!(list.datasets.len() > 1);
+    assert!(
+        list.datasets
+            .iter()
+            .find(|v| v.id.contains(&ds_name))
+            .is_some()
+    );
+
+    let delete = client
+        .delete_dataset(&project_id, &ds_name)
+        .set_delete_contents(true)
+        .send()
+        .await?;
+    println!("DELETE DATASET = {delete:?}");
+
+    //assert!(delete().is_ok());
+
+    Ok(())
+}
+
+async fn cleanup_stale_datasets(
+    client: &bigquery::client::DatasetService,
+    project_id: &str,
+) -> Result<()> {
+    use std::time::{Duration, SystemTime, UNIX_EPOCH};
+    let stale_deadline = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_err(Error::other)?;
+    let stale_deadline = stale_deadline - Duration::from_secs(48 * 60 * 60);
+    let stale_deadline = stale_deadline.as_millis() as i64;
+
+    let list = client.list_datasets(project_id).send().await?;
+    let pending_all_datasets = list
+        .datasets
+        .iter()
+        .map(|v| {
+            return client
+                .get_dataset(
+                    project_id,
+                    v.dataset_reference.as_ref().map_or("", |v| &v.dataset_id),
+                )
+                .send();
+        })
+        .collect::<Vec<_>>();
+
+    let stale_datasets = futures::future::join_all(pending_all_datasets)
+        .await
+        .into_iter()
+        .filter_map(|r| {
+            if r.as_ref()
+                .is_ok_and(|ds| ds.creation_time < stale_deadline && ds.id.contains("bq_rust"))
+            {
+                return r.ok();
+            }
+            return None;
+        })
+        .collect::<Vec<_>>();
+
+    println!("found {} stale datasets", stale_datasets.len());
+
+    let pending_deletion: Vec<_> = stale_datasets
+        .iter()
+        .map(|ds| {
+            client
+                .delete_dataset(
+                    project_id,
+                    ds.dataset_reference.as_ref().map_or("", |v| &v.dataset_id),
+                )
+                .set_delete_contents(true)
+                .send()
+        })
+        .collect();
+
+    futures::future::join_all(pending_deletion)
+        .await
+        .into_iter()
+        .zip(stale_datasets)
+        .for_each(|(r, ds)| println!("{} = {r:?}", ds.id));
+
+    Ok(())
+}

src/integration-tests/src/lib.rs

@@ -14,6 +14,7 @@
 
 use gax::error::Error;
 pub type Result<T> = std::result::Result<T, gax::error::Error>;
+pub mod bigquery;
 pub mod error_details;
 pub mod firestore;
 pub mod secret_manager;

src/integration-tests/tests/driver.rs

@@ -42,6 +42,18 @@ mod driver {
             .map_err(report)
     }
 
+    #[test_case(bigquery::client::DatasetService::builder(); "default")]
+    #[test_case(bigquery::client::DatasetService::builder().with_tracing(); "with tracing enabled")]
+    #[test_case(bigquery::client::DatasetService::builder().with_retry_policy(retry_policy()); "with retry enabled")]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn run_bigquery(
+        builder: bigquery::builder::dataset_service::ClientBuilder,
+    ) -> integration_tests::Result<()> {
+        integration_tests::bigquery::dataset_admin(builder)
+            .await
+            .map_err(report)
+    }
+
     #[test_case(sm::client::SecretManagerService::builder(); "default")]
     #[test_case(sm::client::SecretManagerService::builder().with_tracing(); "with tracing enabled")]
     #[test_case(sm::client::SecretManagerService::builder().with_retry_policy(retry_policy()); "with retry enabled")]

src/wkt/src/empty.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use serde::{Deserialize, Deserializer, Serialize, de::IntoDeserializer};
+
 /// A generic empty message that you can re-use to avoid defining duplicated
 /// empty messages in your APIs. A typical example is to use it as the request
 /// or the response type of an API method. For instance:
@@ -21,10 +23,23 @@
 ///   rpc Bar(google.protobuf.Empty) returns (google.protobuf.Empty);
 /// }
 /// ```
-///
-#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
+#[derive(Clone, Debug, Default, PartialEq, Serialize)]
 pub struct Empty {}
 
+impl<'de> Deserialize<'de> for Empty {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let input = String::deserialize(deserializer).unwrap_or(String::default());
+        if input.trim().is_empty() || input.trim().eq("null") {
+            return Ok(Empty::default());
+        }
+        let string_deserializer = String::into_deserializer(input);
+        Ok(Option::<Empty>::deserialize(string_deserializer)?.unwrap())
+    }
+}
+
 impl crate::message::Message for Empty {
     fn typename() -> &'static str {
         "type.googleapis.com/google.protobuf.Empty"
@@ -51,4 +66,18 @@ mod tests {
         assert_eq!(Empty::default(), got);
         Ok(())
     }
+
+    #[test]
+    fn deserialize_null() -> Result {
+        let got = serde_json::from_value(json!(null))?;
+        assert_eq!(Empty::default(), got);
+        Ok(())
+    }
+
+    #[test]
+    fn deserialize_empty() -> Result {
+        let got = serde_json::from_str("")?;
+        assert_eq!(Empty::default(), got);
+        Ok(())
+    }
 }