mod common;
use std::time::Duration;
use assert_matches::assert_matches;
use common::{S2Basin, SharedS2Basin, unique_stream_name, uuid};
use futures_util::StreamExt;
use s2_sdk::{
error::{RequestError, ServerError},
types::*,
};
use test_context::test_context;
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_list_and_delete_stream(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let stream_info = basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
assert_eq!(stream_info.name, stream_name);
let page = basin.list_streams(ListStreamsInput::new()).await?;
assert_eq!(page.values, vec![stream_info]);
assert!(!page.has_more);
basin
.delete_stream(DeleteStreamInput::new(stream_name.clone()))
.await?;
let page = basin.list_streams(ListStreamsInput::new()).await?;
match page.values.as_slice() {
[] => {}
[
StreamInfo {
name,
deleted_at: Some(_),
..
},
] => {
assert_eq!(name, &stream_name);
}
values => panic!("unexpected stream listing after delete: {values:?}"),
}
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn stream_config_roundtrip(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new()
.with_storage_class(StorageClass::Standard)
.with_retention_policy(RetentionPolicy::Age(3600))
.with_timestamping(TimestampingConfig::new().with_mode(TimestampingMode::ClientRequire));
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config.clone()))
.await?;
let retrieved_config = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(
retrieved_config,
StreamConfig {
storage_class: Some(StorageClass::Standard),
retention_policy: Some(RetentionPolicy::Age(3600)),
timestamping: Some(TimestampingConfig {
mode: Some(TimestampingMode::ClientRequire),
..
}),
..
}
);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let new_config = StreamReconfiguration::new().with_delete_on_empty(
DeleteOnEmptyReconfiguration::new().with_min_age(Duration::from_hours(12)),
);
let updated_config = basin
.reconfigure_stream(ReconfigureStreamInput::new(stream_name.clone(), new_config))
.await?;
assert_matches!(
updated_config.delete_on_empty,
Some(DeleteOnEmptyConfig {
min_age_secs: 43200,
..
})
);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn ensure_stream_created(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let delete_on_empty_config = DeleteOnEmptyConfig::new().with_min_age(Duration::from_hours(24));
let output = basin
.ensure_stream(
EnsureStreamInput::new(stream_name.clone())
.with_config(StreamConfig::new().with_delete_on_empty(delete_on_empty_config)),
)
.await?;
assert_matches!(output, EnsureOutput::Created(info) => {
assert_eq!(stream_name, info.name);
});
let config = basin.get_stream_config(stream_name).await?;
assert_eq!(config.delete_on_empty, Some(delete_on_empty_config));
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn ensure_stream_config_updated(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let output = basin
.ensure_stream(EnsureStreamInput::new(stream_name.clone()).with_config(
StreamConfig::new().with_timestamping(
TimestampingConfig::new().with_mode(TimestampingMode::ClientRequire),
),
))
.await?;
assert_matches!(output, EnsureOutput::Created(info) => {
assert_eq!(stream_name, info.name);
});
let output =
basin
.ensure_stream(EnsureStreamInput::new(stream_name.clone()).with_config(
StreamConfig::new().with_timestamping(
TimestampingConfig::new().with_mode(TimestampingMode::Arrival),
),
))
.await?;
assert_matches!(output, EnsureOutput::ConfigUpdated(_));
let updated_config = basin.get_stream_config(stream_name).await?;
assert_matches!(
updated_config.timestamping,
Some(TimestampingConfig {
mode: Some(TimestampingMode::Arrival),
..
})
);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn ensure_basin_config_unchanged(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let output = basin
.ensure_stream(EnsureStreamInput::new(stream_name.clone()))
.await?;
assert_matches!(output, EnsureOutput::Created(info) => {
assert_eq!(stream_name, info.name);
});
let config = basin.get_stream_config(stream_name.clone()).await?;
let output = basin
.ensure_stream(EnsureStreamInput::new(stream_name.clone()).with_config(config.clone()))
.await?;
assert_matches!(output, EnsureOutput::ConfigUnchanged(_));
let updated_config = basin.get_stream_config(stream_name).await?;
assert_eq!(config, updated_config);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_limit(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name_1 = unique_stream_name();
let stream_name_2 = unique_stream_name();
let stream_name_3 = unique_stream_name();
let stream_info_1 = basin
.create_stream(CreateStreamInput::new(stream_name_1.clone()))
.await?;
let _stream_info_2 = basin
.create_stream(CreateStreamInput::new(stream_name_2.clone()))
.await?;
let _stream_info_3 = basin
.create_stream(CreateStreamInput::new(stream_name_3.clone()))
.await?;
let page = basin
.list_streams(ListStreamsInput::new().with_limit(1))
.await?;
assert_eq!(page.values, vec![stream_info_1]);
assert!(page.has_more);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_prefix(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name_1: StreamName = "users/eu/0001".parse().expect("valid stream name");
let stream_name_2: StreamName = "users/ca/0001".parse().expect("valid stream name");
let stream_name_3: StreamName = "users/ca/0002".parse().expect("valid stream name");
let _stream_info_1 = basin
.create_stream(CreateStreamInput::new(stream_name_1.clone()))
.await?;
let stream_info_2 = basin
.create_stream(CreateStreamInput::new(stream_name_2.clone()))
.await?;
let stream_info_3 = basin
.create_stream(CreateStreamInput::new(stream_name_3.clone()))
.await?;
let page = basin
.list_streams(
ListStreamsInput::new().with_prefix("users/ca/".parse().expect("valid prefix")),
)
.await?;
assert_eq!(page.values, vec![stream_info_2, stream_info_3]);
assert!(!page.has_more);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_start_after(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name_1 = unique_stream_name();
let stream_name_2 = unique_stream_name();
let _stream_info_1 = basin
.create_stream(CreateStreamInput::new(stream_name_1.clone()))
.await?;
let stream_info_2 = basin
.create_stream(CreateStreamInput::new(stream_name_2.clone()))
.await?;
let page = basin
.list_streams(
ListStreamsInput::new()
.with_start_after(stream_name_1.parse().expect("valid start after")),
)
.await?;
assert_eq!(page.values, vec![stream_info_2]);
assert!(!page.has_more);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_start_after_returns_empty_page(
basin: &S2Basin,
) -> Result<(), RequestError> {
let stream_name_1 = unique_stream_name();
let stream_name_2 = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name_1.clone()))
.await?;
basin
.create_stream(CreateStreamInput::new(stream_name_2.clone()))
.await?;
let page = basin
.list_streams(
ListStreamsInput::new()
.with_start_after(stream_name_2.parse().expect("valid start after")),
)
.await?;
assert_eq!(page.values.len(), 0);
assert!(!page.has_more);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn delete_nonexistent_stream_errors(basin: &S2Basin) -> Result<(), RequestError> {
let result = basin
.delete_stream(DeleteStreamInput::new(unique_stream_name()))
.await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert_eq!(code, "stream_not_found")
}
);
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn delete_nonexistent_stream_with_ignore(basin: &S2Basin) -> Result<(), RequestError> {
let result = basin
.delete_stream(DeleteStreamInput::new(unique_stream_name()).with_ignore_not_found(true))
.await;
assert_matches!(result, Ok(()));
Ok(())
}
#[test_context(S2Basin)]
#[tokio_shared_rt::test(shared)]
async fn get_stream_config(basin: &S2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new().with_storage_class(StorageClass::Express);
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await?;
let retrieved_config = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(retrieved_config.storage_class, Some(StorageClass::Express));
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_limit_zero(basin: &SharedS2Basin) -> Result<(), RequestError> {
let prefix = format!("limit0-{}", uuid());
let stream_name: StreamName = format!("{}-0001", prefix)
.parse()
.expect("valid stream name");
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let page = basin
.list_streams(
ListStreamsInput::new()
.with_prefix(prefix.parse().expect("valid prefix"))
.with_limit(0),
)
.await?;
assert!(page.values.iter().any(|info| info.name == stream_name));
assert!(page.values.len() <= 1000);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_limit_over_max(basin: &SharedS2Basin) -> Result<(), RequestError> {
let prefix = format!("limitmax-{}", uuid());
let stream_name: StreamName = format!("{}-0001", prefix)
.parse()
.expect("valid stream name");
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let page = basin
.list_streams(
ListStreamsInput::new()
.with_prefix(prefix.parse().expect("valid prefix"))
.with_limit(1500),
)
.await?;
assert!(page.values.iter().any(|info| info.name == stream_name));
assert!(page.values.len() <= 1000);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_with_pagination(basin: &SharedS2Basin) -> Result<(), RequestError> {
let prefix = format!("page-{}", uuid());
let stream_names: Vec<StreamName> = (0..3)
.map(|idx| {
format!("{}-{:04}", prefix, idx)
.parse()
.expect("valid stream name")
})
.collect();
for name in &stream_names {
basin
.create_stream(CreateStreamInput::new(name.clone()))
.await?;
}
let page_1 = basin
.list_streams(
ListStreamsInput::new()
.with_prefix(prefix.parse().expect("valid prefix"))
.with_limit(2),
)
.await?;
assert!(!page_1.values.is_empty());
let last_name = page_1
.values
.last()
.expect("page should have value")
.name
.clone();
let page_2 = basin
.list_streams(
ListStreamsInput::new()
.with_prefix(prefix.parse().expect("valid prefix"))
.with_start_after(last_name.clone().into())
.with_limit(2),
)
.await?;
assert!(
page_2
.values
.iter()
.all(|info| info.name.as_ref() > last_name.as_ref())
);
let mut listed: Vec<String> = page_1
.values
.into_iter()
.chain(page_2.values.into_iter())
.map(|info| info.name.to_string())
.collect();
listed.sort();
let mut expected: Vec<String> = stream_names.iter().map(|name| name.to_string()).collect();
expected.sort();
assert_eq!(listed, expected);
for name in stream_names {
let _ = basin.delete_stream(DeleteStreamInput::new(name)).await;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_streams_returns_lexicographic_order(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let prefix = format!("order-{}", uuid());
let stream_names: Vec<StreamName> = (1..=3)
.map(|idx| {
format!("{}-{:04}", prefix, idx)
.parse()
.expect("valid stream name")
})
.collect();
for name in &stream_names {
basin
.create_stream(CreateStreamInput::new(name.clone()))
.await?;
}
let page = basin
.list_streams(ListStreamsInput::new().with_prefix(prefix.parse().expect("valid prefix")))
.await?;
let listed: Vec<StreamName> = page.values.into_iter().map(|info| info.name).collect();
assert_eq!(listed, stream_names);
for name in stream_names {
let _ = basin.delete_stream(DeleteStreamInput::new(name)).await;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_all_streams_iterates_with_prefix(
basin: &SharedS2Basin,
) -> Result<(), Box<dyn std::error::Error>> {
let prefix = format!("iter-{}", uuid());
let stream_names: Vec<StreamName> = (1..=3)
.map(|idx| {
format!("{}-{:04}", prefix, idx)
.parse()
.expect("valid stream name")
})
.collect();
for name in &stream_names {
basin
.create_stream(CreateStreamInput::new(name.clone()))
.await?;
}
let mut listed = Vec::new();
let mut stream = basin.list_all_streams(
ListAllStreamsInput::new().with_prefix(prefix.parse().expect("valid prefix")),
);
while let Some(info) = stream.next().await {
listed.push(info?.name);
}
assert_eq!(listed, stream_names);
for name in stream_names {
let _ = basin.delete_stream(DeleteStreamInput::new(name)).await;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn list_all_streams_include_deleted(
basin: &SharedS2Basin,
) -> Result<(), Box<dyn std::error::Error>> {
let prefix = format!("iter-del-{}", uuid());
let stream_name: StreamName = format!("{}-0001", prefix)
.parse()
.expect("valid stream name");
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
basin
.delete_stream(DeleteStreamInput::new(stream_name.clone()))
.await?;
let mut stream = basin.list_all_streams(
ListAllStreamsInput::new()
.with_prefix(prefix.parse().expect("valid prefix"))
.with_include_deleted(true),
);
let mut found = None;
while let Some(info) = stream.next().await {
let info = info?;
if info.name == stream_name {
found = Some(info);
break;
}
}
if let Some(info) = found {
assert!(info.deleted_at.is_some());
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_with_full_config(basin: &SharedS2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new()
.with_storage_class(StorageClass::Standard)
.with_retention_policy(RetentionPolicy::Age(86400))
.with_timestamping(
TimestampingConfig::new()
.with_mode(TimestampingMode::ClientRequire)
.with_uncapped(true),
)
.with_delete_on_empty(DeleteOnEmptyConfig::new().with_min_age(Duration::from_secs(3600)));
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await?;
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(
retrieved,
StreamConfig {
storage_class: Some(StorageClass::Standard),
retention_policy: Some(RetentionPolicy::Age(86400)),
timestamping: Some(TimestampingConfig {
mode: Some(TimestampingMode::ClientRequire),
uncapped: Some(true),
..
}),
delete_on_empty: Some(DeleteOnEmptyConfig {
min_age_secs: 3600,
..
}),
..
}
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_storage_class_express(basin: &SharedS2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new().with_storage_class(StorageClass::Express);
let result = basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await;
let info = match result {
Ok(info) => info,
Err(err) if is_free_tier_limitation(&err) => return Ok(()),
Err(err) => return Err(err),
};
assert_eq!(info.name, stream_name);
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(retrieved.storage_class, Some(StorageClass::Express) | None);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_retention_policy_infinite(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new().with_retention_policy(RetentionPolicy::Infinite);
let result = basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await;
let info = match result {
Ok(info) => info,
Err(err) if is_free_tier_limitation(&err) => return Ok(()),
Err(err) => return Err(err),
};
assert_eq!(info.name, stream_name);
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(retrieved.retention_policy, Some(RetentionPolicy::Infinite));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_timestamping_modes(basin: &SharedS2Basin) -> Result<(), RequestError> {
let modes = [
TimestampingMode::ClientPrefer,
TimestampingMode::ClientRequire,
TimestampingMode::Arrival,
];
for mode in modes {
let stream_name = unique_stream_name();
let config =
StreamConfig::new().with_timestamping(TimestampingConfig::new().with_mode(mode));
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await?;
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
match mode {
TimestampingMode::ClientPrefer => {
if let Some(timestamping) = retrieved.timestamping {
assert_matches!(
timestamping.mode,
Some(TimestampingMode::ClientPrefer) | None
);
}
}
_ => {
assert_matches!(
retrieved.timestamping,
Some(TimestampingConfig {
mode: Some(m),
..
}) if m == mode
);
}
}
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_timestamping_uncapped(basin: &SharedS2Basin) -> Result<(), RequestError> {
for uncapped in [true, false] {
let stream_name = unique_stream_name();
let config = StreamConfig::new()
.with_timestamping(TimestampingConfig::new().with_uncapped(uncapped));
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await?;
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
let timestamping = retrieved
.timestamping
.expect("explicit uncapped setting should be preserved");
assert_eq!(timestamping.uncapped, Some(uncapped));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_delete_on_empty_min_age(basin: &SharedS2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new()
.with_delete_on_empty(DeleteOnEmptyConfig::new().with_min_age(Duration::from_secs(3600)));
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(config))
.await?;
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(
retrieved.delete_on_empty,
Some(DeleteOnEmptyConfig {
min_age_secs: 3600,
..
})
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_invalid_retention_age_zero(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
let config = StreamConfig::new().with_retention_policy(RetentionPolicy::Age(0));
let result = basin
.create_stream(CreateStreamInput::new(stream_name).with_config(config))
.await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert_eq!(code, "invalid");
}
);
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn get_stream_config_nonexistent_errors(basin: &SharedS2Basin) -> Result<(), RequestError> {
let result = basin.get_stream_config(unique_stream_name()).await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert_eq!(code, "stream_not_found");
}
);
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_storage_class_standard(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let config = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_storage_class(StorageClass::Standard),
))
.await?;
assert_matches!(config.storage_class, Some(StorageClass::Standard));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_storage_class_express(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let result = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_storage_class(StorageClass::Express),
))
.await;
let config = match result {
Ok(config) => config,
Err(err) if is_free_tier_limitation(&err) => {
let _ = basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await;
return Ok(());
}
Err(err) => return Err(err),
};
assert_matches!(config.storage_class, Some(StorageClass::Express) | None);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_retention_policy_age(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let config = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_retention_policy(RetentionPolicy::Age(3600)),
))
.await?;
assert_matches!(config.retention_policy, Some(RetentionPolicy::Age(3600)));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_retention_policy_infinite(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let result = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_retention_policy(RetentionPolicy::Infinite),
))
.await;
let config = match result {
Ok(config) => config,
Err(err) if is_free_tier_limitation(&err) => {
let _ = basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await;
return Ok(());
}
Err(err) => return Err(err),
};
assert_matches!(config.retention_policy, Some(RetentionPolicy::Infinite));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_timestamping_modes(basin: &SharedS2Basin) -> Result<(), RequestError> {
let modes = [
TimestampingMode::ClientPrefer,
TimestampingMode::ClientRequire,
TimestampingMode::Arrival,
];
for mode in modes {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let config = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new()
.with_timestamping(TimestampingReconfiguration::new().with_mode(mode)),
))
.await?;
match mode {
TimestampingMode::ClientPrefer => {
if let Some(timestamping) = config.timestamping {
assert_matches!(
timestamping.mode,
Some(TimestampingMode::ClientPrefer) | None
);
}
}
_ => {
assert_matches!(
config.timestamping,
Some(TimestampingConfig {
mode: Some(m),
..
}) if m == mode
);
}
}
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_timestamping_uncapped(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
for uncapped in [true, false] {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let config = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new()
.with_timestamping(TimestampingReconfiguration::new().with_uncapped(uncapped)),
))
.await?;
let timestamping = config
.timestamping
.expect("explicit uncapped setting should be preserved");
assert_eq!(timestamping.uncapped, Some(uncapped));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_delete_on_empty(basin: &SharedS2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let config = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_delete_on_empty(
DeleteOnEmptyReconfiguration::new().with_min_age(Duration::from_secs(3600)),
),
))
.await?;
assert_matches!(
config.delete_on_empty,
Some(DeleteOnEmptyConfig {
min_age_secs: 3600,
..
})
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_disable_delete_on_empty(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()).with_config(
StreamConfig::new().with_delete_on_empty(
DeleteOnEmptyConfig::new().with_min_age(Duration::from_secs(3600)),
),
))
.await?;
let config = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_delete_on_empty(
DeleteOnEmptyReconfiguration::new().with_min_age(Duration::from_secs(0)),
),
))
.await?;
assert!(
config.delete_on_empty.is_none()
|| config.delete_on_empty == Some(DeleteOnEmptyConfig::new())
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_empty_config_no_change(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(
CreateStreamInput::new(stream_name.clone()).with_config(
StreamConfig::new()
.with_storage_class(StorageClass::Standard)
.with_retention_policy(RetentionPolicy::Age(3600)),
),
)
.await?;
let _ = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new(),
))
.await?;
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(retrieved.storage_class, Some(StorageClass::Standard));
assert_matches!(retrieved.retention_policy, Some(RetentionPolicy::Age(3600)));
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_partial_update(basin: &SharedS2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(
CreateStreamInput::new(stream_name.clone()).with_config(
StreamConfig::new()
.with_retention_policy(RetentionPolicy::Age(3600))
.with_timestamping(
TimestampingConfig::new().with_mode(TimestampingMode::ClientPrefer),
),
),
)
.await?;
let _ = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_timestamping(
TimestampingReconfiguration::new().with_mode(TimestampingMode::Arrival),
),
))
.await?;
let retrieved = basin.get_stream_config(stream_name.clone()).await?;
assert_matches!(retrieved.retention_policy, Some(RetentionPolicy::Age(3600)));
assert_matches!(
retrieved.timestamping,
Some(TimestampingConfig {
mode: Some(TimestampingMode::Arrival),
..
})
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_invalid_retention_age_zero(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let result = basin
.reconfigure_stream(ReconfigureStreamInput::new(
stream_name.clone(),
StreamReconfiguration::new().with_retention_policy(RetentionPolicy::Age(0)),
))
.await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert_eq!(code, "invalid");
}
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn reconfigure_stream_nonexistent_errors(basin: &SharedS2Basin) -> Result<(), RequestError> {
let result = basin
.reconfigure_stream(ReconfigureStreamInput::new(
unique_stream_name(),
StreamReconfiguration::new().with_storage_class(StorageClass::Standard),
))
.await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert_eq!(code, "stream_not_found");
}
);
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn create_stream_duplicate_name_errors(basin: &SharedS2Basin) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
let result = basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert_eq!(code, "resource_already_exists");
}
);
basin
.delete_stream(DeleteStreamInput::new(stream_name))
.await?;
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn delete_stream_already_deleting_is_idempotent(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
basin
.delete_stream(DeleteStreamInput::new(stream_name.clone()))
.await?;
let result = basin
.delete_stream(DeleteStreamInput::new(stream_name.clone()))
.await;
match result {
Ok(()) => {}
Err(RequestError::Server(ServerError { code, .. })) if code == "stream_not_found" => {}
Err(err) => return Err(err),
}
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn get_stream_config_for_deleting_stream_errors(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
basin
.delete_stream(DeleteStreamInput::new(stream_name.clone()))
.await?;
let result = basin.get_stream_config(stream_name.clone()).await;
assert_matches!(
result,
Err(RequestError::Server(ServerError { code, .. })) => {
assert!(code == "stream_deletion_pending" || code == "stream_not_found");
}
);
Ok(())
}
#[test_context(SharedS2Basin)]
#[tokio_shared_rt::test(shared)]
async fn deleted_stream_has_deleted_at_when_listed(
basin: &SharedS2Basin,
) -> Result<(), RequestError> {
let stream_name = unique_stream_name();
basin
.create_stream(CreateStreamInput::new(stream_name.clone()))
.await?;
basin
.delete_stream(DeleteStreamInput::new(stream_name.clone()))
.await?;
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
let page = basin
.list_streams(ListStreamsInput::new().with_prefix(stream_name.clone().into()))
.await?;
let mut found = false;
for info in page.values {
if info.name == stream_name {
found = true;
if info.deleted_at.is_some() {
return Ok(());
}
}
}
if !found {
return Ok(());
}
if std::time::Instant::now() >= deadline {
break;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
panic!("deleted stream still listed without deleted_at after timeout");
}
fn is_free_tier_limitation(err: &RequestError) -> bool {
match err {
RequestError::Server(ServerError { code, message, .. }) if code == "invalid" => {
message.to_lowercase().contains("free tier")
}
_ => false,
}
}