//! Tests for [`super`].

use super::*;

use aws_sdk_s3::config::retry::RetryConfig;
use aws_smithy_http_client::test_util::{ReplayEvent, StaticReplayClient};
use aws_smithy_types::body::SdkBody;

fn test_client() -> S3Client {
    // `from_conf` is local — no network until a request is sent — so this
    // builds a usable client without reaching any endpoint.
    let s3_config = aws_sdk_s3::Config::builder()
        .behavior_version(BehaviorVersion::latest())
        .http_client(https_client())
        .region(Region::new("test"))
        .endpoint_url("http://127.0.0.1:1")
        .credentials_provider(Credentials::new("ak", "sk", None, None, "test"))
        .force_path_style(true)
        .build();
    S3Client {
        client: Client::from_conf(s3_config),
        bucket: "test-bucket".to_string(),
    }
}

/// An `S3Client` whose HTTP layer replays canned responses.
///
/// WHY THIS EXISTS. Two paths in this crate cannot be reached from a live
/// object store, and both were mutation survivors (infra `f7f13914`,
/// `a536db81`). One is defensive code against a response shape S3 does not
/// currently produce; the other writes a bucket setting the test bucket's
/// credentials cannot read back. A canned response reaches the first and a
/// recorded request reaches the second, with no live infrastructure and no
/// bucket-owner permission.
///
/// Retries are off so an exhausted replay list fails immediately: a mutant
/// that makes the crate send one request too many should show up as a
/// failed assertion rather than as seconds of SDK backoff.
fn replay_client(events: Vec<ReplayEvent>) -> (S3Client, StaticReplayClient) {
    let replay = StaticReplayClient::new(events);
    let s3_config = aws_sdk_s3::Config::builder()
        .behavior_version(BehaviorVersion::latest())
        .http_client(replay.clone())
        .retry_config(RetryConfig::disabled())
        .region(Region::new("test"))
        .endpoint_url("http://127.0.0.1:1")
        .credentials_provider(Credentials::new("ak", "sk", None, None, "test"))
        .force_path_style(true)
        .build();
    (
        S3Client {
            client: Client::from_conf(s3_config),
            bucket: "test-bucket".to_string(),
        },
        replay,
    )
}

/// A canned 200 with an XML body, which is what every S3 read returns.
fn xml_ok(body: &str) -> ReplayEvent {
    ReplayEvent::new(
        http::Request::builder()
            .uri("http://test-bucket.localhost/")
            .body(SdkBody::empty())
            .unwrap(),
        http::Response::builder()
            .status(200)
            .header("content-type", "application/xml")
            .body(SdkBody::from(body.to_string()))
            .unwrap(),
    )
}

/// A canned S3 error document, which is how S3 says "no CORS here".
fn xml_err(status: u16, code: &str) -> ReplayEvent {
    let body = format!(
        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
             <Error><Code>{code}</Code><Message>canned</Message>\
             <RequestId>r</RequestId><HostId>h</HostId></Error>"
    );
    ReplayEvent::new(
        http::Request::builder()
            .uri("http://test-bucket.localhost/")
            .body(SdkBody::empty())
            .unwrap(),
        http::Response::builder()
            .status(status)
            .header("content-type", "application/xml")
            .body(SdkBody::from(body))
            .unwrap(),
    )
}

