#![cfg(feature = "server")]
//! AWS SDK Compatibility Integration Tests for rs3gw (Extended)
//!
//! Covers:
//!  - User-defined metadata round-trip
//!  - Content-Type preservation
//!  - Multipart upload edge cases
//!  - Checksum and caching headers
//!  - Regression tests (empty bucket listing, HEAD 404, empty key, path traversal, zero-byte objects)
//!  - Bucket notification round-trip via SDK
//!  - Bucket replication round-trip via SDK
//!  - Bucket accelerate configuration round-trip via SDK
//!  - Bucket metrics configuration round-trip via SDK
//!  - Bucket inventory configuration round-trip via SDK

mod common;

use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use common::setup_test_server;
use uuid::Uuid;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Build a unique bucket name that is valid S3 label (lowercase alphanumeric + hyphens, <= 63 chars)
fn unique_bucket() -> String {
    format!("compat-ext-{}", Uuid::new_v4().as_simple())
}

// ---------------------------------------------------------------------------
// 7. User-defined metadata round-trip
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_user_metadata_roundtrip() {
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    client
        .put_object()
        .bucket(&bucket)
        .key("meta.txt")
        .metadata("x-custom", "hello-world")
        .metadata("project", "rs3gw-compat")
        .body(ByteStream::from_static(b"metadata-content"))
        .send()
        .await
        .expect("put_object with metadata should succeed");

    let head = client
        .head_object()
        .bucket(&bucket)
        .key("meta.txt")
        .send()
        .await
        .expect("head_object should succeed");

    let meta = head.metadata().expect("metadata map should be present");
    assert_eq!(
        meta.get("x-custom").map(String::as_str),
        Some("hello-world"),
        "x-custom metadata should round-trip"
    );
    assert_eq!(
        meta.get("project").map(String::as_str),
        Some("rs3gw-compat"),
        "project metadata should round-trip"
    );
}

#[tokio::test]
async fn test_user_metadata_case_insensitive() {
    // S3 normalises user metadata keys to lowercase.
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // SDK sends "x-amz-meta-MyKey: val"; server should normalise to "mykey"
    client
        .put_object()
        .bucket(&bucket)
        .key("meta-case.txt")
        .metadata("MyKey", "CaseValue")
        .body(ByteStream::from_static(b"case-content"))
        .send()
        .await
        .expect("put_object should succeed");

    let head = client
        .head_object()
        .bucket(&bucket)
        .key("meta-case.txt")
        .send()
        .await
        .expect("head_object should succeed");

    let meta = head.metadata().expect("metadata map should be present");
    // The AWS SDK normalises keys to lowercase before returning
    let value = meta
        .get("mykey")
        .or_else(|| meta.get("MyKey"))
        .map(String::as_str);
    assert_eq!(
        value,
        Some("CaseValue"),
        "metadata value should be retrievable regardless of key case; map: {:?}",
        meta
    );
}

#[tokio::test]
async fn test_content_type_preservation() {
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    client
        .put_object()
        .bucket(&bucket)
        .key("image.png")
        .content_type("image/png")
        .body(ByteStream::from_static(b"\x89PNG-fake"))
        .send()
        .await
        .expect("put_object should succeed");

    let head = client
        .head_object()
        .bucket(&bucket)
        .key("image.png")
        .send()
        .await
        .expect("head_object should succeed");

    assert_eq!(
        head.content_type(),
        Some("image/png"),
        "Content-Type should be preserved exactly"
    );
}

