From f600af470459ea532867f18302bb2cdf9fbca3c0 Mon Sep 17 00:00:00 2001 From: Chris Date: Thu, 8 Oct 2026 10:47:10 -0700 Subject: [PATCH] test(e2e): compare two RUSTFS_S3_STACK servers over every S3 operation (#8435) Add dual_stack_differential to e2e_test. It starts two rustfs processes from the same binary, stack A on RUSTFS_S3_STACK=legacy and stack B on RUSTFS_E2E_STACK_B (default legacy until rustfs/backlog#2734 T2.5), sends both the same aws-sdk-s3 script covering the 73 methods of impl S3 for FS (a normal request and a 4xx negative each), and compares status, error code, an allowlist of response headers and a few typed body fields. Error message text is never compared. Each run writes /dual-stack/.json with compared_ops and the (op, field, a, b) differences; any difference fails the test. differential_detects_an_injected_difference starts B with RUSTFS_REGION=eu-west-1 and requires GetBucketLocation to be reported. The comparator is pure and unit-tested without a server. Wire the module into the e2e-smoke filter and update the e2e-smoke and e2e-full selection digests and the e2e_test test-count row. Refs rustfs/backlog#2740 --- .config/e2e-full-selection.txt | 4 +- .config/e2e-smoke-selection.txt | 2 +- .config/nextest.toml | 9 + .config/test-count-baseline.txt | 2 +- .../src/dual_stack_differential/compare.rs | 512 ++++++++ .../src/dual_stack_differential/mod.rs | 197 +++ .../src/dual_stack_differential/scenarios.rs | 1103 +++++++++++++++++ crates/e2e_test/src/lib.rs | 6 + 8 files changed, 1831 insertions(+), 4 deletions(-) create mode 100644 crates/e2e_test/src/dual_stack_differential/compare.rs create mode 100644 crates/e2e_test/src/dual_stack_differential/mod.rs create mode 100644 crates/e2e_test/src/dual_stack_differential/scenarios.rs diff --git a/.config/e2e-full-selection.txt b/.config/e2e-full-selection.txt index 8669a2126..bc02ce37a 100644 --- a/.config/e2e-full-selection.txt +++ b/.config/e2e-full-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=98dff0961c6eb21d6128bfd2d4a6100a624ddf1fb7db05dba360a5d3d0a310c2 -sha256-linux=890f5925b87e1bf3710c356253f3b2276d901e156df9ccc04ec9315374de6d22 +sha256-darwin=b9438fd1b82558f1fcd7e5a19dc0ceb0b34b5357d4ce99b7033de3ef019333ab +sha256-linux=d4622517c13283247a022f49962c2cdb69e5ed0e13874fb55e0ab34fb923c044 diff --git a/.config/e2e-smoke-selection.txt b/.config/e2e-smoke-selection.txt index eb68a79f1..45af4ce67 100644 --- a/.config/e2e-smoke-selection.txt +++ b/.config/e2e-smoke-selection.txt @@ -1 +1 @@ -sha256=ba8f392b617cf9f2338142233a417e43927b28096635e628233ce8a60cf49d10 +sha256=58df0d240cd74b916ce8f094ed1d8ad3e136f2f9bac157a155192a8786dae65c diff --git a/.config/nextest.toml b/.config/nextest.toml index 80b2d8398..53951fdd3 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -579,6 +579,14 @@ test-group = 'ecstore-serial-flaky' # the PR-lane e2e coverage for scanner usage snapshots consumed by quota and # admin surfaces. It uses the same single-node, random-port, isolated-temp-dir # fixture as the existing smoke modules. +# +# Dual-stack differential (rustfs/backlog#2740): `dual_stack_differential::` +# starts two single-node servers from the same binary, one per +# RUSTFS_S3_STACK value, on random ports with isolated temp dirs, and runs one +# SDK script covering every operation of `impl S3 for FS` against both. Its +# comparator unit tests need no server. It meets the admission criteria +# unchanged; the module clause admits a new differential test without another +# filter edit. [profile.e2e-smoke] default-filter = """ package(e2e_test) & ( @@ -586,6 +594,7 @@ default-filter = """ | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/) | test(/^reliant::lifecycle::/) | test(/^reliant::tiering::/) + | test(/^dual_stack_differential::/) | test(/^upgrade_compatibility_test::upgrade_write_readiness_tests::/) | test(/^on_demand_migration::(get_basic_test::(get_miss_pulls_inline_and_serves_locally_afterwards|head_miss_answers_from_the_source_without_persisting)|interaction_test::test_odm_admin_config_is_redacted_and_status_counts_match_the_source)$/) ) diff --git a/.config/test-count-baseline.txt b/.config/test-count-baseline.txt index 8eabafeac..2052a12c9 100644 --- a/.config/test-count-baseline.txt +++ b/.config/test-count-baseline.txt @@ -9,7 +9,7 @@ # the PR that adds tests; lowering one is a reviewed statement that tests were # intentionally removed. Regenerate with: # scripts/check_test_count_ratchet.sh --update -e2e_test 870 18 +e2e_test 890 18 rustfs 5490 62 rustfs-audit 70 0 rustfs-checksums 26 0 diff --git a/crates/e2e_test/src/dual_stack_differential/compare.rs b/crates/e2e_test/src/dual_stack_differential/compare.rs new file mode 100644 index 000000000..8e90738bf --- /dev/null +++ b/crates/e2e_test/src/dual_stack_differential/compare.rs @@ -0,0 +1,512 @@ +// Copyright 2024 RustFS Team +// +// 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 +// +// http://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. + +//! What one stack answered, and how two answers are compared. +//! +//! An [`Observation`] keeps only what a client acts on: the HTTP status, the S3 error code, the +//! allowlisted response headers and a few typed fields decoded from the response. The error +//! message is kept for diagnostics and never compared: message wording is judged by the wire +//! corpus, not here. The comparator is pure so it can be tested without a server. + +use serde::Serialize; +use std::collections::{BTreeMap, BTreeSet}; +use std::path::{Path, PathBuf}; + +/// Response headers compared exactly (after [`normalized_header_value`]). +const HEADER_ALLOWLIST: [&str; 7] = [ + "content-type", + "x-amz-version-id", + "etag", + "x-amz-delete-marker", + "content-range", + "accept-ranges", + "x-amz-request-charged", +]; + +/// Every response header with this prefix is compared as well. +const HEADER_ALLOWLIST_PREFIX: &str = "x-amz-server-side-encryption"; + +/// Placeholder for a field one side did not send. +const ABSENT: &str = ""; + +/// The lowercase header name if the comparator looks at this header. +pub(crate) fn compared_header(name: &str) -> Option { + let name = name.to_ascii_lowercase(); + (HEADER_ALLOWLIST.contains(&name.as_str()) || name.starts_with(HEADER_ALLOWLIST_PREFIX)).then_some(name) +} + +/// A version ID is minted per process, so its value never matches across two servers; only +/// whether it is the literal `null` (an unversioned write) carries meaning. +fn normalized_header_value(name: &str, value: &str) -> String { + if name == "x-amz-version-id" && value != "null" { + "".to_owned() + } else { + value.to_owned() + } +} + +/// One request's outcome as a client sees it. +#[derive(Clone, Debug, Serialize)] +pub(crate) struct Observation { + /// The S3 operation, named as in the `s3s::S3` trait (`GetObject`). + pub op: &'static str, + /// Which request of the script this was (`normal`, `negative`, or a setup label). + pub case: &'static str, + /// HTTP status; 0 when no response arrived. + pub status: u16, + /// The S3 error code, or the SDK's error kind when there was no service error. + pub error_code: Option, + /// Allowlisted headers only, lowercase names, normalized values. + pub headers: BTreeMap, + /// Typed values decoded from the response body, never free text. + pub fields: BTreeMap, + /// Diagnostic only; never compared. + #[serde(skip)] + pub message: Option, +} + +impl Observation { + /// Builds an observation, keeping only the allowlisted headers. + pub(crate) fn new<'h>( + op: &'static str, + case: &'static str, + status: u16, + error_code: Option, + headers: impl IntoIterator, + ) -> Self { + let headers = headers + .into_iter() + .filter_map(|(name, value)| compared_header(name).map(|name| (name, value))) + .map(|(name, value)| { + let value = normalized_header_value(&name, value); + (name, value) + }) + .collect(); + Self { + op, + case, + status, + error_code, + headers, + fields: BTreeMap::new(), + message: None, + } + } + + pub(crate) fn with_field(mut self, name: &str, value: impl Into) -> Self { + self.fields.insert(name.to_owned(), value.into()); + self + } + + pub(crate) fn is_success(&self) -> bool { + (200..300).contains(&self.status) + } +} + +/// One field on which the two stacks answered differently. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub(crate) struct Difference { + pub op: String, + pub case: String, + pub field: String, + pub a: String, + pub b: String, +} + +/// Every difference between two transcripts of the same script, matched by `(op, case)`. +pub(crate) fn compare(a: &[Observation], b: &[Observation]) -> Vec { + let mut differences = Vec::new(); + let (index_a, duplicates_a) = index(a); + let (index_b, duplicates_b) = index(b); + for (op, case) in duplicates_a.union(&duplicates_b) { + let count = |side: &[Observation]| { + side.iter() + .filter(|observation| (observation.op, observation.case) == (*op, *case)) + .count() + .to_string() + }; + differences.push(difference(op, case, "duplicate_case", count(a), count(b))); + } + let keys: BTreeSet<_> = index_a.keys().chain(index_b.keys()).copied().collect(); + for key @ (op, case) in keys { + match (index_a.get(&key), index_b.get(&key)) { + (Some(a), Some(b)) => compare_one(a, b, &mut differences), + (a, b) => { + let presence = |side: Option<&&Observation>| if side.is_some() { "present" } else { ABSENT }.to_owned(); + differences.push(difference(op, case, "observation", presence(a), presence(b))); + } + } + } + differences +} + +type Key = (&'static str, &'static str); + +fn index(side: &[Observation]) -> (BTreeMap, BTreeSet) { + let mut index = BTreeMap::new(); + let mut duplicates = BTreeSet::new(); + for observation in side { + let key = (observation.op, observation.case); + if index.insert(key, observation).is_some() { + duplicates.insert(key); + } + } + (index, duplicates) +} + +fn difference(op: &str, case: &str, field: &str, a: String, b: String) -> Difference { + Difference { + op: op.to_owned(), + case: case.to_owned(), + field: field.to_owned(), + a, + b, + } +} + +fn compare_one(a: &Observation, b: &Observation, differences: &mut Vec) { + let mut field = |name: String, value_a: Option<&str>, value_b: Option<&str>| { + if value_a != value_b { + let show = |value: Option<&str>| value.unwrap_or(ABSENT).to_owned(); + differences.push(difference(a.op, a.case, &name, show(value_a), show(value_b))); + } + }; + field( + "status".to_owned(), + Some(a.status.to_string().as_str()), + Some(b.status.to_string().as_str()), + ); + field("error_code".to_owned(), a.error_code.as_deref(), b.error_code.as_deref()); + let headers: BTreeSet<_> = a.headers.keys().chain(b.headers.keys()).collect(); + for name in headers { + field( + format!("header.{name}"), + a.headers.get(name).map(String::as_str), + b.headers.get(name).map(String::as_str), + ); + } + let fields: BTreeSet<_> = a.fields.keys().chain(b.fields.keys()).collect(); + for name in fields { + field( + format!("body.{name}"), + a.fields.get(name).map(String::as_str), + b.fields.get(name).map(String::as_str), + ); + } +} + +/// Operations observed on both sides. +pub(crate) fn compared_ops(a: &[Observation], b: &[Observation]) -> BTreeSet<&'static str> { + let ops_b: BTreeSet<_> = b.iter().map(|observation| observation.op).collect(); + a.iter() + .map(|observation| observation.op) + .filter(|op| ops_b.contains(op)) + .collect() +} + +/// `(op, case)` pairs observed on both sides. +pub(crate) fn compared_cases(a: &[Observation], b: &[Observation]) -> usize { + let cases_b: BTreeSet<_> = b.iter().map(|observation| (observation.op, observation.case)).collect(); + a.iter() + .map(|observation| (observation.op, observation.case)) + .collect::>() + .intersection(&cases_b) + .count() +} + +/// The JSON report written for every run. +#[derive(Debug, Serialize)] +pub(crate) struct Report { + pub schema: u32, + pub run: String, + pub stack_a: String, + pub stack_b: String, + pub region_a: Option, + pub region_b: Option, + pub compared_ops: usize, + pub compared_cases: usize, + pub differences: Vec, +} + +impl Report { + /// Writes the report as `/.json` and returns its path. + pub(crate) fn write_to(&self, directory: &Path) -> std::io::Result { + std::fs::create_dir_all(directory)?; + let path = directory.join(format!("{}.json", self.run)); + let json = serde_json::to_vec_pretty(self).map_err(std::io::Error::other)?; + std::fs::write(&path, json)?; + Ok(path) + } +} + +/// One line per difference, for a failing assertion. +pub(crate) fn render(differences: &[Difference]) -> String { + differences + .iter() + .map(|d| format!("({}, {}, {}, a={:?}, b={:?})", d.op, d.case, d.field, d.a, d.b)) + .collect::>() + .join("\n") +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ok(op: &'static str) -> Observation { + Observation::new(op, "normal", 200, None, [("content-type", "application/xml")]) + } + + fn fields(differences: &[Difference]) -> Vec<&str> { + differences.iter().map(|d| d.field.as_str()).collect() + } + + #[test] + fn identical_transcripts_have_no_difference() { + let a = vec![ + ok("HeadBucket"), + ok("GetBucketLocation").with_field("location_constraint", "us-east-1"), + ]; + let b = a.clone(); + assert_eq!(compare(&a, &b), Vec::new()); + assert_eq!(compared_ops(&a, &b).len(), 2); + } + + #[test] + fn a_status_difference_is_reported() { + let a = vec![ok("PutBucketAcl")]; + let b = vec![Observation::new( + "PutBucketAcl", + "normal", + 403, + None, + [("content-type", "application/xml")], + )]; + assert_eq!( + compare(&a, &b), + vec![Difference { + op: "PutBucketAcl".into(), + case: "normal".into(), + field: "status".into(), + a: "200".into(), + b: "403".into(), + }] + ); + } + + #[test] + fn an_error_code_difference_is_reported() { + let a = vec![Observation::new("GetObject", "negative", 404, Some("NoSuchKey".into()), [])]; + let b = vec![Observation::new( + "GetObject", + "negative", + 404, + Some("NoSuchBucket".into()), + [], + )]; + let differences = compare(&a, &b); + assert_eq!(fields(&differences), ["error_code"]); + assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("NoSuchKey", "NoSuchBucket")); + } + + #[test] + fn an_error_code_on_one_side_only_is_reported() { + let a = vec![Observation::new("HeadBucket", "negative", 404, Some("NotFound".into()), [])]; + let b = vec![Observation::new("HeadBucket", "negative", 404, None, [])]; + let differences = compare(&a, &b); + assert_eq!(fields(&differences), ["error_code"]); + assert_eq!(differences[0].b, ABSENT); + } + + #[test] + fn an_allowlisted_header_difference_is_reported() { + let a = vec![Observation::new("HeadObject", "normal", 200, None, [("ETag", "\"aa\"")])]; + let b = vec![Observation::new("HeadObject", "normal", 200, None, [("etag", "\"bb\"")])]; + assert_eq!(fields(&compare(&a, &b)), ["header.etag"]); + } + + #[test] + fn a_header_sent_by_one_side_only_is_reported() { + let a = vec![Observation::new( + "DeleteObject", + "normal", + 204, + None, + [("x-amz-delete-marker", "true")], + )]; + let b = vec![Observation::new("DeleteObject", "normal", 204, None, [])]; + let differences = compare(&a, &b); + assert_eq!(fields(&differences), ["header.x-amz-delete-marker"]); + assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("true", ABSENT)); + } + + #[test] + fn every_server_side_encryption_header_is_compared() { + let header = "x-amz-server-side-encryption-customer-key-md5"; + let a = vec![Observation::new("PutObject", "sse-c", 200, None, [(header, "one")])]; + let b = vec![Observation::new("PutObject", "sse-c", 200, None, [(header, "two")])]; + assert_eq!(fields(&compare(&a, &b)), [format!("header.{header}")]); + } + + #[test] + fn headers_outside_the_allowlist_are_not_compared() { + let a = vec![Observation::new( + "ListBuckets", + "normal", + 200, + None, + [ + ("date", "Mon"), + ("x-amz-request-id", "1"), + ("server", "a"), + ("content-length", "10"), + ], + )]; + let b = vec![Observation::new( + "ListBuckets", + "normal", + 200, + None, + [ + ("date", "Tue"), + ("x-amz-request-id", "2"), + ("server", "b"), + ("content-length", "11"), + ], + )]; + assert_eq!(compare(&a, &b), Vec::new()); + assert!(a[0].headers.is_empty(), "non-allowlisted headers were kept: {:?}", a[0].headers); + } + + #[test] + fn distinct_version_ids_are_not_a_difference() { + let a = vec![Observation::new( + "PutObject", + "normal", + 200, + None, + [("x-amz-version-id", "0b1c")], + )]; + let b = vec![Observation::new( + "PutObject", + "normal", + 200, + None, + [("x-amz-version-id", "9f8e")], + )]; + assert_eq!(compare(&a, &b), Vec::new()); + } + + #[test] + fn a_null_version_id_against_a_real_one_is_reported() { + let a = vec![Observation::new( + "PutObject", + "normal", + 200, + None, + [("x-amz-version-id", "null")], + )]; + let b = vec![Observation::new( + "PutObject", + "normal", + 200, + None, + [("x-amz-version-id", "9f8e")], + )]; + assert_eq!(fields(&compare(&a, &b)), ["header.x-amz-version-id"]); + } + + #[test] + fn a_body_field_difference_is_reported() { + let a = vec![ok("GetBucketLocation").with_field("location_constraint", "us-east-1")]; + let b = vec![ok("GetBucketLocation").with_field("location_constraint", "eu-west-1")]; + let differences = compare(&a, &b); + assert_eq!(fields(&differences), ["body.location_constraint"]); + assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("us-east-1", "eu-west-1")); + } + + #[test] + fn a_body_field_on_one_side_only_is_reported() { + let a = vec![ok("ListObjectsV2").with_field("keys", "[]")]; + let b = vec![ok("ListObjectsV2")]; + assert_eq!(fields(&compare(&a, &b)), ["body.keys"]); + } + + #[test] + fn error_messages_are_never_compared() { + let mut a = Observation::new("GetObject", "negative", 404, Some("NoSuchKey".into()), []); + let mut b = a.clone(); + a.message = Some("The specified key does not exist.".into()); + b.message = Some("Object not found".into()); + assert_eq!(compare(&[a], &[b]), Vec::new()); + } + + #[test] + fn a_case_observed_on_one_side_only_is_reported() { + let a = vec![ok("HeadBucket"), ok("ListBuckets")]; + let b = vec![ok("HeadBucket")]; + let differences = compare(&a, &b); + assert_eq!(fields(&differences), ["observation"]); + assert_eq!(differences[0].op, "ListBuckets"); + assert_eq!(compared_ops(&a, &b).into_iter().collect::>(), ["HeadBucket"]); + } + + #[test] + fn a_case_recorded_twice_is_reported() { + let a = vec![ok("HeadBucket"), ok("HeadBucket")]; + let b = vec![ok("HeadBucket")]; + assert_eq!(fields(&compare(&a, &b)), ["duplicate_case"]); + } + + #[test] + fn compared_ops_counts_operations_not_cases() { + let negative = Observation::new("HeadBucket", "negative", 404, Some("NotFound".into()), []); + let a = vec![ok("HeadBucket"), negative, ok("ListBuckets")]; + let b = a.clone(); + assert_eq!(compared_ops(&a, &b).len(), 2); + } + + #[test] + fn the_report_lists_each_difference_as_op_field_a_b() { + let directory = std::env::temp_dir().join(format!("dual-stack-report-{}", uuid::Uuid::new_v4())); + let report = Report { + schema: 1, + run: "unit".into(), + stack_a: "legacy".into(), + stack_b: "legacy".into(), + region_a: None, + region_b: Some("eu-west-1".into()), + compared_ops: 1, + compared_cases: 1, + differences: compare( + &[ok("GetBucketLocation").with_field("location_constraint", "us-east-1")], + &[ok("GetBucketLocation").with_field("location_constraint", "eu-west-1")], + ), + }; + let path = report.write_to(&directory).expect("write the report"); + let json: serde_json::Value = serde_json::from_slice(&std::fs::read(&path).expect("read the report")).expect("parse"); + let _ = std::fs::remove_dir_all(&directory); + assert_eq!(path.file_name().and_then(|name| name.to_str()), Some("unit.json")); + assert_eq!(json["compared_ops"], 1); + assert_eq!( + json["differences"], + serde_json::json!([{ + "op": "GetBucketLocation", + "case": "normal", + "field": "body.location_constraint", + "a": "us-east-1", + "b": "eu-west-1", + }]) + ); + } +} diff --git a/crates/e2e_test/src/dual_stack_differential/mod.rs b/crates/e2e_test/src/dual_stack_differential/mod.rs new file mode 100644 index 000000000..1de05e122 --- /dev/null +++ b/crates/e2e_test/src/dual_stack_differential/mod.rs @@ -0,0 +1,197 @@ +// Copyright 2024 RustFS Team +// +// 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 +// +// http://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. + +//! Dual-stack differential: one `rustfs` binary, two processes, one SDK script (rustfs/backlog#2740). +//! +//! Stack A always runs `RUSTFS_S3_STACK=legacy`. Stack B runs the value of `RUSTFS_E2E_STACK_B` +//! (default `legacy` until the gateway stack is assembled), and `RUSTFS_E2E_STACK_B_REGION`, when +//! set, starts B in another region so that a difference can be injected by hand. Both servers +//! answer the script in `scenarios`, which covers every operation of `impl S3 for FS`; `compare` +//! reports each `(op, field, a, b)` on which the answers differ, and every run writes its report +//! to `/dual-stack/.json`. Two processes, not two embedded servers: bucket metadata +//! is process-global, so two stacks in one process would read each other's buckets. + +mod compare; +mod scenarios; + +use crate::common::{RustFSTestEnvironment, init_logging, workspace_root}; +use compare::{Difference, Observation, Report, compare, compared_cases, compared_ops, render}; +use std::path::PathBuf; +use std::time::{SystemTime, UNIX_EPOCH}; + +type BoxError = Box; + +const STACK_A: &str = "legacy"; +const ENV_STACK_B: &str = "RUSTFS_E2E_STACK_B"; +const ENV_STACK_B_REGION: &str = "RUSTFS_E2E_STACK_B_REGION"; +/// Becomes `gateway` once rustfs/backlog#2734 T2.5 assembles the gateway stack. +const DEFAULT_STACK_B: &str = "legacy"; +/// The region the negative control gives stack A explicitly, so an inherited `RUSTFS_REGION` +/// cannot erase the injected difference. +const CONTROL_REGION_A: &str = "us-east-1"; +const CONTROL_REGION_B: &str = "eu-west-1"; +/// `normal` cases the legacy stack must answer with a 2xx. The comparison alone cannot tell a +/// script that exercises real behavior from one in which both stacks fail the same way. Five of +/// the 73 fail on legacy by design of this script: PutBucketReplication (no registered remote +/// target), GetBucketReplication (so no configuration), PutBucketNotificationConfiguration (no +/// notification target configured), RestoreObject (the object is not archived) and +/// GetObjectTorrent (legacy answers `NoSuchKey`). +const MIN_NORMAL_SUCCESSES: usize = 68; + +struct PairConfig { + stack_b: String, + region_a: Option, + region_b: Option, +} + +impl PairConfig { + fn from_env() -> Self { + Self { + stack_b: std::env::var(ENV_STACK_B).unwrap_or_else(|_| DEFAULT_STACK_B.to_owned()), + region_a: None, + region_b: std::env::var(ENV_STACK_B_REGION).ok(), + } + } +} + +/// Starts one server on `stack`, overriding its region only when asked. +async fn start_stack(stack: &str, region: Option<&str>) -> Result { + let mut env = RustFSTestEnvironment::new().await?; + let mut vars = vec![("RUSTFS_S3_STACK", stack)]; + if let Some(region) = region { + vars.push(("RUSTFS_REGION", region)); + } + env.start_rustfs_server_without_cleanup_with_env(&vars).await?; + Ok(env) +} + +/// Starts stack A and stack B side by side, each with its own port and data directory. +async fn spawn_pair(config: &PairConfig) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), BoxError> { + tokio::try_join!( + start_stack(STACK_A, config.region_a.as_deref()), + start_stack(&config.stack_b, config.region_b.as_deref()) + ) +} + +fn report_directory() -> PathBuf { + let target = std::env::var_os("CARGO_TARGET_DIR").map_or_else(|| PathBuf::from("target"), PathBuf::from); + let target = if target.is_absolute() { + target + } else { + workspace_root().join(target) + }; + target.join("dual-stack") +} + +struct Run { + report: Report, + path: PathBuf, + a: Vec, +} + +/// Runs the script against both stacks concurrently and writes the report. +async fn run_differential(label: &str, config: &PairConfig) -> Run { + let (server_a, server_b) = spawn_pair(config).await.expect("start stack A and stack B"); + let (a, b) = tokio::join!(Box::pin(scenarios::run(&server_a)), Box::pin(scenarios::run(&server_b))); + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_millis()); + let report = Report { + schema: 1, + run: format!("{label}-{millis}-{}", std::process::id()), + stack_a: STACK_A.to_owned(), + stack_b: config.stack_b.clone(), + region_a: config.region_a.clone(), + region_b: config.region_b.clone(), + compared_ops: compared_ops(&a, &b).len(), + compared_cases: compared_cases(&a, &b), + differences: compare(&a, &b), + }; + let path = report.write_to(&report_directory()).expect("write the dual-stack report"); + Run { report, path, a } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn both_stacks_answer_every_operation_identically() { + init_logging(); + let config = PairConfig::from_env(); + let run = run_differential("self-comparison", &config).await; + let report = &run.report; + + let mut observed: Vec<_> = run.a.iter().map(|observation| observation.op).collect(); + observed.sort_unstable(); + observed.dedup(); + assert_eq!( + observed, + scenarios::OPERATIONS, + "the script must drive exactly the operations of impl S3 for FS" + ); + assert_eq!(report.compared_ops, scenarios::OPERATIONS.len(), "report: {}", run.path.display()); + assert!( + report.differences.is_empty(), + "stack {} and stack {} differ on {} of {} cases (report: {}):\n{}", + report.stack_a, + report.stack_b, + report.differences.len(), + report.compared_cases, + run.path.display(), + render(&report.differences) + ); + + let not_4xx: Vec<_> = run + .a + .iter() + .filter(|observation| observation.case == "negative" && !(400..500).contains(&observation.status)) + .collect(); + assert!(not_4xx.is_empty(), "negative cases must fail with a 4xx on {STACK_A}: {not_4xx:#?}"); + let successes = run + .a + .iter() + .filter(|observation| observation.case == "normal" && observation.is_success()) + .count(); + assert!( + successes >= MIN_NORMAL_SUCCESSES, + "only {successes} normal cases succeeded on {STACK_A}: {:#?}", + run.a + ); + } + + #[tokio::test] + async fn differential_detects_an_injected_difference() { + init_logging(); + let config = PairConfig { + stack_b: STACK_A.to_owned(), + region_a: Some(CONTROL_REGION_A.to_owned()), + region_b: Some(CONTROL_REGION_B.to_owned()), + }; + let run = run_differential("injected-region", &config).await; + let expected = Difference { + op: "GetBucketLocation".to_owned(), + case: "normal".to_owned(), + field: "body.location_constraint".to_owned(), + a: CONTROL_REGION_A.to_owned(), + b: CONTROL_REGION_B.to_owned(), + }; + assert!( + run.report.differences.contains(&expected), + "B started in {CONTROL_REGION_B} but GetBucketLocation was not reported (report: {}):\n{}", + run.path.display(), + render(&run.report.differences) + ); + } +} diff --git a/crates/e2e_test/src/dual_stack_differential/scenarios.rs b/crates/e2e_test/src/dual_stack_differential/scenarios.rs new file mode 100644 index 000000000..24fdc9aab --- /dev/null +++ b/crates/e2e_test/src/dual_stack_differential/scenarios.rs @@ -0,0 +1,1103 @@ +// Copyright 2024 RustFS Team +// +// 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 +// +// http://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. + +//! The SDK script both stacks answer, and the operation list it must cover. +//! +//! Every operation of `impl S3 for FS` (`rustfs/src/storage/ecfs.rs`) gets a `normal` request +//! and, where the operation has one, a `negative` request that must fail with a 4xx. Setup +//! requests go through the same recorder under their own case label, so nothing the script +//! sends is left out of the comparison. Values minted per process (upload IDs, timestamps) are +//! never recorded, and the comparator reduces a version ID to whether it is `null`. + +use super::compare::Observation; +use crate::common::{RustFSTestEnvironment, build_test_s3_config}; +use aws_sdk_s3::Client; +use aws_sdk_s3::config::interceptors::BeforeDeserializationInterceptorContextRef; +use aws_sdk_s3::config::retry::RetryConfig; +use aws_sdk_s3::config::{ConfigBag, Intercept, RuntimeComponents}; +use aws_sdk_s3::error::{BoxError, ProvideErrorMetadata, SdkError}; +use aws_sdk_s3::primitives::{ByteStream, DateTime}; +use aws_sdk_s3::types::{ + AccelerateConfiguration, BucketAccelerateStatus, BucketCannedAcl, BucketLifecycleConfiguration, BucketLoggingStatus, + BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, CorsConfiguration, CorsRule, CsvInput, CsvOutput, + DefaultRetention, Delete, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExpirationStatus, + ExpressionType, FileHeaderInfo, IndexDocument, InputSerialization, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter, + NotificationConfiguration, ObjectAttributes, ObjectCannedAcl, ObjectIdentifier, ObjectLockConfiguration, ObjectLockEnabled, + ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode, ObjectLockRule, + OutputSerialization, Payer, PublicAccessBlockConfiguration, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, + ReplicationRuleStatus, RequestPaymentConfiguration, RestoreRequest, SelectObjectContentEventStream, ServerSideEncryption, + ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Tagging, + VersioningConfiguration, WebsiteConfiguration, +}; +use md5::{Digest as _, Md5}; +use std::future::Future; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +const BUCKET: &str = "dual-stack-main"; +const LOCK_BUCKET: &str = "dual-stack-lock"; +const SCRATCH_BUCKET: &str = "dual-stack-scratch"; +const MISSING_BUCKET: &str = "dual-stack-missing"; +/// Uppercase and `_` are not allowed in a bucket name. +const INVALID_BUCKET: &str = "Dual_Stack_Invalid"; +const README: &str = "docs/readme.txt"; +const COPY: &str = "docs/copy.txt"; +const CSV: &str = "data.csv"; +const LOCKED: &str = "locked.txt"; +const SSE_C_KEY_NAME: &str = "sse-c.txt"; +const ABSENT: &str = "absent.txt"; +const MPU: &str = "mpu/one.bin"; +const MPU_COPY: &str = "mpu/copy.bin"; +const UNKNOWN_UPLOAD: &str = "dual-stack-unknown-upload"; +/// 2099-01-01T00:00:00Z: a retention date both stacks receive verbatim. +const RETAIN_UNTIL_SECS: i64 = 4_070_908_800; +const SELECT_DEADLINE: Duration = Duration::from_secs(30); + +type Fields = Vec<(&'static str, String)>; +type Captured = (u16, Vec<(String, String)>); + +/// Every method of `impl S3 for FS` in `rustfs/src/storage/ecfs.rs`, by S3 operation name, sorted. +/// Derived with +/// `sed -n '/^impl S3 for FS {/,/^}/p' rustfs/src/storage/ecfs.rs | grep -oE '^ async fn [a-z_0-9]+'`; +/// this crate does not read another crate's source, so a new method there must be added here +/// and to the script. +pub(crate) const OPERATIONS: [&str; 73] = [ + "AbortMultipartUpload", + "CompleteMultipartUpload", + "CopyObject", + "CreateBucket", + "CreateMultipartUpload", + "DeleteBucket", + "DeleteBucketCors", + "DeleteBucketEncryption", + "DeleteBucketLifecycle", + "DeleteBucketPolicy", + "DeleteBucketReplication", + "DeleteBucketTagging", + "DeleteBucketWebsite", + "DeleteObject", + "DeleteObjectTagging", + "DeleteObjects", + "DeletePublicAccessBlock", + "GetBucketAccelerateConfiguration", + "GetBucketAcl", + "GetBucketCors", + "GetBucketEncryption", + "GetBucketLifecycleConfiguration", + "GetBucketLocation", + "GetBucketLogging", + "GetBucketNotificationConfiguration", + "GetBucketPolicy", + "GetBucketPolicyStatus", + "GetBucketReplication", + "GetBucketRequestPayment", + "GetBucketTagging", + "GetBucketVersioning", + "GetBucketWebsite", + "GetObject", + "GetObjectAcl", + "GetObjectAttributes", + "GetObjectLegalHold", + "GetObjectLockConfiguration", + "GetObjectRetention", + "GetObjectTagging", + "GetObjectTorrent", + "GetPublicAccessBlock", + "HeadBucket", + "HeadObject", + "ListBuckets", + "ListMultipartUploads", + "ListObjectVersions", + "ListObjects", + "ListObjectsV2", + "ListParts", + "PutBucketAccelerateConfiguration", + "PutBucketAcl", + "PutBucketCors", + "PutBucketEncryption", + "PutBucketLifecycleConfiguration", + "PutBucketLogging", + "PutBucketNotificationConfiguration", + "PutBucketPolicy", + "PutBucketReplication", + "PutBucketRequestPayment", + "PutBucketTagging", + "PutBucketVersioning", + "PutBucketWebsite", + "PutObject", + "PutObjectAcl", + "PutObjectLegalHold", + "PutObjectLockConfiguration", + "PutObjectRetention", + "PutObjectTagging", + "PutPublicAccessBlock", + "RestoreObject", + "SelectObjectContent", + "UploadPart", + "UploadPartCopy", +]; + +/// Records the status and headers of the one response each request receives (retries are off). +#[derive(Clone, Debug, Default)] +struct ResponseCapture(Arc>>); + +impl ResponseCapture { + fn take(&self) -> Option { + self.0.lock().ok().and_then(|mut slot| slot.take()) + } +} + +impl Intercept for ResponseCapture { + fn name(&self) -> &'static str { + "dual-stack-response-capture" + } + + fn read_before_deserialization( + &self, + context: &BeforeDeserializationInterceptorContextRef<'_>, + _runtime_components: &RuntimeComponents, + _cfg: &mut ConfigBag, + ) -> Result<(), BoxError> { + let response = context.response(); + let headers = response + .headers() + .iter() + .map(|(name, value)| (name.to_owned(), value.to_owned())) + .collect(); + *self + .0 + .lock() + .map_err(|_| std::io::Error::other("response capture mutex was poisoned"))? = + Some((response.status().as_u16(), headers)); + Ok(()) + } +} + +fn error_code(error: &SdkError) -> String { + match error { + SdkError::ServiceError(service) => service.err().code().unwrap_or("").to_owned(), + SdkError::ConstructionFailure(_) => "".to_owned(), + SdkError::TimeoutError(_) => "".to_owned(), + SdkError::DispatchFailure(_) => "".to_owned(), + SdkError::ResponseError(_) => "".to_owned(), + _ => "".to_owned(), + } +} + +fn none(_: &T) -> Fields { + Vec::new() +} + +fn md5_hex(bytes: &[u8]) -> String { + hex_simd::encode_to_string(Md5::digest(bytes), hex_simd::AsciiCase::Lower) +} + +struct Recorder { + capture: ResponseCapture, + observations: Vec, +} + +impl Recorder { + /// Sends one request and records what came back. The request future is lazy, so nothing is + /// on the wire before the stale capture is cleared. + async fn step( + &mut self, + op: &'static str, + case: &'static str, + request: impl Future>>, + fields: impl FnOnce(&T) -> Fields, + ) -> Option + where + E: ProvideErrorMetadata, + { + let _ = self.capture.take(); + let result = request.await; + let (status, headers) = self.capture.take().unwrap_or_default(); + let headers = headers.iter().map(|(name, value)| (name.as_str(), value.as_str())); + match result { + Ok(output) => { + let observation = fields(&output) + .into_iter() + .fold(Observation::new(op, case, status, None, headers), |observation, (name, value)| { + observation.with_field(name, value) + }); + self.observations.push(observation); + Some(output) + } + Err(error) => { + let mut observation = Observation::new(op, case, status, Some(error_code(&error)), headers); + observation.message = error.as_service_error().and_then(|e| e.message()).map(str::to_owned); + self.observations.push(observation); + None + } + } + } + + /// Adds a field decoded after the response headers arrived (a streamed body). + fn amend(&mut self, name: &str, value: String) { + if let Some(last) = self.observations.last_mut() { + last.fields.insert(name.to_owned(), value); + } + } +} + +fn client(env: &RustFSTestEnvironment, capture: &ResponseCapture, access_key: &str, secret_key: &str) -> Client { + let config = build_test_s3_config(&env.url, access_key, secret_key, None, "dual-stack-differential") + .to_builder() + // One request, one response: a retry would hide the first answer. + .retry_config(RetryConfig::disabled()) + .interceptor(capture.clone()) + .build(); + Client::from_conf(config) +} + +struct SseC { + key: String, + key_md5: String, +} + +fn sse_c() -> SseC { + let raw = [7u8; 32]; + SseC { + key: base64_simd::STANDARD.encode_to_string(raw), + key_md5: base64_simd::STANDARD.encode_to_string(Md5::digest(raw)), + } +} + +fn tagging(key: &str, value: &str) -> Tagging { + Tagging::builder() + .tag_set(Tag::builder().key(key).value(value).build().expect("tag")) + .build() + .expect("tagging") +} + +fn tags(set: &[Tag]) -> String { + format!("{:?}", set.iter().map(|tag| (tag.key(), tag.value())).collect::>()) +} + +fn grants(grants: &[aws_sdk_s3::types::Grant]) -> String { + let mut permissions: Vec = grants.iter().map(|grant| format!("{:?}", grant.permission())).collect(); + permissions.sort(); + format!("{permissions:?}") +} + +/// Runs the whole script against one server and returns its transcript. +pub(crate) async fn run(env: &RustFSTestEnvironment) -> Vec { + let capture = ResponseCapture::default(); + let c = client(env, &capture, &env.access_key, &env.secret_key); + let stranger = client(env, &capture, "dual-stack-stranger", "dual-stack-stranger-secret"); + let mut s = Recorder { + capture, + observations: Vec::new(), + }; + // Each section holds dozens of SDK futures; boxing keeps the test future off the thread + // stack, which a debug build otherwise overflows. + Box::pin(buckets(&mut s, &c, &stranger)).await; + Box::pin(objects(&mut s, &c)).await; + Box::pin(bucket_subresources(&mut s, &c)).await; + Box::pin(bucket_configuration(&mut s, &c)).await; + Box::pin(object_lock(&mut s, &c)).await; + Box::pin(multipart(&mut s, &c)).await; + Box::pin(object_extras(&mut s, &c)).await; + Box::pin(deletes_and_listings(&mut s, &c)).await; + s.observations +} + +async fn buckets(s: &mut Recorder, c: &Client, stranger: &Client) { + s.step("CreateBucket", "normal", c.create_bucket().bucket(BUCKET).send(), none) + .await; + // Re-creating an owned bucket answers 200 in us-east-1, so it is not the negative case. + s.step("CreateBucket", "owned", c.create_bucket().bucket(BUCKET).send(), none) + .await; + s.step("CreateBucket", "negative", c.create_bucket().bucket(INVALID_BUCKET).send(), none) + .await; + let lock = c.create_bucket().bucket(LOCK_BUCKET).object_lock_enabled_for_bucket(true); + s.step("CreateBucket", "object-lock", lock.send(), none).await; + s.step("CreateBucket", "scratch", c.create_bucket().bucket(SCRATCH_BUCKET).send(), none) + .await; + s.step("HeadBucket", "normal", c.head_bucket().bucket(BUCKET).send(), none) + .await; + s.step("HeadBucket", "negative", c.head_bucket().bucket(MISSING_BUCKET).send(), none) + .await; + s.step("ListBuckets", "normal", c.list_buckets().send(), |o| { + let mut names: Vec<_> = o.buckets().iter().filter_map(|bucket| bucket.name()).collect(); + names.sort_unstable(); + vec![("buckets", format!("{names:?}"))] + }) + .await; + s.step("ListBuckets", "negative", stranger.list_buckets().send(), none).await; + s.step("GetBucketLocation", "normal", c.get_bucket_location().bucket(BUCKET).send(), |o| { + let location = o + .location_constraint() + .map(|constraint| constraint.as_str()) + .unwrap_or_default(); + vec![("location_constraint", location.to_owned())] + }) + .await; + let missing = c.get_bucket_location().bucket(MISSING_BUCKET); + s.step("GetBucketLocation", "negative", missing.send(), none).await; + let enabled = VersioningConfiguration::builder() + .status(BucketVersioningStatus::Enabled) + .build(); + let put = c + .put_bucket_versioning() + .bucket(BUCKET) + .versioning_configuration(enabled.clone()); + s.step("PutBucketVersioning", "normal", put.send(), none).await; + let put = c + .put_bucket_versioning() + .bucket(MISSING_BUCKET) + .versioning_configuration(enabled); + s.step("PutBucketVersioning", "negative", put.send(), none).await; + s.step("GetBucketVersioning", "normal", c.get_bucket_versioning().bucket(BUCKET).send(), |o| { + vec![("status", format!("{:?}", o.status()))] + }) + .await; + let missing = c.get_bucket_versioning().bucket(MISSING_BUCKET); + s.step("GetBucketVersioning", "negative", missing.send(), none).await; +} + +async fn objects(s: &mut Recorder, c: &Client) { + let put = |bucket: &str, key: &str, body: &'static [u8]| { + c.put_object() + .bucket(bucket) + .key(key) + .content_type("text/plain") + .body(ByteStream::from_static(body)) + }; + s.step("PutObject", "normal", put(BUCKET, README, b"hello from both stacks\n").send(), none) + .await; + s.step("PutObject", "negative", put(MISSING_BUCKET, README, b"x").send(), none) + .await; + s.step("PutObject", "seed-a", put(BUCKET, "docs/a.txt", b"a").send(), none) + .await; + s.step("PutObject", "seed-b", put(BUCKET, "logs/b.txt", b"b").send(), none) + .await; + let csv = put(BUCKET, CSV, b"name,size\nalpha,1\nbeta,2\n").content_type("text/csv"); + s.step("PutObject", "seed-csv", csv.send(), none).await; + s.step("PutObject", "seed-locked", put(LOCK_BUCKET, LOCKED, b"locked").send(), none) + .await; + let key = sse_c(); + let encrypted = put(BUCKET, SSE_C_KEY_NAME, b"customer key") + .sse_customer_algorithm("AES256") + .sse_customer_key(&key.key) + .sse_customer_key_md5(&key.key_md5); + s.step("PutObject", "sse-c", encrypted.send(), none).await; + + s.step("HeadObject", "normal", c.head_object().bucket(BUCKET).key(README).send(), none) + .await; + s.step("HeadObject", "negative", c.head_object().bucket(BUCKET).key(ABSENT).send(), none) + .await; + let head = c + .head_object() + .bucket(BUCKET) + .key(SSE_C_KEY_NAME) + .sse_customer_algorithm("AES256") + .sse_customer_key(&key.key) + .sse_customer_key_md5(&key.key_md5); + s.step("HeadObject", "sse-c", head.send(), none).await; + + if let Some(output) = s + .step("GetObject", "normal", c.get_object().bucket(BUCKET).key(README).send(), none) + .await + { + let body = match output.body.collect().await { + Ok(body) => md5_hex(&body.into_bytes()), + Err(_) => "".to_owned(), + }; + s.amend("body_md5", body); + } + let unsatisfiable = c.get_object().bucket(BUCKET).key(README).range("bytes=1000-2000"); + s.step("GetObject", "negative", unsatisfiable.send(), none).await; + + let attributes = |key: &str| { + c.get_object_attributes() + .bucket(BUCKET) + .key(key) + .object_attributes(ObjectAttributes::Etag) + .object_attributes(ObjectAttributes::ObjectSize) + }; + s.step("GetObjectAttributes", "normal", attributes(README).send(), |o| { + vec![ + ("etag", format!("{:?}", o.e_tag())), + ("size", format!("{:?}", o.object_size())), + ] + }) + .await; + s.step("GetObjectAttributes", "negative", attributes(ABSENT).send(), none) + .await; + + let copy = |source: &str| { + c.copy_object() + .bucket(BUCKET) + .key(COPY) + .copy_source(format!("{BUCKET}/{source}")) + }; + s.step("CopyObject", "normal", copy(README).send(), |o| { + vec![("etag", format!("{:?}", o.copy_object_result().and_then(|r| r.e_tag())))] + }) + .await; + s.step("CopyObject", "negative", copy(ABSENT).send(), none).await; + + let put_tags = |key: &str| c.put_object_tagging().bucket(BUCKET).key(key).tagging(tagging("k", "v")); + s.step("PutObjectTagging", "normal", put_tags(README).send(), none).await; + s.step("PutObjectTagging", "negative", put_tags(ABSENT).send(), none).await; + let get_tags = |key: &str| c.get_object_tagging().bucket(BUCKET).key(key); + s.step("GetObjectTagging", "normal", get_tags(README).send(), |o| { + vec![("tags", tags(o.tag_set()))] + }) + .await; + s.step("GetObjectTagging", "negative", get_tags(ABSENT).send(), none).await; + let delete_tags = |key: &str| c.delete_object_tagging().bucket(BUCKET).key(key); + s.step("DeleteObjectTagging", "normal", delete_tags(README).send(), none) + .await; + s.step("DeleteObjectTagging", "negative", delete_tags(ABSENT).send(), none) + .await; + + let put_acl = |key: &str| c.put_object_acl().bucket(BUCKET).key(key).acl(ObjectCannedAcl::Private); + s.step("PutObjectAcl", "normal", put_acl(README).send(), none).await; + s.step("PutObjectAcl", "negative", put_acl(ABSENT).send(), none).await; + let get_acl = |key: &str| c.get_object_acl().bucket(BUCKET).key(key); + s.step("GetObjectAcl", "normal", get_acl(README).send(), |o| vec![("grants", grants(o.grants()))]) + .await; + s.step("GetObjectAcl", "negative", get_acl(ABSENT).send(), none).await; +} + +async fn bucket_subresources(s: &mut Recorder, c: &Client) { + let put_acl = |bucket: &str| c.put_bucket_acl().bucket(bucket).acl(BucketCannedAcl::Private); + s.step("PutBucketAcl", "normal", put_acl(BUCKET).send(), none).await; + s.step("PutBucketAcl", "negative", put_acl(MISSING_BUCKET).send(), none).await; + let get_acl = |bucket: &str| c.get_bucket_acl().bucket(bucket); + s.step("GetBucketAcl", "normal", get_acl(BUCKET).send(), |o| vec![("grants", grants(o.grants()))]) + .await; + s.step("GetBucketAcl", "negative", get_acl(MISSING_BUCKET).send(), none).await; + + let put_tags = |bucket: &str| c.put_bucket_tagging().bucket(bucket).tagging(tagging("team", "storage")); + s.step("PutBucketTagging", "normal", put_tags(BUCKET).send(), none).await; + s.step("PutBucketTagging", "negative", put_tags(MISSING_BUCKET).send(), none) + .await; + let get_tags = || c.get_bucket_tagging().bucket(BUCKET); + s.step("GetBucketTagging", "normal", get_tags().send(), |o| vec![("tags", tags(o.tag_set()))]) + .await; + let delete_tags = |bucket: &str| c.delete_bucket_tagging().bucket(bucket); + s.step("DeleteBucketTagging", "normal", delete_tags(BUCKET).send(), none) + .await; + s.step("GetBucketTagging", "negative", get_tags().send(), none).await; + s.step("DeleteBucketTagging", "negative", delete_tags(MISSING_BUCKET).send(), none) + .await; + + let cors = CorsConfiguration::builder() + .cors_rules( + CorsRule::builder() + .allowed_methods("GET") + .allowed_origins("https://example.com") + .build() + .expect("cors rule"), + ) + .build() + .expect("cors configuration"); + let put_cors = |bucket: &str| c.put_bucket_cors().bucket(bucket).cors_configuration(cors.clone()); + s.step("PutBucketCors", "normal", put_cors(BUCKET).send(), none).await; + s.step("PutBucketCors", "negative", put_cors(MISSING_BUCKET).send(), none) + .await; + let get_cors = || c.get_bucket_cors().bucket(BUCKET); + s.step("GetBucketCors", "normal", get_cors().send(), |o| { + let rules: Vec<_> = o + .cors_rules() + .iter() + .map(|rule| (rule.allowed_methods(), rule.allowed_origins())) + .collect(); + vec![("rules", format!("{rules:?}"))] + }) + .await; + let delete_cors = |bucket: &str| c.delete_bucket_cors().bucket(bucket); + s.step("DeleteBucketCors", "normal", delete_cors(BUCKET).send(), none).await; + s.step("GetBucketCors", "negative", get_cors().send(), none).await; + s.step("DeleteBucketCors", "negative", delete_cors(MISSING_BUCKET).send(), none) + .await; + + let policy = format!( + r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Principal":{{"AWS":["*"]}},"Action":["s3:GetObject"],"Resource":["arn:aws:s3:::{BUCKET}/public/*"]}}]}}"# + ); + s.step( + "PutBucketPolicy", + "normal", + c.put_bucket_policy().bucket(BUCKET).policy(policy).send(), + none, + ) + .await; + let malformed = c.put_bucket_policy().bucket(BUCKET).policy("{"); + s.step("PutBucketPolicy", "negative", malformed.send(), none).await; + let get_policy = || c.get_bucket_policy().bucket(BUCKET); + s.step("GetBucketPolicy", "normal", get_policy().send(), none).await; + let status = |bucket: &str| c.get_bucket_policy_status().bucket(bucket); + s.step("GetBucketPolicyStatus", "normal", status(BUCKET).send(), |o| { + vec![("is_public", format!("{:?}", o.policy_status().and_then(|p| p.is_public())))] + }) + .await; + s.step("GetBucketPolicyStatus", "negative", status(MISSING_BUCKET).send(), none) + .await; + let delete_policy = |bucket: &str| c.delete_bucket_policy().bucket(bucket); + s.step("DeleteBucketPolicy", "normal", delete_policy(BUCKET).send(), none) + .await; + s.step("GetBucketPolicy", "negative", get_policy().send(), none).await; + s.step("DeleteBucketPolicy", "negative", delete_policy(MISSING_BUCKET).send(), none) + .await; +} + +async fn bucket_configuration(s: &mut Recorder, c: &Client) { + let encryption = ServerSideEncryptionConfiguration::builder() + .rules( + ServerSideEncryptionRule::builder() + .apply_server_side_encryption_by_default( + ServerSideEncryptionByDefault::builder() + .sse_algorithm(ServerSideEncryption::Aes256) + .build() + .expect("default encryption"), + ) + .build(), + ) + .build() + .expect("encryption configuration"); + let put = |bucket: &str| { + c.put_bucket_encryption() + .bucket(bucket) + .server_side_encryption_configuration(encryption.clone()) + }; + s.step("PutBucketEncryption", "normal", put(SCRATCH_BUCKET).send(), none) + .await; + s.step("PutBucketEncryption", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = || c.get_bucket_encryption().bucket(SCRATCH_BUCKET); + s.step("GetBucketEncryption", "normal", get().send(), |o| { + let algorithms: Vec<_> = o + .server_side_encryption_configuration() + .map(|config| config.rules()) + .unwrap_or_default() + .iter() + .filter_map(|rule| rule.apply_server_side_encryption_by_default()) + .map(|default| default.sse_algorithm().as_str().to_owned()) + .collect(); + vec![("algorithms", format!("{algorithms:?}"))] + }) + .await; + let delete = |bucket: &str| c.delete_bucket_encryption().bucket(bucket); + s.step("DeleteBucketEncryption", "normal", delete(SCRATCH_BUCKET).send(), none) + .await; + s.step("GetBucketEncryption", "negative", get().send(), none).await; + s.step("DeleteBucketEncryption", "negative", delete(MISSING_BUCKET).send(), none) + .await; + + let lifecycle = BucketLifecycleConfiguration::builder() + .rules( + LifecycleRule::builder() + .id("expire-tmp") + .filter(LifecycleRuleFilter::builder().prefix("tmp/").build()) + .status(ExpirationStatus::Enabled) + .expiration(LifecycleExpiration::builder().days(30).build()) + .build() + .expect("lifecycle rule"), + ) + .build() + .expect("lifecycle configuration"); + let put = |bucket: &str| { + c.put_bucket_lifecycle_configuration() + .bucket(bucket) + .lifecycle_configuration(lifecycle.clone()) + }; + s.step("PutBucketLifecycleConfiguration", "normal", put(BUCKET).send(), none) + .await; + s.step("PutBucketLifecycleConfiguration", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = || c.get_bucket_lifecycle_configuration().bucket(BUCKET); + s.step("GetBucketLifecycleConfiguration", "normal", get().send(), |o| { + let rules: Vec<_> = o.rules().iter().map(|rule| (rule.id(), rule.status().as_str())).collect(); + vec![("rules", format!("{rules:?}"))] + }) + .await; + let delete = |bucket: &str| c.delete_bucket_lifecycle().bucket(bucket); + s.step("DeleteBucketLifecycle", "normal", delete(BUCKET).send(), none).await; + s.step("GetBucketLifecycleConfiguration", "negative", get().send(), none) + .await; + s.step("DeleteBucketLifecycle", "negative", delete(MISSING_BUCKET).send(), none) + .await; + + let replication = ReplicationConfiguration::builder() + .role("") + .rules( + ReplicationRule::builder() + .id("to-remote") + .priority(1) + .status(ReplicationRuleStatus::Enabled) + .filter(ReplicationRuleFilter::builder().prefix("").build()) + .delete_marker_replication( + DeleteMarkerReplication::builder() + .status(DeleteMarkerReplicationStatus::Disabled) + .build(), + ) + .destination( + Destination::builder() + .bucket("arn:aws:s3:::dual-stack-remote") + .build() + .expect("replication destination"), + ) + .build() + .expect("replication rule"), + ) + .build() + .expect("replication configuration"); + let put = |bucket: &str| { + c.put_bucket_replication() + .bucket(bucket) + .replication_configuration(replication.clone()) + }; + s.step("PutBucketReplication", "normal", put(BUCKET).send(), none).await; + s.step("PutBucketReplication", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = |bucket: &str| c.get_bucket_replication().bucket(bucket); + s.step("GetBucketReplication", "normal", get(BUCKET).send(), none).await; + s.step("GetBucketReplication", "negative", get(MISSING_BUCKET).send(), none) + .await; + let delete = |bucket: &str| c.delete_bucket_replication().bucket(bucket); + s.step("DeleteBucketReplication", "normal", delete(BUCKET).send(), none).await; + s.step("DeleteBucketReplication", "negative", delete(MISSING_BUCKET).send(), none) + .await; + + let website = WebsiteConfiguration::builder() + .index_document(IndexDocument::builder().suffix("index.html").build().expect("index document")) + .build(); + let put = |bucket: &str| c.put_bucket_website().bucket(bucket).website_configuration(website.clone()); + s.step("PutBucketWebsite", "normal", put(BUCKET).send(), none).await; + s.step("PutBucketWebsite", "negative", put(MISSING_BUCKET).send(), none).await; + let get = || c.get_bucket_website().bucket(BUCKET); + s.step("GetBucketWebsite", "normal", get().send(), |o| { + vec![("index", format!("{:?}", o.index_document().map(|index| index.suffix())))] + }) + .await; + let delete = |bucket: &str| c.delete_bucket_website().bucket(bucket); + s.step("DeleteBucketWebsite", "normal", delete(BUCKET).send(), none).await; + s.step("GetBucketWebsite", "negative", get().send(), none).await; + s.step("DeleteBucketWebsite", "negative", delete(MISSING_BUCKET).send(), none) + .await; + + let block = PublicAccessBlockConfiguration::builder() + .block_public_acls(true) + .ignore_public_acls(true) + .block_public_policy(true) + .restrict_public_buckets(true) + .build(); + let put = |bucket: &str| { + c.put_public_access_block() + .bucket(bucket) + .public_access_block_configuration(block.clone()) + }; + s.step("PutPublicAccessBlock", "normal", put(SCRATCH_BUCKET).send(), none) + .await; + s.step("PutPublicAccessBlock", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = |bucket: &str| c.get_public_access_block().bucket(bucket); + s.step("GetPublicAccessBlock", "normal", get(SCRATCH_BUCKET).send(), |o| { + vec![("configuration", format!("{:?}", o.public_access_block_configuration()))] + }) + .await; + let delete = |bucket: &str| c.delete_public_access_block().bucket(bucket); + s.step("DeletePublicAccessBlock", "normal", delete(SCRATCH_BUCKET).send(), none) + .await; + // Legacy answers a missing configuration with a 5xx, so that request is not the negative case. + s.step("GetPublicAccessBlock", "unconfigured", get(SCRATCH_BUCKET).send(), none) + .await; + s.step("GetPublicAccessBlock", "negative", get(MISSING_BUCKET).send(), none) + .await; + s.step("DeletePublicAccessBlock", "negative", delete(MISSING_BUCKET).send(), none) + .await; + + let accelerate = AccelerateConfiguration::builder() + .status(BucketAccelerateStatus::Suspended) + .build(); + let put = |bucket: &str| { + c.put_bucket_accelerate_configuration() + .bucket(bucket) + .accelerate_configuration(accelerate.clone()) + }; + s.step("PutBucketAccelerateConfiguration", "normal", put(BUCKET).send(), none) + .await; + s.step("PutBucketAccelerateConfiguration", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = |bucket: &str| c.get_bucket_accelerate_configuration().bucket(bucket); + s.step("GetBucketAccelerateConfiguration", "normal", get(BUCKET).send(), |o| { + vec![("status", format!("{:?}", o.status()))] + }) + .await; + s.step("GetBucketAccelerateConfiguration", "negative", get(MISSING_BUCKET).send(), none) + .await; + + let payment = RequestPaymentConfiguration::builder() + .payer(Payer::BucketOwner) + .build() + .expect("request payment configuration"); + let put = |bucket: &str| { + c.put_bucket_request_payment() + .bucket(bucket) + .request_payment_configuration(payment.clone()) + }; + s.step("PutBucketRequestPayment", "normal", put(BUCKET).send(), none).await; + s.step("PutBucketRequestPayment", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = |bucket: &str| c.get_bucket_request_payment().bucket(bucket); + s.step("GetBucketRequestPayment", "normal", get(BUCKET).send(), |o| { + vec![("payer", format!("{:?}", o.payer()))] + }) + .await; + s.step("GetBucketRequestPayment", "negative", get(MISSING_BUCKET).send(), none) + .await; + + let put = |bucket: &str| { + c.put_bucket_logging() + .bucket(bucket) + .bucket_logging_status(BucketLoggingStatus::builder().build()) + }; + s.step("PutBucketLogging", "normal", put(BUCKET).send(), none).await; + s.step("PutBucketLogging", "negative", put(MISSING_BUCKET).send(), none).await; + let get = |bucket: &str| c.get_bucket_logging().bucket(bucket); + s.step("GetBucketLogging", "normal", get(BUCKET).send(), |o| { + vec![("enabled", o.logging_enabled().is_some().to_string())] + }) + .await; + s.step("GetBucketLogging", "negative", get(MISSING_BUCKET).send(), none).await; + + let put = |bucket: &str| { + c.put_bucket_notification_configuration() + .bucket(bucket) + .notification_configuration(NotificationConfiguration::builder().build()) + }; + s.step("PutBucketNotificationConfiguration", "normal", put(BUCKET).send(), none) + .await; + s.step("PutBucketNotificationConfiguration", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = |bucket: &str| c.get_bucket_notification_configuration().bucket(bucket); + s.step("GetBucketNotificationConfiguration", "normal", get(BUCKET).send(), |o| { + let counts = ( + o.topic_configurations().len(), + o.queue_configurations().len(), + o.lambda_function_configurations().len(), + ); + vec![("configurations", format!("{counts:?}"))] + }) + .await; + s.step("GetBucketNotificationConfiguration", "negative", get(MISSING_BUCKET).send(), none) + .await; +} + +async fn object_lock(s: &mut Recorder, c: &Client) { + let lock = ObjectLockConfiguration::builder() + .object_lock_enabled(ObjectLockEnabled::Enabled) + .rule( + ObjectLockRule::builder() + .default_retention( + DefaultRetention::builder() + .mode(ObjectLockRetentionMode::Governance) + .days(1) + .build(), + ) + .build(), + ) + .build(); + let put = |bucket: &str| { + c.put_object_lock_configuration() + .bucket(bucket) + .object_lock_configuration(lock.clone()) + }; + s.step("PutObjectLockConfiguration", "normal", put(LOCK_BUCKET).send(), none) + .await; + s.step("PutObjectLockConfiguration", "negative", put(MISSING_BUCKET).send(), none) + .await; + let get = |bucket: &str| c.get_object_lock_configuration().bucket(bucket); + s.step("GetObjectLockConfiguration", "normal", get(LOCK_BUCKET).send(), |o| { + let config = o.object_lock_configuration(); + vec![ + ("enabled", format!("{:?}", config.and_then(|config| config.object_lock_enabled()))), + ( + "default_retention", + format!( + "{:?}", + config + .and_then(|config| config.rule()) + .and_then(|rule| rule.default_retention()) + ), + ), + ] + }) + .await; + s.step("GetObjectLockConfiguration", "negative", get(BUCKET).send(), none) + .await; + + let retention = ObjectLockRetention::builder() + .mode(ObjectLockRetentionMode::Governance) + .retain_until_date(DateTime::from_secs(RETAIN_UNTIL_SECS)) + .build(); + let put = |key: &str| { + c.put_object_retention() + .bucket(LOCK_BUCKET) + .key(key) + .retention(retention.clone()) + }; + s.step("PutObjectRetention", "normal", put(LOCKED).send(), none).await; + s.step("PutObjectRetention", "negative", put(ABSENT).send(), none).await; + let get = |bucket: &str, key: &str| c.get_object_retention().bucket(bucket).key(key); + s.step("GetObjectRetention", "normal", get(LOCK_BUCKET, LOCKED).send(), |o| { + let retention = o.retention(); + vec![ + ("mode", format!("{:?}", retention.and_then(|r| r.mode()))), + ("retain_until", format!("{:?}", retention.and_then(|r| r.retain_until_date()))), + ] + }) + .await; + s.step("GetObjectRetention", "negative", get(BUCKET, README).send(), none) + .await; + + // Legacy answers a legal hold on a missing key with a 5xx, so that request is not the + // negative case; a bucket without object lock is. + let hold = ObjectLockLegalHold::builder().status(ObjectLockLegalHoldStatus::On).build(); + let put = |bucket: &str, key: &str| c.put_object_legal_hold().bucket(bucket).key(key).legal_hold(hold.clone()); + s.step("PutObjectLegalHold", "normal", put(LOCK_BUCKET, LOCKED).send(), none) + .await; + s.step("PutObjectLegalHold", "absent-key", put(LOCK_BUCKET, ABSENT).send(), none) + .await; + s.step("PutObjectLegalHold", "negative", put(BUCKET, README).send(), none) + .await; + let get = |bucket: &str, key: &str| c.get_object_legal_hold().bucket(bucket).key(key); + s.step("GetObjectLegalHold", "normal", get(LOCK_BUCKET, LOCKED).send(), |o| { + vec![("status", format!("{:?}", o.legal_hold().and_then(|hold| hold.status())))] + }) + .await; + s.step("GetObjectLegalHold", "absent-key", get(LOCK_BUCKET, ABSENT).send(), none) + .await; + s.step("GetObjectLegalHold", "negative", get(BUCKET, README).send(), none) + .await; +} + +async fn multipart(s: &mut Recorder, c: &Client) { + let create = |bucket: &str, key: &str| c.create_multipart_upload().bucket(bucket).key(key); + let upload = s + .step("CreateMultipartUpload", "normal", create(BUCKET, MPU).send(), |o| { + vec![("key", format!("{:?}", o.key()))] + }) + .await + .and_then(|output| output.upload_id().map(str::to_owned)) + .unwrap_or_else(|| UNKNOWN_UPLOAD.to_owned()); + s.step("CreateMultipartUpload", "negative", create(MISSING_BUCKET, MPU).send(), none) + .await; + let part = |upload_id: &str| { + c.upload_part() + .bucket(BUCKET) + .key(MPU) + .upload_id(upload_id) + .part_number(1) + .body(ByteStream::from_static(b"the only part")) + }; + let etag = s + .step("UploadPart", "normal", part(&upload).send(), none) + .await + .and_then(|output| output.e_tag().map(str::to_owned)) + .unwrap_or_default(); + s.step("UploadPart", "negative", part(UNKNOWN_UPLOAD).send(), none).await; + let list = |upload_id: &str| c.list_parts().bucket(BUCKET).key(MPU).upload_id(upload_id); + s.step("ListParts", "normal", list(&upload).send(), |o| { + let parts: Vec<_> = o + .parts() + .iter() + .map(|part| (part.part_number(), part.size(), part.e_tag())) + .collect(); + vec![("parts", format!("{parts:?}"))] + }) + .await; + s.step("ListParts", "negative", list(UNKNOWN_UPLOAD).send(), none).await; + let completed = CompletedMultipartUpload::builder() + .parts(CompletedPart::builder().part_number(1).e_tag(etag).build()) + .build(); + let complete = |upload_id: &str| { + c.complete_multipart_upload() + .bucket(BUCKET) + .key(MPU) + .upload_id(upload_id) + .multipart_upload(completed.clone()) + }; + s.step("CompleteMultipartUpload", "normal", complete(&upload).send(), |o| { + vec![("etag", format!("{:?}", o.e_tag()))] + }) + .await; + s.step("CompleteMultipartUpload", "negative", complete(UNKNOWN_UPLOAD).send(), none) + .await; + + let copy_upload = s + .step("CreateMultipartUpload", "copy-target", create(BUCKET, MPU_COPY).send(), none) + .await + .and_then(|output| output.upload_id().map(str::to_owned)) + .unwrap_or_else(|| UNKNOWN_UPLOAD.to_owned()); + let copy = |source: &str| { + c.upload_part_copy() + .bucket(BUCKET) + .key(MPU_COPY) + .upload_id(©_upload) + .part_number(1) + .copy_source(format!("{BUCKET}/{source}")) + }; + s.step("UploadPartCopy", "normal", copy(README).send(), |o| { + vec![("etag", format!("{:?}", o.copy_part_result().and_then(|r| r.e_tag())))] + }) + .await; + s.step("UploadPartCopy", "negative", copy(ABSENT).send(), none).await; + let uploads = |bucket: &str| c.list_multipart_uploads().bucket(bucket); + s.step("ListMultipartUploads", "normal", uploads(BUCKET).send(), |o| { + let keys: Vec<_> = o.uploads().iter().map(|upload| upload.key()).collect(); + vec![("keys", format!("{keys:?}"))] + }) + .await; + s.step("ListMultipartUploads", "negative", uploads(MISSING_BUCKET).send(), none) + .await; + let abort = || { + c.abort_multipart_upload() + .bucket(BUCKET) + .key(MPU_COPY) + .upload_id(©_upload) + }; + s.step("AbortMultipartUpload", "normal", abort().send(), none).await; + s.step("AbortMultipartUpload", "negative", abort().send(), none).await; +} + +async fn object_extras(s: &mut Recorder, c: &Client) { + let restore = |key: &str| { + c.restore_object() + .bucket(BUCKET) + .key(key) + .restore_request(RestoreRequest::builder().days(1).build()) + }; + s.step("RestoreObject", "normal", restore(README).send(), none).await; + s.step("RestoreObject", "negative", restore(ABSENT).send(), none).await; + + let select = |key: &str| { + c.select_object_content() + .bucket(BUCKET) + .key(key) + .expression("SELECT s.name FROM S3Object s") + .expression_type(ExpressionType::Sql) + .input_serialization( + InputSerialization::builder() + .csv(CsvInput::builder().file_header_info(FileHeaderInfo::Use).build()) + .build(), + ) + .output_serialization(OutputSerialization::builder().csv(CsvOutput::builder().build()).build()) + }; + if let Some(mut output) = s.step("SelectObjectContent", "normal", select(CSV).send(), none).await { + let mut records = Vec::new(); + let drained = tokio::time::timeout(SELECT_DEADLINE, async { + loop { + match output.payload.recv().await { + Ok(Some(SelectObjectContentEventStream::Records(event))) => { + records.extend_from_slice(event.payload().map(|blob| blob.as_ref()).unwrap_or_default()); + } + Ok(Some(_)) => {} + Ok(None) => return "end".to_owned(), + Err(error) => return format!("error:{:?}", error.as_service_error().and_then(|e| e.code())), + } + } + }) + .await + .unwrap_or_else(|_| "".to_owned()); + s.amend("stream", drained); + s.amend("records_md5", md5_hex(&records)); + } + s.step("SelectObjectContent", "negative", select(ABSENT).send(), none).await; + + let torrent = |key: &str| c.get_object_torrent().bucket(BUCKET).key(key); + s.step("GetObjectTorrent", "normal", torrent(README).send(), none).await; + s.step("GetObjectTorrent", "negative", torrent(ABSENT).send(), none).await; +} + +async fn deletes_and_listings(s: &mut Recorder, c: &Client) { + s.step("DeleteObject", "normal", c.delete_object().bucket(BUCKET).key(COPY).send(), none) + .await; + let missing = c.delete_object().bucket(MISSING_BUCKET).key(COPY); + s.step("DeleteObject", "negative", missing.send(), none).await; + let delete = Delete::builder() + .objects( + ObjectIdentifier::builder() + .key("docs/a.txt") + .build() + .expect("object identifier"), + ) + .objects(ObjectIdentifier::builder().key(ABSENT).build().expect("object identifier")) + .build() + .expect("delete request"); + let delete_objects = |bucket: &str| c.delete_objects().bucket(bucket).delete(delete.clone()); + s.step("DeleteObjects", "normal", delete_objects(BUCKET).send(), |o| { + let mut deleted: Vec<_> = o.deleted().iter().map(|d| (d.key(), d.delete_marker())).collect(); + deleted.sort_unstable(); + let errors: Vec<_> = o.errors().iter().map(|e| (e.key(), e.code())).collect(); + vec![("deleted", format!("{deleted:?}")), ("errors", format!("{errors:?}"))] + }) + .await; + s.step("DeleteObjects", "negative", delete_objects(MISSING_BUCKET).send(), none) + .await; + + let list = |bucket: &str| c.list_objects().bucket(bucket).prefix("docs/"); + s.step("ListObjects", "normal", list(BUCKET).send(), |o| { + let keys: Vec<_> = o.contents().iter().map(|object| (object.key(), object.size())).collect(); + vec![("keys", format!("{keys:?}"))] + }) + .await; + s.step("ListObjects", "negative", list(MISSING_BUCKET).send(), none).await; + let list = |bucket: &str| c.list_objects_v2().bucket(bucket).delimiter("/"); + s.step("ListObjectsV2", "normal", list(BUCKET).send(), |o| { + let keys: Vec<_> = o.contents().iter().map(|object| object.key()).collect(); + let prefixes: Vec<_> = o.common_prefixes().iter().map(|prefix| prefix.prefix()).collect(); + vec![ + ("keys", format!("{keys:?}")), + ("prefixes", format!("{prefixes:?}")), + ("key_count", format!("{:?}", o.key_count())), + ] + }) + .await; + s.step("ListObjectsV2", "negative", list(MISSING_BUCKET).send(), none).await; + let versions = |bucket: &str| c.list_object_versions().bucket(bucket); + s.step("ListObjectVersions", "normal", versions(BUCKET).send(), |o| { + let versions: Vec<_> = o.versions().iter().map(|v| (v.key(), v.is_latest(), v.size())).collect(); + let markers: Vec<_> = o.delete_markers().iter().map(|m| (m.key(), m.is_latest())).collect(); + vec![ + ("versions", format!("{versions:?}")), + ("delete_markers", format!("{markers:?}")), + ] + }) + .await; + s.step("ListObjectVersions", "negative", versions(MISSING_BUCKET).send(), none) + .await; + + s.step("DeleteBucket", "normal", c.delete_bucket().bucket(SCRATCH_BUCKET).send(), none) + .await; + s.step("DeleteBucket", "negative", c.delete_bucket().bucket(BUCKET).send(), none) + .await; +} + +#[cfg(test)] +mod tests { + use super::OPERATIONS; + + #[test] + fn the_operation_list_is_sorted_without_duplicates() { + assert!( + OPERATIONS.windows(2).all(|pair| pair[0] < pair[1]), + "OPERATIONS must be strictly sorted: {OPERATIONS:?}" + ); + } +} diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index b4e0f4ce9..0d2a058f1 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -31,6 +31,12 @@ pub mod chaos; #[cfg(test)] pub mod fake_s3_target; +// Dual-stack differential (rustfs/backlog#2740): one binary, two processes on +// different RUSTFS_S3_STACK values, the same SDK script for every S3 operation, +// compared on status, error code, allowlisted headers and typed body fields. +#[cfg(test)] +mod dual_stack_differential; + // On-demand migration (backlog#2147): shared two-server environment, admin // wrappers, and the harness self-test (backlog#2151). Behavior scenarios are // added by later ODM tasks.