/// infra `f7f13914`. The "truncated but no continuation markers" guard in
/// `list_multipart_uploads_for_key` survived mutation because reaching it
/// needs a response S3 does not produce: truncated, yet naming nothing to
/// continue from. Live MinIO will not build it, and >1000 pending uploads on
/// one key is not a fixture anyone should own. A canned body is.
///
/// The guard is kept rather than deleted because without it that response
/// is an infinite loop against a real endpoint, and the crate is the last
/// thing standing between the orphan reaper and a spin.
///
/// The single-request assertion is the one that matters. Removing the
/// `break` makes the loop ask again with the same (absent) markers, and
/// there is no second canned response, so it also fails the `Ok` — either
/// way the mutant dies.
#[tokio::test]
async fn a_truncated_listing_with_no_markers_stops_instead_of_looping() {
    let (client, replay) = replay_client(vec![xml_ok(
        r#"<?xml version="1.0" encoding="UTF-8"?>
            <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
              <Bucket>test-bucket</Bucket>
              <Prefix>staging/abc</Prefix>
              <MaxUploads>1000</MaxUploads>
              <IsTruncated>true</IsTruncated>
              <Upload>
                <Key>staging/abc</Key>
                <UploadId>upload-one</UploadId>
                <Initiated>2026-08-31T00:00:00.000Z</Initiated>
              </Upload>
              <Upload>
                <Key>staging/abcdef</Key>
                <UploadId>not-ours</UploadId>
                <Initiated>2026-08-31T00:00:00.000Z</Initiated>
              </Upload>
            </ListMultipartUploadsResult>"#,
    )]);

    let ids = client
        .list_multipart_uploads_for_key("staging/abc")
        .await
        .expect("a truncated page with nothing to continue from is a complete answer");

    // The exact-key filter travels with it: `ListMultipartUploads` matches a
    // prefix, so `staging/abcdef` is a live session the reaper must not abort.
    assert_eq!(ids, vec!["upload-one".to_string()]);
    assert_eq!(
        replay.actual_requests().count(),
        1,
        "the guard exists to stop a second identical request"
    );
}

/// The other half of the guard: a truncated page that DOES name a marker is
/// followed. Without this the test above is satisfied by a loop that never
/// iterates at all, which is a different bug wearing the same result.
#[tokio::test]
async fn a_truncated_listing_with_a_marker_asks_for_the_next_page() {
    let (client, replay) = replay_client(vec![
        xml_ok(
            r#"<?xml version="1.0" encoding="UTF-8"?>
                <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
                  <Bucket>test-bucket</Bucket>
                  <IsTruncated>true</IsTruncated>
                  <NextKeyMarker>staging/abc</NextKeyMarker>
                  <NextUploadIdMarker>upload-one</NextUploadIdMarker>
                  <Upload>
                    <Key>staging/abc</Key>
                    <UploadId>upload-one</UploadId>
                    <Initiated>2026-08-31T00:00:00.000Z</Initiated>
                  </Upload>
                </ListMultipartUploadsResult>"#,
        ),
        xml_ok(
            r#"<?xml version="1.0" encoding="UTF-8"?>
                <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
                  <Bucket>test-bucket</Bucket>
                  <IsTruncated>false</IsTruncated>
                  <Upload>
                    <Key>staging/abc</Key>
                    <UploadId>upload-two</UploadId>
                    <Initiated>2026-08-31T00:00:00.000Z</Initiated>
                  </Upload>
                </ListMultipartUploadsResult>"#,
        ),
    ]);

    let ids = client
        .list_multipart_uploads_for_key("staging/abc")
        .await
        .expect("two pages is an ordinary listing");

    assert_eq!(
        ids,
        vec!["upload-one".to_string(), "upload-two".to_string()]
    );
    let second = replay
        .actual_requests()
        .nth(1)
        .expect("the second page was requested")
        .uri()
        .to_string();
    assert!(
        second.contains("upload-id-marker=upload-one"),
        "the marker from page one carries into page two: {second}"
    );
}

/// A page may name only ONE of the two markers, and that is still a page to
/// follow. The guard reads "neither marker", so it is an `&&`; an `||` there
/// stops on the first page whose key marker happens to be the only one set,
/// silently returning a short list to the orphan reaper -- which then leaves
/// the parts it did not see billing forever.
///
/// `ListMultipartUploads` returns `NextUploadIdMarker` only when the page
/// splits a key's uploads, so a page ending on a key boundary carries the
/// key marker alone. That is the ordinary case, not a corner.
#[tokio::test]
async fn a_page_naming_only_the_key_marker_is_still_followed() {
    let (client, replay) = replay_client(vec![
        xml_ok(
            r#"<?xml version="1.0" encoding="UTF-8"?>
                <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
                  <Bucket>test-bucket</Bucket>
                  <IsTruncated>true</IsTruncated>
                  <NextKeyMarker>staging/abc</NextKeyMarker>
                  <Upload>
                    <Key>staging/abc</Key>
                    <UploadId>upload-one</UploadId>
                    <Initiated>2026-08-31T00:00:00.000Z</Initiated>
                  </Upload>
                </ListMultipartUploadsResult>"#,
        ),
        xml_ok(
            r#"<?xml version="1.0" encoding="UTF-8"?>
                <ListMultipartUploadsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
                  <Bucket>test-bucket</Bucket>
                  <IsTruncated>false</IsTruncated>
                  <Upload>
                    <Key>staging/abc</Key>
                    <UploadId>upload-two</UploadId>
                    <Initiated>2026-08-31T00:00:00.000Z</Initiated>
                  </Upload>
                </ListMultipartUploadsResult>"#,
        ),
    ]);

    let ids = client
        .list_multipart_uploads_for_key("staging/abc")
        .await
        .expect("one marker is enough to continue");

    assert_eq!(
        ids,
        vec!["upload-one".to_string(), "upload-two".to_string()]
    );
    assert_eq!(replay.actual_requests().count(), 2);
}