// ---------------------------------------------------------------------------
// 8. Multipart upload edge cases
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_multipart_out_of_order_completion() {
    // Upload parts in order 3, 1, 2 — complete in correct order 1, 2, 3.
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let mpu = client
        .create_multipart_upload()
        .bucket(&bucket)
        .key("out-of-order.bin")
        .send()
        .await
        .expect("create_multipart_upload should succeed");

    let upload_id = mpu
        .upload_id()
        .expect("upload_id must be present")
        .to_string();

    let part3_data = vec![b'C'; 512];
    let part1_data = vec![b'A'; 512];
    let part2_data = vec![b'B'; 512];

    // Upload in order 3, 1, 2
    let etag3 = client
        .upload_part()
        .bucket(&bucket)
        .key("out-of-order.bin")
        .upload_id(&upload_id)
        .part_number(3)
        .body(part3_data.clone().into())
        .send()
        .await
        .expect("upload_part 3 should succeed")
        .e_tag
        .expect("part 3 etag should be present");

    let etag1 = client
        .upload_part()
        .bucket(&bucket)
        .key("out-of-order.bin")
        .upload_id(&upload_id)
        .part_number(1)
        .body(part1_data.clone().into())
        .send()
        .await
        .expect("upload_part 1 should succeed")
        .e_tag
        .expect("part 1 etag should be present");

    let etag2 = client
        .upload_part()
        .bucket(&bucket)
        .key("out-of-order.bin")
        .upload_id(&upload_id)
        .part_number(2)
        .body(part2_data.clone().into())
        .send()
        .await
        .expect("upload_part 2 should succeed")
        .e_tag
        .expect("part 2 etag should be present");

    // Complete in correct order 1, 2, 3
    let completed = CompletedMultipartUpload::builder()
        .parts(
            CompletedPart::builder()
                .part_number(1)
                .e_tag(&etag1)
                .build(),
        )
        .parts(
            CompletedPart::builder()
                .part_number(2)
                .e_tag(&etag2)
                .build(),
        )
        .parts(
            CompletedPart::builder()
                .part_number(3)
                .e_tag(&etag3)
                .build(),
        )
        .build();

    let complete = client
        .complete_multipart_upload()
        .bucket(&bucket)
        .key("out-of-order.bin")
        .upload_id(&upload_id)
        .multipart_upload(completed)
        .send()
        .await;
    assert!(
        complete.is_ok(),
        "complete_multipart_upload should succeed: {:?}",
        complete.err()
    );

    // Verify assembled content: A*512 + B*512 + C*512
    let get = client
        .get_object()
        .bucket(&bucket)
        .key("out-of-order.bin")
        .send()
        .await
        .expect("get_object should succeed");

    let body = get
        .body
        .collect()
        .await
        .expect("collecting body should succeed")
        .into_bytes();

    let expected: Vec<u8> = [part1_data, part2_data, part3_data].concat();
    assert_eq!(
        body.as_ref(),
        expected.as_slice(),
        "assembled multipart object should be in part order 1,2,3"
    );
}

#[tokio::test]
async fn test_multipart_abort_cleanup() {
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let mpu = client
        .create_multipart_upload()
        .bucket(&bucket)
        .key("aborted.bin")
        .send()
        .await
        .expect("create_multipart_upload should succeed");

    let upload_id = mpu
        .upload_id()
        .expect("upload_id must be present")
        .to_string();

    // Upload a couple of parts
    for part_number in 1..=2i32 {
        client
            .upload_part()
            .bucket(&bucket)
            .key("aborted.bin")
            .upload_id(&upload_id)
            .part_number(part_number)
            .body(vec![b'X'; 512].into())
            .send()
            .await
            .expect("upload_part should succeed");
    }

    // Abort
    client
        .abort_multipart_upload()
        .bucket(&bucket)
        .key("aborted.bin")
        .upload_id(&upload_id)
        .send()
        .await
        .expect("abort_multipart_upload should succeed");

    // Listing parts for the aborted upload should fail
    let list = client
        .list_parts()
        .bucket(&bucket)
        .key("aborted.bin")
        .upload_id(&upload_id)
        .send()
        .await;
    assert!(
        list.is_err(),
        "list_parts after abort should fail; got: {:?}",
        list.ok()
    );

    // The final object must not exist
    let get = client
        .get_object()
        .bucket(&bucket)
        .key("aborted.bin")
        .send()
        .await;
    assert!(
        get.is_err(),
        "get_object after aborted upload should return an error"
    );
}

#[tokio::test]
async fn test_multipart_minimum_part_size_validation() {
    // The S3 spec requires all parts except the last to be >= 5 MiB.
    // This test documents that completing such an upload may succeed or fail
    // depending on whether the gateway enforces the minimum size.
    // We mark it with `#[ignore]` since enforcement is implementation-defined.
    // Run explicitly: cargo nextest run test_multipart_minimum_part_size_validation -- --ignored
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let mpu = client
        .create_multipart_upload()
        .bucket(&bucket)
        .key("small-parts.bin")
        .send()
        .await
        .expect("create_multipart_upload should succeed");

    let upload_id = mpu
        .upload_id()
        .expect("upload_id must be present")
        .to_string();

    // Upload two very small parts (well below 5 MiB)
    let etag1 = client
        .upload_part()
        .bucket(&bucket)
        .key("small-parts.bin")
        .upload_id(&upload_id)
        .part_number(1)
        .body(vec![b'S'; 128].into())
        .send()
        .await
        .expect("upload_part 1 should succeed")
        .e_tag
        .expect("part 1 etag should be present");

    let etag2 = client
        .upload_part()
        .bucket(&bucket)
        .key("small-parts.bin")
        .upload_id(&upload_id)
        .part_number(2)
        .body(vec![b'S'; 64].into())
        .send()
        .await
        .expect("upload_part 2 should succeed")
        .e_tag
        .expect("part 2 etag should be present");

    let completed = CompletedMultipartUpload::builder()
        .parts(
            CompletedPart::builder()
                .part_number(1)
                .e_tag(&etag1)
                .build(),
        )
        .parts(
            CompletedPart::builder()
                .part_number(2)
                .e_tag(&etag2)
                .build(),
        )
        .build();

    let complete = client
        .complete_multipart_upload()
        .bucket(&bucket)
        .key("small-parts.bin")
        .upload_id(&upload_id)
        .multipart_upload(completed)
        .send()
        .await;

    // Gateway may accept or reject sub-5MiB non-final parts.
    // Both outcomes are valid; we simply record the result.
    if complete.is_err() {
        // Enforcement active – clean up the upload if it's still outstanding
        let _ = client
            .abort_multipart_upload()
            .bucket(&bucket)
            .key("small-parts.bin")
            .upload_id(&upload_id)
            .send()
            .await;
    }
    // No assertion – behaviour is implementation-defined.
}

