diff --git a/tests/blackbox/test_cp_command.py b/tests/blackbox/test_cp_command.py
index f2e8151a63f1..4dda85677464 100644
--- a/tests/blackbox/test_cp_command.py
+++ b/tests/blackbox/test_cp_command.py
@@ -7,7 +7,13 @@
 
 import pytest
 from localstub.handlers import handle_expect_header
-from localstub.server import DropConnection, FaultyTransmission, HTTPResponse
+from localstub.server import (
+    DropConnection,
+    FaultyTransmission,
+    HTTPResponse,
+    ImmediateTransmission,
+    TruncateBody,
+)
 
 from tests.blackbox.s3_assertions import (
     assert_abort_multipart_upload,
@@ -52,7 +58,7 @@
 
 
 def relative_path(filename):
-    """Cross platform relative path of a filename."""
+    """Cross-platform relative path of a filename."""
     try:
         dirname, basename = os.path.split(filename)
         relative_dir = os.path.relpath(dirname)
@@ -6498,3 +6504,314 @@ async def test_content_type_not_guessed_on_s3_to_s3_copy(aws_cli, tmp_path):
     assert (
         ct != "text/html"
     ), f"Content-Type should not be guessed on s3-to-s3 copy, got {ct!r}"
+
+
+@pytest.mark.skip(
+    reason="urllib3 2.x enforce_content_length regression: "
+    "urllib3 raises ProtocolError before botocore's "
+    "IncompleteReadError can fire. Fix: "
+    "https://github.com/aws/aws-cli/pull/10622"
+)
+@pytest.mark.asyncio
+async def test_streaming_download_retries_on_truncated_response(
+    aws_cli, tmp_path
+):
+    """cp s3://bucket/key - retries when server closes mid-transfer.
+
+    Simulates the production scenario where a slow consumer (stdout pipe)
+    causes the CLI to pause reading, S3 times out the idle connection,
+    and the CLI sees EOF before reading all expected bytes. botocore
+    raises IncompleteReadError, which s3transfer
+    catches and retries. The partial data from the first attempt is
+    already written to stdout (non-seekable), so the final output
+    contains bytes from both the truncated and retried responses.
+    """
+    body = b"A" * 10
+    async with mock_server(on_headers_received=handle_expect_header) as (
+        server,
+        proxy,
+    ):
+        setup_responses(
+            server,
+            [
+                head_object_response(
+                    content_length=10,
+                    **{
+                        "Accept-Ranges": "bytes",
+                        "Content-Type": "binary/octet-stream",
+                    },
+                ),
+                # First GET: server sends 5 of 10 bytes then closes.
+                # Connection: close triggers a clean TLS shutdown so the
+                # client sees EOF rather than a socket error. This is
+                # how we simulate S3 closing an idle connection
+                HTTPResponse.raw(
+                    body,
+                    status=200,
+                    headers={
+                        "Content-Length": "10",
+                        "ETag": '"foo-1"',
+                        "Accept-Ranges": "bytes",
+                        "Connection": "close",
+                    },
+                ),
+                # Retry: full response
+                get_object_response(
+                    body,
+                    **{
+                        "Accept-Ranges": "bytes",
+                        "Content-Type": "binary/octet-stream",
+                    },
+                ),
+            ],
+        )
+        server.set_transmission_strategy(FaultyTransmission([TruncateBody(5)]))
+
+        async def reset_after_truncated():
+            # HeadObject
+            await server.next_request()
+            # First GET (truncated)
+            await server.next_request()
+            server.set_transmission_strategy(ImmediateTransmission())
+
+        (stdout, stderr, rc), _ = await asyncio.gather(
+            run_cli(
+                aws_cli,
+                ["s3", "cp", "s3://bucket/key.txt", "-"],
+                cli_env(proxy),
+            ),
+            reset_after_truncated(),
+        )
+
+    assert rc == 0, stderr.decode()
+    assert len(server.requests) == 3, format_requests(server)
+    assert_head_object(server.requests[0], Bucket="bucket", Key="key.txt")
+    assert_get_object(server.requests[1], Bucket="bucket", Key="key.txt")
+    assert_get_object(server.requests[2], Bucket="bucket", Key="key.txt")
+    assert len(stdout) == 10
+    assert stdout == b"A" * 10
+
+
+@pytest.mark.asyncio
+async def test_upload_retries_on_502(aws_cli, tmp_path):
+    """cp retries on 502 BadGateway and succeeds on second attempt.
+
+    Verifies the Python CLI retries 502 responses.
+    """
+    src = tmp_path / "foo.txt"
+    src.write_text("content")
+    async with mock_server(on_headers_received=handle_expect_header) as (
+        server,
+        proxy,
+    ):
+        setup_responses(
+            server,
+            [
+                error_response("BadGateway", "Bad Gateway", status=502),
+                put_object_response(),
+            ],
+        )
+        env = cli_env(proxy)
+        env["AWS_MAX_ATTEMPTS"] = "2"
+        stdout, stderr, rc = await run_cli(
+            aws_cli,
+            ["s3", "cp", str(src), "s3://bucket/foo.txt"],
+            env,
+        )
+
+    assert rc == 0, stderr.decode()
+    assert len(server.requests) == 2, format_requests(server)
+    assert_put_object(server.requests[0], Bucket="bucket", Key="foo.txt")
+    assert_put_object(server.requests[1], Bucket="bucket", Key="foo.txt")
+
+
+@pytest.mark.asyncio
+async def test_upload_retries_on_504(aws_cli, tmp_path):
+    """cp retries on 504 GatewayTimeout and succeeds on second attempt.
+
+    Verifies the Python CLI retries 504 responses.
+    """
+    src = tmp_path / "foo.txt"
+    src.write_text("content")
+    async with mock_server(on_headers_received=handle_expect_header) as (
+        server,
+        proxy,
+    ):
+        setup_responses(
+            server,
+            [
+                error_response(
+                    "GatewayTimeout", "Gateway Timeout", status=504
+                ),
+                put_object_response(),
+            ],
+        )
+        env = cli_env(proxy)
+        env["AWS_MAX_ATTEMPTS"] = "2"
+        stdout, stderr, rc = await run_cli(
+            aws_cli,
+            ["s3", "cp", str(src), "s3://bucket/foo.txt"],
+            env,
+        )
+
+    assert rc == 0, stderr.decode()
+    assert len(server.requests) == 2, format_requests(server)
+    assert_put_object(server.requests[0], Bucket="bucket", Key="foo.txt")
+    assert_put_object(server.requests[1], Bucket="bucket", Key="foo.txt")
+
+
+@pytest.mark.asyncio
+async def test_upload_retries_on_500(aws_cli, tmp_path):
+    """cp retries on 500 InternalError and succeeds on second attempt."""
+    src = tmp_path / "foo.txt"
+    src.write_text("content")
+    async with mock_server(on_headers_received=handle_expect_header) as (
+        server,
+        proxy,
+    ):
+        setup_responses(
+            server,
+            [
+                error_response(
+                    "InternalError", "Internal Server Error", status=500
+                ),
+                put_object_response(),
+            ],
+        )
+        env = cli_env(proxy)
+        env["AWS_MAX_ATTEMPTS"] = "2"
+        stdout, stderr, rc = await run_cli(
+            aws_cli,
+            ["s3", "cp", str(src), "s3://bucket/foo.txt"],
+            env,
+        )
+
+    assert rc == 0, stderr.decode()
+    assert len(server.requests) == 2, format_requests(server)
+    assert_put_object(server.requests[0], Bucket="bucket", Key="foo.txt")
+    assert_put_object(server.requests[1], Bucket="bucket", Key="foo.txt")
+
+
+@pytest.mark.asyncio
+async def test_upload_follows_301_region_redirect(aws_cli, tmp_path):
+    """cp follows 301 PermanentRedirect to the correct bucket region.
+
+    When S3 returns 301 with x-amz-bucket-region, the CLI retries the
+    request against the indicated region.
+    """
+    src = tmp_path / "foo.txt"
+    src.write_text("content")
+    redirect_body = (
+        '<?xml version="1.0" encoding="UTF-8"?>'
+        "<Error>"
+        "<Code>PermanentRedirect</Code>"
+        "<Message>The bucket must be addressed using the specified endpoint.</Message>"
+        "<Bucket>bucket</Bucket>"
+        "</Error>"
+    )
+    async with mock_server(on_headers_received=handle_expect_header) as (
+        server,
+        proxy,
+    ):
+        setup_responses(
+            server,
+            [
+                HTTPResponse.raw(
+                    redirect_body.encode(),
+                    status=301,
+                    headers={
+                        "Content-Type": "application/xml",
+                        "x-amz-bucket-region": "eu-west-1",
+                    },
+                ),
+                put_object_response(),
+            ],
+        )
+        stdout, stderr, rc = await run_cli(
+            aws_cli,
+            [
+                "s3",
+                "cp",
+                str(src),
+                "s3://bucket/foo.txt",
+                "--region",
+                "us-east-1",
+            ],
+            cli_env(proxy),
+        )
+
+    assert rc == 0, stderr.decode()
+    assert len(server.requests) == 2, format_requests(server)
+    # First request goes to the configured region
+    assert (
+        server.requests[0].headers.get("host")
+        == "bucket.s3.us-east-1.amazonaws.com"
+    )
+    # Redirect sends the retry to the correct region
+    assert (
+        server.requests[1].headers.get("host")
+        == "bucket.s3.eu-west-1.amazonaws.com"
+    )
+
+
+@pytest.mark.asyncio
+async def test_upload_follows_auth_header_malformed_redirect(
+    aws_cli, tmp_path
+):
+    """cp follows AuthorizationHeaderMalformed error to the correct region.
+
+    S3 returns 400 with the correct region in the error body when the
+    request is signed for the wrong region.  The CLI extracts the region
+    and retries.
+    """
+    src = tmp_path / "foo.txt"
+    src.write_text("content")
+    error_body = (
+        '<?xml version="1.0" encoding="UTF-8"?>'
+        "<Error>"
+        "<Code>AuthorizationHeaderMalformed</Code>"
+        "<Message>The authorization header is malformed</Message>"
+        "<Region>eu-west-1</Region>"
+        "</Error>"
+    )
+    async with mock_server(on_headers_received=handle_expect_header) as (
+        server,
+        proxy,
+    ):
+        setup_responses(
+            server,
+            [
+                HTTPResponse.raw(
+                    error_body.encode(),
+                    status=400,
+                    headers={
+                        "Content-Type": "application/xml",
+                        "x-amz-bucket-region": "eu-west-1",
+                    },
+                ),
+                put_object_response(),
+            ],
+        )
+        stdout, stderr, rc = await run_cli(
+            aws_cli,
+            [
+                "s3",
+                "cp",
+                str(src),
+                "s3://bucket/foo.txt",
+                "--region",
+                "us-east-1",
+            ],
+            cli_env(proxy),
+        )
+
+    assert rc == 0, stderr.decode()
+    assert len(server.requests) == 2, format_requests(server)
+    assert (
+        server.requests[0].headers.get("host")
+        == "bucket.s3.us-east-1.amazonaws.com"
+    )
+    assert (
+        server.requests[1].headers.get("host")
+        == "bucket.s3.eu-west-1.amazonaws.com"
+    )
diff --git a/tests/blackbox/test_mb_command.py b/tests/blackbox/test_mb_command.py
index cebd52590a97..22c28ca3965b 100644
--- a/tests/blackbox/test_mb_command.py
+++ b/tests/blackbox/test_mb_command.py
@@ -319,7 +319,7 @@ async def test_create_bucket_with_non_ascii_tag_value(aws_cli):
 
     assert rc == 0, stderr.decode()
     req = server.requests[0]
-    body_text = req.body.decode("utf-8") if req.body else ""
-    assert (
-        "José" in body_text
-    ), f"Expected non-ASCII tag value in body, got: {body_text[:200]}"
+    body_text = req.body if isinstance(req.body, str) else (req.body.decode("utf-8") if req.body else "")
+    assert "José" in body_text, (
+        f"Expected non-ASCII tag value in body, got: {body_text[:200]}"
+    )