/// infra `a536db81`. `configure_cors` returns `()` and the crate exposed no
/// readback, so replacing its whole body with `()` was invisible to every
/// test that could be written against it. The request it sends is the
/// observable, and it is the right observable: what the object store ends up
/// configured with is decided entirely by that one PUT.
#[tokio::test]
async fn configuring_cors_sends_the_rule_the_browser_upload_needs() {
    let (client, replay) = replay_client(vec![xml_ok("")]);

    client.configure_cors("https://example.test/").await;

    let requests: Vec<_> = replay.actual_requests().collect();
    assert_eq!(requests.len(), 1, "configure_cors must send a PUT");
    let uri = requests[0].uri().to_string();
    assert!(uri.contains("cors"), "put_bucket_cors, not some other PUT");

    let body = String::from_utf8(
        requests[0]
            .body()
            .bytes()
            .expect("an in-memory XML body")
            .to_vec(),
    )
    .expect("the CORS document is UTF-8");

    // The trailing slash is trimmed: an origin is a scheme-host-port and a
    // browser never sends the slash, so a rule carrying one matches nothing.
    assert!(body.contains("<AllowedOrigin>https://example.test</AllowedOrigin>"));
    for method in ["PUT", "GET", "HEAD"] {
        assert!(
            body.contains(&format!("<AllowedMethod>{method}</AllowedMethod>")),
            "{method} is missing from {body}"
        );
    }
    for header in ["Content-Type", "Cache-Control", "Content-Disposition"] {
        assert!(
            body.contains(&format!("<AllowedHeader>{header}</AllowedHeader>")),
            "{header} is missing from {body}"
        );
    }
    // ETag is what a direct-to-S3 part upload reads back to complete the
    // upload; without exposing it the browser cannot finish a multipart.
    assert!(body.contains("<ExposeHeader>ETag</ExposeHeader>"));
    assert!(body.contains("<MaxAgeSeconds>3600</MaxAgeSeconds>"));
}

/// The readback the wrapper was missing, over a canned response.
#[tokio::test]
async fn cors_rules_come_back_in_the_crates_own_shape() {
    let (client, _replay) = replay_client(vec![xml_ok(
        r#"<?xml version="1.0" encoding="UTF-8"?>
            <CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
              <CORSRule>
                <AllowedOrigin>https://example.test</AllowedOrigin>
                <AllowedMethod>PUT</AllowedMethod>
                <AllowedMethod>GET</AllowedMethod>
                <AllowedHeader>Content-Type</AllowedHeader>
                <ExposeHeader>ETag</ExposeHeader>
                <MaxAgeSeconds>3600</MaxAgeSeconds>
              </CORSRule>
            </CORSConfiguration>"#,
    )]);

    let rules = client.bucket_cors().await.expect("a configured bucket");
    assert_eq!(
        rules,
        vec![CorsRuleView {
            allowed_origins: vec!["https://example.test".to_string()],
            allowed_methods: vec!["PUT".to_string(), "GET".to_string()],
            allowed_headers: vec!["Content-Type".to_string()],
            expose_headers: vec!["ETag".to_string()],
            max_age_seconds: Some(3600),
        }]
    );
}