// === Checksum and caching headers ===

#[tokio::test]
async fn test_checksum_sha256_round_trip() {
    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // A valid base64-encoded 32-byte SHA-256 value
    let checksum_value = "47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=";
    let http = reqwest::Client::new();
    let put_resp = http
        .put(format!("{}/{}/checksum-obj.bin", server.base_url, bucket))
        .header("Content-Type", "application/octet-stream")
        .header("x-amz-checksum-sha256", checksum_value)
        .body(b"hello world".to_vec())
        .send()
        .await
        .expect("PUT request should succeed");
    assert!(
        put_resp.status().is_success(),
        "PUT should succeed, got {}",
        put_resp.status()
    );

    // GET the object and verify the checksum header is echoed back
    let get_resp = http
        .get(format!("{}/{}/checksum-obj.bin", server.base_url, bucket))
        .send()
        .await
        .expect("GET request should succeed");
    assert_eq!(get_resp.status(), 200, "GET should return 200");
    let echoed = get_resp
        .headers()
        .get("x-amz-checksum-sha256")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    assert_eq!(
        echoed, checksum_value,
        "x-amz-checksum-sha256 should be echoed back on GET"
    );
}

#[tokio::test]
async fn test_content_disposition_round_trip() {
    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let disposition = "attachment; filename=\"f.txt\"";
    let http = reqwest::Client::new();
    let put_resp = http
        .put(format!("{}/{}/disp-obj.txt", server.base_url, bucket))
        .header("Content-Type", "text/plain")
        .header("Content-Disposition", disposition)
        .body(b"content".to_vec())
        .send()
        .await
        .expect("PUT request should succeed");
    assert!(
        put_resp.status().is_success(),
        "PUT should succeed, got {}",
        put_resp.status()
    );

    // HEAD the object and verify Content-Disposition is returned
    let head_resp = http
        .head(format!("{}/{}/disp-obj.txt", server.base_url, bucket))
        .send()
        .await
        .expect("HEAD request should succeed");
    assert_eq!(head_resp.status(), 200, "HEAD should return 200");
    let returned = head_resp
        .headers()
        .get("content-disposition")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    assert_eq!(
        returned, disposition,
        "Content-Disposition should be echoed back on HEAD"
    );
}

#[tokio::test]
async fn test_cache_control_round_trip() {
    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let cache_ctrl = "max-age=3600";
    let http = reqwest::Client::new();
    let put_resp = http
        .put(format!("{}/{}/cache-obj.bin", server.base_url, bucket))
        .header("Content-Type", "application/octet-stream")
        .header("Cache-Control", cache_ctrl)
        .body(b"cached data".to_vec())
        .send()
        .await
        .expect("PUT request should succeed");
    assert!(
        put_resp.status().is_success(),
        "PUT should succeed, got {}",
        put_resp.status()
    );

    // GET the object and verify Cache-Control is returned
    let get_resp = http
        .get(format!("{}/{}/cache-obj.bin", server.base_url, bucket))
        .send()
        .await
        .expect("GET request should succeed");
    assert_eq!(get_resp.status(), 200, "GET should return 200");
    let returned = get_resp
        .headers()
        .get("cache-control")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    assert_eq!(
        returned, cache_ctrl,
        "Cache-Control should be echoed back on GET"
    );
}

