test(bigquery): add integration tests for dataset admin operations #1914
Conversation
product-auto-label
bot
added
the
api: bigquery
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## main #1914 +/- ##
==========================================
- Coverage 95.68% 95.24% -0.44%
==========================================
Files 57 58 +1
Lines 2061 2102 +41
==========================================
+ Hits 1972 2002 +30
- Misses 89 100 +11 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
| 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}"); |
There was a problem hiding this comment.
In general we prefer using labels to identify test resources. Can we do that here?
| assert!(!list.datasets.is_empty()); | ||
| assert!(list.datasets.len() > 1); | ||
| assert!( | ||
| list.datasets | ||
| .iter() | ||
| .find(|v| v.id.contains(&ds_name)) | ||
| .is_some() | ||
| ); |
There was a problem hiding this comment.
All of this can be changed to:
| assert!(!list.datasets.is_empty()); | |
| assert!(list.datasets.len() > 1); | |
| assert!( | |
| list.datasets | |
| .iter() | |
| .find(|v| v.id.contains(&ds_name)) | |
| .is_some() | |
| ); | |
| assert!( | |
| list.datasets | |
| .iter() | |
| .find(|v| v.id.contains(&ds_name)) | |
| .is_some(), | |
| "{:?}", list.datasets | |
| ); |
Separately, why "contains"? Can we make that a more specific predicate?
There was a problem hiding this comment.
the Dataset id contains the fully qualified id of the dataset, so it would be something like projects/{projectId}/datasets/{datasetId}, so that's why I used contains.
There was a problem hiding this comment.
Yes, I was suggesting we do something more specific, such as:
v.id.strip_suffix(&ds_name).and(Some(true)).unwrap_or_default()or
v.id == format!("projects/{project_id}/datasets/{ds_name}")or (gross):
v.id.strip_prefix("projects/").and_then(|s| s.strip_prefix(&project_id)).and_then(|s| s.strip_prefix("/datasets/")).and_then(|s| s == &ds_name).unwrap_or_default()| return client | ||
| .get_dataset( | ||
| project_id, | ||
| v.dataset_reference.as_ref().map_or("", |v| &v.dataset_id), |
There was a problem hiding this comment.
This deserves some comment explaining what is going on. If the dataset_reference is not set it seems we cannot make a successful call to .get_dataset()? Maybe returning None is more appropriate? or maybe using v.id and stripping the project_id?
Maybe we should have something like:
fn extract_dataset_id(project_id: &str, v: bigquery::model::ListFormatDataset) -> Option<String> {
match v.dataset_reference {
Some(r) => r.dataset_id.clone,
None => v.id.strip_prefix("projects/").strip_prefix(project_id).owned(),
}
}and then we issue the request only if we have Some()...
| 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<_>>(); |
There was a problem hiding this comment.
This is a case where using into_stream() and StreamExt should help.
To start with, we should be able to write:
use futures::StreamExt;
... ... ..
let list = client.list_datasets(project_id).send().await?;
futures::stream::iter(list.datasets.iter())
.flat_map_unordered(16, // limit the number of concurrent requests
async |item| {
if let Some(id) = extract_dataset_id(item) {
return client.get_dataset(project_id, id).send().await;
}
None
})
.flat_map_unordered(16, |result| -> Option<Result<()>> {
let result = result?;
match result {
Err(e) => Some(Err(e)),
Ok(dataset) => {
let id = extract_dataset_id(project_id, dataset.)?; // Blegh, this needs some refactoring
if dataset.creation_time < stale_deadline && ds.labels.find("integration-test") == Some("true") {
return Some(client.delete_dataset(project_id, id).await);
}
None
}
}
});Finally you may need to collect all the errors.
| #[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")] |
There was a problem hiding this comment.
I think we do not need all three cases for each library. One is enough, maybe this?
| #[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")] | |
| #[test_case(bigquery::client::DatasetService::builder().with_retry_policy(retry_policy()).with_tracing(); "with retry and tracing enabled")] |
Add basic admin operation on BigQuery
Datasets. Later the methodcleanup_stale_datasetscan be reused to clean up resources on future BigQuery integration tests, as aDatasetis the highest level in the hierarchy and delete operations can delete all child resources.Towards #1773