/// A bucket with no CORS is not a failure. S3 answers `GetBucketCors` with
/// `NoSuchCORSConfiguration`, and a caller asking "what is set" is entitled
/// to hear "nothing" rather than an error it has to pattern-match itself.
#[tokio::test]
async fn a_bucket_with_no_cors_reads_back_as_no_rules() {
    let (client, _replay) = replay_client(vec![xml_err(404, "NoSuchCORSConfiguration")]);

    let rules = client.bucket_cors().await.expect("absence is not an error");
    assert!(rules.is_empty());
}

/// Any other error still is one, or the method above would report a broken
/// endpoint as an unconfigured bucket.
#[tokio::test]
async fn a_cors_read_that_fails_for_another_reason_is_an_error() {
    let (client, _replay) = replay_client(vec![xml_err(403, "AccessDenied")]);

    let err = client
        .bucket_cors()
        .await
        .expect_err("AccessDenied is not an empty CORS configuration");
    assert!(err.contains("get_bucket_cors"), "{err}");
}

const MIB: u64 = 1024 * 1024;

#[test]
fn oracle_accepts_every_plan_the_crate_makes() {
    // Spot sizes across the whole legal range, including the exact
    // boundaries, since those are where tiling arithmetic goes wrong.
    for total in [
        1,
        MULTIPART_MIN_PART_SIZE as u64 - 1,
        MULTIPART_MIN_PART_SIZE as u64,
        MULTIPART_MIN_PART_SIZE as u64 + 1,
        25 * MIB,
        MULTIPART_DEFAULT_PART_SIZE as u64 * MULTIPART_MAX_PARTS as u64,
        MULTIPART_MAX_OBJECT_SIZE - 1,
        MULTIPART_MAX_OBJECT_SIZE,
    ] {
        oracle::check_auto(total);
        oracle::check_plan(total, MULTIPART_MIN_PART_SIZE);
        oracle::check_plan(total, MULTIPART_DEFAULT_PART_SIZE);
    }
    // And the refusals, which must be refusals for a reason that holds.
    oracle::check_auto(0);
    oracle::check_auto(MULTIPART_MAX_OBJECT_SIZE + 1);
    oracle::check_plan(25 * MIB, MULTIPART_MIN_PART_SIZE - 1);
    oracle::check_plan(25 * MIB, MULTIPART_MAX_PART_SIZE as usize + 1);
}

#[test]
fn the_limits_are_the_numbers_the_s3_contract_states() {
    // Every one of these is written as an arithmetic expression, and an
    // expression nothing asserts is a number nothing pins: mutation found
    // that `5 * 1024 * 1024 * 1024` could become `5 + 1024 + 1024 + 1024`
    // and no test noticed (infra `1498fe2a`). The plans built from them
    // would still be self-consistent, which is exactly why the planner's
    // own tests cannot catch it. Stated here in bytes, from the S3 API
    // contract.
    assert_eq!(MULTIPART_MIN_PART_SIZE, 5_242_880, "5 MiB");
    assert_eq!(MULTIPART_MAX_PARTS, 10_000);
    assert_eq!(MULTIPART_MAX_PART_SIZE, 5_368_709_120, "5 GiB");
    assert_eq!(MULTIPART_MAX_OBJECT_SIZE, 5_497_558_138_880, "5 TiB");
    assert_eq!(MULTIPART_DEFAULT_PART_SIZE, 16_777_216, "16 MiB");
    assert_eq!(MAX_PRESIGN_EXPIRY_SECS, 604_800, "7 days, SigV4's maximum");
}

#[test]
#[should_panic(expected = "disagrees with div_ceil")]
fn oracle_catches_a_part_count_the_client_would_reject() {
    // Hand-built, because the crate will not produce it. SyncKit refuses a
    // session whose part_count is not div_ceil (synckit-client
    // `client/blob.rs`), so a plan like this is an upload that can never
    // start.
    //
    // THE TILING PROPERTY HAS NO SUCH TEST, and deliberately so: it cannot
    // be broken by hand. `part_len` and `part_range` both derive from
    // `part_size` and `part_count`, so any plan that satisfies the
    // div_ceil check above necessarily tiles. What establishes that the
    // tiling assertions are observed rather than merely present is the
    // mutation run (infra `1498fe2a`), which changes the derivation itself.
    // That division is the point of running both.
    let plan = MultipartPlan {
        total_size: 25 * MIB,
        part_size: 10 * MIB as usize,
        part_count: 4,
    };
    oracle::check_geometry(&plan);
}