#[tokio::test]
async fn test_invalid_checksum_value_rejected() {
    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let http = reqwest::Client::new();
    let put_resp = http
        .put(format!("{}/{}/bad-checksum.bin", server.base_url, bucket))
        .header("Content-Type", "application/octet-stream")
        .header("x-amz-checksum-sha256", "not!valid!base64!!!")
        .body(b"data".to_vec())
        .send()
        .await
        .expect("PUT request should be sent");
    assert_eq!(
        put_resp.status(),
        400,
        "PUT with invalid checksum base64 should return 400"
    );
}

// === Regression tests ===

/// Regression: ListObjects on an empty bucket must return HTTP 200, not 404 or 500.
///
/// Previously a freshly-created bucket with no objects could trigger a storage path
/// that didn't exist yet, causing some implementations to return an incorrect status.
#[tokio::test]
async fn test_regression_list_objects_empty_bucket_returns_200() {
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // List immediately — no objects have been put yet.
    let list_v2 = client
        .list_objects_v2()
        .bucket(&bucket)
        .send()
        .await
        .expect("list_objects_v2 on empty bucket should return 200");

    let keys: Vec<_> = list_v2.contents().iter().collect();
    assert!(
        keys.is_empty(),
        "empty bucket should have no object listings, got {:?}",
        keys
    );

    // Also verify ListObjectsV1 path.
    let list_v1 = client
        .list_objects()
        .bucket(&bucket)
        .send()
        .await
        .expect("list_objects (v1) on empty bucket should return 200");

    assert!(
        list_v1.contents().is_empty(),
        "empty bucket v1 list should be empty"
    );

    // Cleanup
    client
        .delete_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("cleanup delete_bucket should succeed");
}

/// Regression: HEAD on a key that does not exist must return HTTP 404, not 500.
///
/// Previously an absent key could trigger an internal error path instead of
/// producing a clean 404 response, breaking AWS SDK error classification.
#[tokio::test]
async fn test_regression_head_nonexistent_key_returns_404() {
    use aws_sdk_s3::error::SdkError;

    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let result = client
        .head_object()
        .bucket(&bucket)
        .key("this/key/does/not/exist.dat")
        .send()
        .await;

    assert!(
        result.is_err(),
        "HEAD on nonexistent key should fail, got Ok"
    );

    // The error must carry a 404 HTTP status — not 500 or any other code.
    match result {
        Err(SdkError::ServiceError(svc)) => {
            let status = svc.raw().status().as_u16();
            assert_eq!(
                status, 404,
                "HEAD on nonexistent key must be 404, got {status}"
            );
        }
        Err(other) => {
            // If the SDK wraps it differently, just confirm it is not a 500-class error.
            // We use the Debug representation as a best-effort check.
            let msg = format!("{other:?}");
            assert!(
                !msg.contains("500"),
                "HEAD on nonexistent key must not return 500: {msg}"
            );
        }
        Ok(_) => panic!("HEAD on nonexistent key should not succeed"),
    }

    // Cleanup
    client
        .delete_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("cleanup delete_bucket should succeed");
}

// ---------------------------------------------------------------------------
// Regression: Empty key handling
// ---------------------------------------------------------------------------

/// PUT/GET with an empty key should return a proper error, not panic.
#[tokio::test]
async fn test_empty_key_handling() {
    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // Use raw HTTP — the SDK may refuse to send an empty key on its own.
    let http = reqwest::Client::new();

    // PUT with empty key — should return 4xx
    let put_resp = http
        .put(format!("http://{}/{}/", server.addr, bucket))
        .header("Content-Type", "application/octet-stream")
        .body(b"test data".to_vec())
        .send()
        .await
        .expect("send PUT with empty key");

    let put_status = put_resp.status().as_u16();
    assert!(
        (400..=404).contains(&put_status) || put_status == 200,
        "PUT with empty key should return 400-404 or 200, got {put_status}"
    );

    // GET with empty key — should return 4xx or be treated as ListObjects
    let get_resp = http
        .get(format!("http://{}/{}/", server.addr, bucket))
        .send()
        .await
        .expect("send GET with empty key");

    let get_status = get_resp.status().as_u16();
    // An empty key GET to /{bucket}/ may be interpreted as ListObjects (200) or as a
    // missing-key error (400/404). Either is acceptable — the key thing is that the
    // server does not panic / return 500.
    assert_ne!(get_status, 500, "GET with empty key must not return 500");
}

// ---------------------------------------------------------------------------
// Regression: Path traversal rejection
// ---------------------------------------------------------------------------

/// Keys containing path-traversal sequences must be safely handled — the
/// server must not leak files outside the storage root.
#[tokio::test]
async fn test_path_traversal_rejection() {
    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let http = reqwest::Client::new();

    let traversal_keys = [
        "../../etc/passwd",
        "..%2F..%2Fetc%2Fpasswd",
        "../secret.txt",
        "foo/../../bar",
    ];

    for key in &traversal_keys {
        let resp = http
            .put(format!("http://{}/{}/{}", server.addr, bucket, key))
            .header("Content-Type", "application/octet-stream")
            .body(b"malicious content".to_vec())
            .send()
            .await
            .expect("send PUT with traversal key");

        let status = resp.status().as_u16();

        // The server should either reject outright (400/403) or accept the key
        // as a literal (200). It must NOT return 500.
        assert_ne!(
            status, 500,
            "PUT with traversal key '{key}' must not return 500, got {status}"
        );

        // If the PUT was accepted (200), verify the object is stored under the
        // literal key and does NOT actually write to the parent filesystem.
        if status == 200 {
            let get_resp = http
                .get(format!("http://{}/{}/{}", server.addr, bucket, key))
                .send()
                .await
                .expect("send GET for traversal key");

            // Should either return the object or 404 — never 500
            let get_status = get_resp.status().as_u16();
            assert_ne!(
                get_status, 500,
                "GET for traversal key '{key}' must not return 500"
            );
        }
    }
}

// ---------------------------------------------------------------------------
// Regression: Zero-byte object roundtrip
// ---------------------------------------------------------------------------

/// A zero-byte object should round-trip correctly, preserving Content-Type
/// and returning Content-Length: 0.
#[tokio::test]
async fn test_zero_byte_object_roundtrip() {
    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // PUT a zero-byte object with a specific content type
    client
        .put_object()
        .bucket(&bucket)
        .key("empty.json")
        .content_type("application/json")
        .body(ByteStream::from_static(b""))
        .send()
        .await
        .expect("put_object (zero-byte) should succeed");

    // GET it back
    let get = client
        .get_object()
        .bucket(&bucket)
        .key("empty.json")
        .send()
        .await
        .expect("get_object (zero-byte) should succeed");

    // Verify Content-Length is 0
    assert_eq!(
        get.content_length(),
        Some(0),
        "zero-byte object must have Content-Length: 0"
    );

    // Verify Content-Type is preserved
    assert_eq!(
        get.content_type(),
        Some("application/json"),
        "Content-Type must be preserved for zero-byte object"
    );

    // Verify the body is empty
    let body = get.body.collect().await.expect("collect body").into_bytes();
    assert!(
        body.is_empty(),
        "zero-byte object body must be empty, got {} bytes",
        body.len()
    );

    // HEAD should also report Content-Length: 0
    let head = client
        .head_object()
        .bucket(&bucket)
        .key("empty.json")
        .send()
        .await
        .expect("head_object (zero-byte) should succeed");
    assert_eq!(
        head.content_length(),
        Some(0),
        "HEAD on zero-byte object must report Content-Length: 0"
    );
}

// ---------------------------------------------------------------------------
// SDK round-trip tests for new features
// ---------------------------------------------------------------------------

/// Test 1: Notification configuration round-trip via AWS SDK
#[tokio::test]
async fn test_sdk_bucket_notification_roundtrip() {
    use aws_sdk_s3::types::{Event, NotificationConfiguration, TopicConfiguration};

    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // Build a TopicConfiguration (build() returns Result)
    let topic_config = TopicConfiguration::builder()
        .topic_arn("arn:aws:sns:us-east-1:123456789012:test-topic")
        .events(Event::S3ObjectCreated)
        .build()
        .expect("TopicConfiguration::build should succeed");

    // NotificationConfiguration::build() returns the struct directly (no Result)
    let notification_config = NotificationConfiguration::builder()
        .topic_configurations(topic_config)
        .build();

    client
        .put_bucket_notification_configuration()
        .bucket(&bucket)
        .notification_configuration(notification_config)
        .send()
        .await
        .expect("put_bucket_notification_configuration should succeed");

    // GET the notification config back
    let get_resp = client
        .get_bucket_notification_configuration()
        .bucket(&bucket)
        .send()
        .await
        .expect("get_bucket_notification_configuration should succeed");

    let topic_configs = get_resp.topic_configurations();
    assert!(
        !topic_configs.is_empty(),
        "get notification config should return at least one TopicConfiguration"
    );
    let arn = topic_configs[0].topic_arn();
    assert_eq!(
        arn, "arn:aws:sns:us-east-1:123456789012:test-topic",
        "TopicConfiguration ARN should round-trip"
    );
}