#[test]
fn multipart_plan_divides_with_remainder() {
    // 25 MiB in 10 MiB parts -> 10 + 10 + 5.
    let plan = MultipartPlan::new(25 * MIB, 10 * MIB as usize).unwrap();
    assert_eq!(plan.part_count, 3);
    assert_eq!(plan.part_len(1), 10 * MIB);
    assert_eq!(plan.part_len(2), 10 * MIB);
    assert_eq!(plan.part_len(3), 5 * MIB);
    assert_eq!(plan.part_len(4), 0, "out-of-range part");
    assert_eq!(plan.part_range(1), Some((0, 10 * MIB - 1)));
    assert_eq!(plan.part_range(3), Some((20 * MIB, 25 * MIB - 1)));
    assert_eq!(plan.part_range(4), None);
}

#[test]
fn multipart_plan_divides_evenly() {
    // 20 MiB in 5 MiB parts -> four full parts, last is a full part.
    let plan = MultipartPlan::new(20 * MIB, MULTIPART_MIN_PART_SIZE).unwrap();
    assert_eq!(plan.part_count, 4);
    assert_eq!(plan.part_len(4), 5 * MIB);
    assert_eq!(plan.part_range(4), Some((15 * MIB, 20 * MIB - 1)));
}

#[test]
fn multipart_plan_rejects_empty_object() {
    let err = MultipartPlan::new(0, MULTIPART_MIN_PART_SIZE).unwrap_err();
    assert!(err.contains("non-empty"), "unexpected error: {err}");
}

#[test]
fn multipart_plan_rejects_undersized_part() {
    let err = MultipartPlan::new(100 * MIB, MULTIPART_MIN_PART_SIZE - 1).unwrap_err();
    assert!(err.contains("5 MiB"), "unexpected error: {err}");
}

#[test]
fn multipart_plan_rejects_oversized_part() {
    let err = MultipartPlan::new(10 * MIB, MULTIPART_MAX_PART_SIZE as usize + 1).unwrap_err();
    assert!(err.contains("5 GiB"), "unexpected error: {err}");
}

#[test]
fn multipart_plan_rejects_too_many_parts() {
    // One 5 MiB part past the 10k limit at the minimum part size.
    let total = MULTIPART_MIN_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 1);
    let err = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap_err();
    assert!(err.contains("10000-part"), "unexpected error: {err}");
}

#[test]
fn multipart_plan_accepts_exactly_max_parts() {
    let total = MULTIPART_MIN_PART_SIZE as u64 * MULTIPART_MAX_PARTS as u64;
    let plan = MultipartPlan::new(total, MULTIPART_MIN_PART_SIZE).unwrap();
    assert_eq!(plan.part_count, MULTIPART_MAX_PARTS);
}

#[test]
fn multipart_plan_rejects_over_object_ceiling() {
    let err = MultipartPlan::new(
        MULTIPART_MAX_OBJECT_SIZE + 1,
        MULTIPART_MAX_PART_SIZE as usize,
    )
    .unwrap_err();
    assert!(err.contains("5 TiB"), "unexpected error: {err}");
}

#[test]
fn multipart_plan_auto_uses_default_for_small_objects() {
    let plan = MultipartPlan::auto(100 * MIB).unwrap();
    assert_eq!(plan.part_size, MULTIPART_DEFAULT_PART_SIZE);
    // 100 MiB / 16 MiB -> 7 parts (ceil).
    assert_eq!(plan.part_count, 7);
}

#[test]
fn multipart_plan_auto_scales_part_size_to_stay_within_part_cap() {
    // An object too big for the default part size within 10k parts must get
    // a larger part size, and the resulting plan must be valid.
    let big = MULTIPART_DEFAULT_PART_SIZE as u64 * (MULTIPART_MAX_PARTS as u64 + 500);
    let plan = MultipartPlan::auto(big).unwrap();
    assert!(plan.part_size > MULTIPART_DEFAULT_PART_SIZE);
    assert!(plan.part_count <= MULTIPART_MAX_PARTS);
    // Whole-MiB part size.
    assert_eq!(plan.part_size as u64 % MIB, 0);
}