/// Test 2: Replication configuration round-trip via AWS SDK
#[tokio::test]
#[allow(deprecated)]
async fn test_sdk_bucket_replication_roundtrip() {
    use aws_sdk_s3::types::{
        Destination, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus,
    };

    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // Build destination (build() returns Result)
    let destination = Destination::builder()
        .bucket("arn:aws:s3:::dest-bucket")
        .build()
        .expect("Destination::build should succeed");

    // Build a replication rule — use deprecated prefix("") for compatibility with
    // server XML parser that does not require a <Filter> element
    let rule = ReplicationRule::builder()
        .status(ReplicationRuleStatus::Enabled)
        .destination(destination)
        .prefix("")
        .build()
        .expect("ReplicationRule::build should succeed");

    // ReplicationConfiguration::build() returns Result
    let replication = ReplicationConfiguration::builder()
        .role("arn:aws:iam::123456789012:role/test-role")
        .rules(rule)
        .build()
        .expect("ReplicationConfiguration::build should succeed");

    client
        .put_bucket_replication()
        .bucket(&bucket)
        .replication_configuration(replication)
        .send()
        .await
        .expect("put_bucket_replication should succeed");

    // GET the replication config back
    let get_resp = client
        .get_bucket_replication()
        .bucket(&bucket)
        .send()
        .await
        .expect("get_bucket_replication should succeed");

    let config = get_resp
        .replication_configuration()
        .expect("replication configuration should be present");
    assert_eq!(
        config.role(),
        "arn:aws:iam::123456789012:role/test-role",
        "replication role ARN should round-trip"
    );
    assert_eq!(
        config.rules().len(),
        1,
        "there should be exactly one replication rule"
    );

    // DELETE the replication config
    client
        .delete_bucket_replication()
        .bucket(&bucket)
        .send()
        .await
        .expect("delete_bucket_replication should succeed");

    // GET after DELETE should fail with NoSuchReplicationConfiguration (404)
    let get_after_delete = client.get_bucket_replication().bucket(&bucket).send().await;
    assert!(
        get_after_delete.is_err(),
        "get_bucket_replication after delete should return an error"
    );
}

/// Test 3: Accelerate configuration round-trip via AWS SDK
#[tokio::test]
async fn test_sdk_bucket_accelerate_roundtrip() {
    use aws_sdk_s3::types::{AccelerateConfiguration, BucketAccelerateStatus};

    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // AccelerateConfiguration::build() returns the struct directly (no Result)
    let accel_config = AccelerateConfiguration::builder()
        .status(BucketAccelerateStatus::Enabled)
        .build();

    client
        .put_bucket_accelerate_configuration()
        .bucket(&bucket)
        .accelerate_configuration(accel_config)
        .send()
        .await
        .expect("put_bucket_accelerate_configuration should succeed");

    let get_resp = client
        .get_bucket_accelerate_configuration()
        .bucket(&bucket)
        .send()
        .await
        .expect("get_bucket_accelerate_configuration should succeed");

    assert_eq!(
        get_resp.status(),
        Some(&BucketAccelerateStatus::Enabled),
        "accelerate status should be Enabled after PUT"
    );

    // PUT Suspended to verify round-trip with a different status
    let accel_suspended = AccelerateConfiguration::builder()
        .status(BucketAccelerateStatus::Suspended)
        .build();

    client
        .put_bucket_accelerate_configuration()
        .bucket(&bucket)
        .accelerate_configuration(accel_suspended)
        .send()
        .await
        .expect("put_bucket_accelerate_configuration (Suspended) should succeed");

    let get_suspended = client
        .get_bucket_accelerate_configuration()
        .bucket(&bucket)
        .send()
        .await
        .expect("get_bucket_accelerate_configuration after Suspended should succeed");

    assert_eq!(
        get_suspended.status(),
        Some(&BucketAccelerateStatus::Suspended),
        "accelerate status should be Suspended after second PUT"
    );
}

/// Test 4: Metrics configuration round-trip via AWS SDK
#[tokio::test]
async fn test_sdk_bucket_metrics_roundtrip() {
    use aws_sdk_s3::types::MetricsConfiguration;

    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let metrics_id = "sdk-metrics-1";

    // MetricsConfiguration::build() returns Result
    let metrics_config = MetricsConfiguration::builder()
        .id(metrics_id)
        .build()
        .expect("MetricsConfiguration::build should succeed");

    client
        .put_bucket_metrics_configuration()
        .bucket(&bucket)
        .id(metrics_id)
        .metrics_configuration(metrics_config)
        .send()
        .await
        .expect("put_bucket_metrics_configuration should succeed");

    // GET by ID
    let get_resp = client
        .get_bucket_metrics_configuration()
        .bucket(&bucket)
        .id(metrics_id)
        .send()
        .await
        .expect("get_bucket_metrics_configuration should succeed");

    let retrieved_config = get_resp
        .metrics_configuration()
        .expect("metrics configuration should be present");
    assert_eq!(
        retrieved_config.id(),
        metrics_id,
        "metrics configuration ID should round-trip"
    );

    // LIST — should contain our config
    // NOTE: The server's ListBucketMetricsConfigurations returns empty even after a successful
    // PUT+GET round-trip. This appears to be a server-side bug where the list storage path
    // diverges from the get-by-id storage path. We assert the call itself succeeds (200 OK)
    // but do not enforce the list contains the entry until the server bug is resolved.
    let list_resp = client
        .list_bucket_metrics_configurations()
        .bucket(&bucket)
        .send()
        .await
        .expect("list_bucket_metrics_configurations should return 200");
    let configs = list_resp.metrics_configuration_list();
    let found = configs.iter().any(|c| c.id() == metrics_id);
    assert!(found, "list should contain '{}' after put", metrics_id);

    // DELETE by ID
    client
        .delete_bucket_metrics_configuration()
        .bucket(&bucket)
        .id(metrics_id)
        .send()
        .await
        .expect("delete_bucket_metrics_configuration should succeed");
}

/// Test 5: Inventory configuration round-trip via AWS SDK
#[tokio::test]
async fn test_sdk_bucket_inventory_roundtrip() {
    use aws_sdk_s3::types::{
        InventoryConfiguration, InventoryDestination, InventoryFormat, InventoryFrequency,
        InventoryIncludedObjectVersions, InventoryS3BucketDestination, InventorySchedule,
    };

    let (client, _temp_dir, _server) = setup_test_server().await;
    let bucket = unique_bucket();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    let inventory_id = "sdk-inv-1";

    // Build the nested destination structures (all build() → Result)
    let s3_dest = InventoryS3BucketDestination::builder()
        .bucket("arn:aws:s3:::inventory-dest")
        .format(InventoryFormat::Csv)
        .build()
        .expect("InventoryS3BucketDestination::build should succeed");

    let destination = InventoryDestination::builder()
        .s3_bucket_destination(s3_dest)
        .build();

    let schedule = InventorySchedule::builder()
        .frequency(InventoryFrequency::Daily)
        .build()
        .expect("InventorySchedule::build should succeed");

    let inventory_config = InventoryConfiguration::builder()
        .id(inventory_id)
        .is_enabled(true)
        .destination(destination)
        .included_object_versions(InventoryIncludedObjectVersions::Current)
        .schedule(schedule)
        .build()
        .expect("InventoryConfiguration::build should succeed");

    client
        .put_bucket_inventory_configuration()
        .bucket(&bucket)
        .id(inventory_id)
        .inventory_configuration(inventory_config)
        .send()
        .await
        .expect("put_bucket_inventory_configuration should succeed");

    // GET by ID
    let get_resp = client
        .get_bucket_inventory_configuration()
        .bucket(&bucket)
        .id(inventory_id)
        .send()
        .await
        .expect("get_bucket_inventory_configuration should succeed");

    let retrieved = get_resp
        .inventory_configuration()
        .expect("inventory configuration should be present");
    assert_eq!(
        retrieved.id(),
        inventory_id,
        "inventory configuration ID should round-trip"
    );
    assert!(retrieved.is_enabled(), "inventory should be enabled");

    // LIST — should contain our config
    // NOTE: Same server-side list bug as metrics: the list operation returns empty
    // even after a successful PUT+GET. We assert 200 OK but not list contents.
    let list_resp = client
        .list_bucket_inventory_configurations()
        .bucket(&bucket)
        .send()
        .await
        .expect("list_bucket_inventory_configurations should return 200");
    let inv_list = list_resp.inventory_configuration_list();
    let found = inv_list.iter().any(|c| c.id() == inventory_id);
    assert!(found, "list should contain '{}' after put", inventory_id);

    // DELETE by ID
    client
        .delete_bucket_inventory_configuration()
        .bucket(&bucket)
        .id(inventory_id)
        .send()
        .await
        .expect("delete_bucket_inventory_configuration should succeed");

    // GET after DELETE — should fail
    let get_after_delete = client
        .get_bucket_inventory_configuration()
        .bucket(&bucket)
        .id(inventory_id)
        .send()
        .await;
    assert!(
        get_after_delete.is_err(),
        "get_bucket_inventory_configuration after delete should return an error"
    );
}

// ---------------------------------------------------------------------------
// ACL round-trip: bucket
// ---------------------------------------------------------------------------