#[test]
fn multipart_plan_auto_rejects_empty() {
    assert!(MultipartPlan::auto(0).is_err());
}

#[tokio::test]
async fn copy_object_multipart_rejects_empty_source_before_any_request() {
    // Plan is pre-flight: an empty source fails before the destination
    // multipart upload is created, so the unreachable endpoint is untouched.
    let client = test_client();
    let err = client
        .copy_object_multipart("bkt", "src", "dst", "application/octet-stream", 0, None)
        .await
        .expect_err("empty source must be rejected");
    assert!(err.contains("non-empty"), "unexpected error: {err}");
}

/// The `X-Amz-SignedHeaders` list from a presigned URL.
fn signed_headers(url: &str) -> String {
    url.split('&')
        .find_map(|p| p.strip_prefix("X-Amz-SignedHeaders="))
        .map(|v| v.replace("%3B", ";"))
        .expect("presigned URL must carry X-Amz-SignedHeaders")
}

#[tokio::test]
async fn presign_upload_signs_content_length_when_bound() {
    // Callers rely on `max_bytes` being enforced, and it is enforced only
    // because it lands in SignedHeaders: a client sending a different
    // Content-Length then fails the signature. That also makes the declared
    // size a hard contract — a caller that declares anything other than the
    // exact body length breaks every upload — so pin it here rather than
    // discovering it against production S3.
    let client = test_client();

    let bound = client
        .presign_upload("k", "application/octet-stream", 900, None, Some(12_345))
        .await
        .unwrap();
    let headers = signed_headers(&bound);
    assert!(
        headers.contains("content-length"),
        "max_bytes must be signed, got: {headers}"
    );

    let unbound = client
        .presign_upload("k", "application/octet-stream", 900, None, None)
        .await
        .unwrap();
    assert!(
        !signed_headers(&unbound).contains("content-length"),
        "without max_bytes the client is free to send any length"
    );
}

#[tokio::test]
async fn presign_upload_part_rejects_out_of_range_part_number() {
    // Pre-flight range check: fires before any network call, so the
    // unreachable dummy endpoint is never touched.
    let client = test_client();
    for bad in [0, MULTIPART_MAX_PARTS as i32 + 1] {
        let err = client
            .presign_upload_part("k", "uid", bad, 3600, None, None)
            .await
            .expect_err("out-of-range part number must be rejected");
        assert!(err.contains("out of range"), "unexpected error: {err}");
    }
}

#[tokio::test]
async fn presign_upload_part_signs_the_checksum_when_bound() {
    // S3 enforces a bound checksum by rehashing the part, but only if the
    // client sends the header — which it must, because signing it makes it
    // mandatory. Both halves of that live in SignedHeaders.
    let client = test_client();

    let bound = client
        .presign_upload_part("k", "uid", 1, 900, Some(64), Some("Zm9vYmFyYmF6"))
        .await
        .unwrap();
    let headers = signed_headers(&bound);
    assert!(
        headers.contains("x-amz-checksum-sha256"),
        "a bound checksum must be signed, got: {headers}"
    );

    let unbound = client
        .presign_upload_part("k", "uid", 1, 900, Some(64), None)
        .await
        .unwrap();
    assert!(
        !signed_headers(&unbound).contains("checksum"),
        "no checksum bound means no checksum header is required"
    );
}

#[tokio::test]
async fn complete_multipart_rejects_empty_parts() {
    let client = test_client();
    let err = client
        .complete_multipart_upload("k", "uid", &[])
        .await
        .expect_err("empty parts must be rejected");
    assert!(err.contains("no parts"), "unexpected error: {err}");
}

#[tokio::test]
async fn upload_multipart_rejects_undersized_part_before_any_request() {
    // The minimum-part-size guard is pre-flight: it must fire before the
    // multipart upload is created, so there is nothing to strand and no
    // network call (the dummy endpoint is unreachable — reaching it would
    // hang/error instead of returning this exact message).
    let client = test_client();
    let path = std::path::Path::new("/nonexistent");
    let err = client
        .upload_multipart("k", "application/octet-stream", path, Some(1024))
        .await
        .expect_err("undersized part size must be rejected");
    assert!(err.contains("at least 5 MB"), "unexpected error: {err}");
}