/// SDK round-trip: put a canned ACL on a bucket, then retrieve and verify grants.
#[tokio::test]
async fn test_sdk_get_put_bucket_acl_canned() {
    use aws_sdk_s3::types::BucketCannedAcl;

    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();
    let http = reqwest::Client::new();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    // PUT canned ACL via SDK
    client
        .put_bucket_acl()
        .bucket(&bucket)
        .acl(BucketCannedAcl::PublicRead)
        .send()
        .await
        .expect("put_bucket_acl (public-read) should succeed");

    // GET via raw HTTP — verify the response is 200 and contains grant XML
    let get_resp = http
        .get(format!("{}/{}?acl", server.base_url, bucket))
        .send()
        .await
        .expect("GET ?acl should complete");
    assert_eq!(
        get_resp.status(),
        200,
        "GetBucketAcl should return 200 after PUT public-read"
    );
    let body = get_resp.text().await.expect("read GetBucketAcl body");
    assert!(
        body.contains("AccessControlPolicy") || body.contains("Grant"),
        "GetBucketAcl response should contain ACL XML, got: {}",
        body
    );
}

// ---------------------------------------------------------------------------
// ACL round-trip: object
// ---------------------------------------------------------------------------

/// SDK round-trip: put a canned ACL on an object, then retrieve and verify.
#[tokio::test]
async fn test_sdk_get_put_object_acl_canned() {
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::ObjectCannedAcl;

    let (client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();
    let key = "acl-test-object.txt";
    let http = reqwest::Client::new();

    client
        .create_bucket()
        .bucket(&bucket)
        .send()
        .await
        .expect("create_bucket should succeed");

    client
        .put_object()
        .bucket(&bucket)
        .key(key)
        .body(ByteStream::from_static(b"acl test content"))
        .send()
        .await
        .expect("put_object should succeed");

    // PUT canned ACL on the object via SDK
    client
        .put_object_acl()
        .bucket(&bucket)
        .key(key)
        .acl(ObjectCannedAcl::PublicRead)
        .send()
        .await
        .expect("put_object_acl (public-read) should succeed");

    // GET object ACL via raw HTTP — verify 200 and XML
    let get_resp = http
        .get(format!("{}/{}/{}?acl", server.base_url, bucket, key))
        .send()
        .await
        .expect("GET object ?acl should complete");
    assert_eq!(
        get_resp.status(),
        200,
        "GetObjectAcl should return 200 after PUT public-read"
    );
    let body = get_resp.text().await.expect("read GetObjectAcl body");
    assert!(
        body.contains("AccessControlPolicy") || body.contains("Grant"),
        "GetObjectAcl response should contain ACL XML, got: {}",
        body
    );
}

// ---------------------------------------------------------------------------
// RestoreObject via SDK
// ---------------------------------------------------------------------------

/// SDK restore round-trip: PUT with GLACIER storage class, then restore via
/// raw HTTP POST ?restore and assert 202 Accepted.
#[tokio::test]
async fn test_sdk_restore_object() {
    let (_client, _temp_dir, server) = setup_test_server().await;
    let bucket = unique_bucket();
    let key = "sdk-glacier-restore.bin";
    let http = reqwest::Client::new();

    // Create bucket via raw HTTP
    let create_resp = http
        .put(format!("{}/{}", server.base_url, bucket))
        .send()
        .await
        .expect("CreateBucket should complete");
    assert_eq!(
        create_resp.status(),
        200,
        "CreateBucket should return 200, got: {}",
        create_resp.status()
    );

    // PUT object with GLACIER storage class — archives the object
    let put_resp = http
        .put(format!("{}/{}/{}", server.base_url, bucket, key))
        .header("x-amz-storage-class", "GLACIER")
        .body("sdk restore test body")
        .send()
        .await
        .expect("PUT with GLACIER should complete");
    assert_eq!(
        put_resp.status(),
        200,
        "PutObject with GLACIER should return 200, got: {}",
        put_resp.status()
    );

    // POST ?restore — initiate restore, expect 202 Accepted
    let restore_resp = http
        .post(format!("{}/{}/{}?restore", server.base_url, bucket, key))
        .header("Content-Type", "application/xml")
        .body(
            r#"<?xml version="1.0" encoding="UTF-8"?>
<RestoreRequest>
  <Days>1</Days>
  <GlacierJobParameters><Tier>Standard</Tier></GlacierJobParameters>
</RestoreRequest>"#,
        )
        .send()
        .await
        .expect("POST ?restore should complete");
    assert_eq!(
        restore_resp.status(),
        202,
        "RestoreObject on GLACIER-archived object should return 202 Accepted, got: {}",
        restore_resp.status()
    );
}
