Repository: getmoto/moto (python)

Issue:
S3: get/head with partNumber is returning 200 status code instead of 206
```
s3 = boto3.resource("s3")
s3.create_bucket(Bucket="bucket")
s3.Object("bucket", "some-key").put(Body="foo")

assert s3.meta.client.head_object(Bucket="bucket", Key="some-key", PartNumber=1)["ResponseMetadata"]["HTTPStatusCode"] == 206
```

Since it's effectively a range request, it should return 206, not 200.  Same for `get_object`.

Pull request:
S3: head_object()/get_object() now support the PartNumber for regular keys
Fixes #7779 

If we call `get_object(.., PartNumber=x)` on a multipart upload, we should probably return only that specific part - but that's not yet implemented.
For now, we only support `PartNumber=1` on an entire key, and treat it as a RangeRequest for the entire key (just like AWS does).

Patch:
diff --git a/moto/s3/exceptions.py b/moto/s3/exceptions.py
index 7f12ae885..237096533 100644
--- a/moto/s3/exceptions.py
+++ b/moto/s3/exceptions.py
@@ -481,6 +481,13 @@ class InvalidRange(S3ClientError):
         )
 
 
+class RangeNotSatisfiable(S3ClientError):
+    code = 416
+
+    def __init__(self) -> None:
+        super().__init__(RangeNotSatisfiable.code, "Requested Range Not Satisfiable")
+
+
 class InvalidContinuationToken(S3ClientError):
     code = 400
 
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index f451252c7..c9c899776 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -50,6 +50,7 @@ from .exceptions import (
     NotAnIntegerException,
     ObjectNotInActiveTierError,
     PreconditionFailed,
+    RangeNotSatisfiable,
     S3AclAndGrantError,
     S3ClientError,
 )
@@ -1423,6 +1424,10 @@ class S3Response(BaseResponse):
                     part_number_marker=part_number_marker,
                 ),
             )
+
+        part_number = query.get("partNumber", [None])[0]
+        if part_number:
+            part_number = int(part_number)
         version_id = query.get("versionId", [None])[0]
         if_modified_since = headers.get("If-Modified-Since", None)
         if_match = headers.get("If-Match", None)
@@ -1481,6 +1486,15 @@ class S3Response(BaseResponse):
         response_headers.update(key.metadata)
         response_headers.update(key.response_dict)
         response_headers.update({"Accept-Ranges": "bytes"})
+        if part_number:
+            if key.multipart:
+                # TODO
+                pass
+            else:
+                if part_number > 1:
+                    raise RangeNotSatisfiable
+                response_headers["content-range"] = f"bytes 0-{key.size - 1}/{key.size}"
+                return 206, response_headers, key.value
         return 200, response_headers, key.value
 
     def _key_response_put(
@@ -1834,6 +1848,13 @@ class S3Response(BaseResponse):
                 if full_key.multipart:  # type: ignore
                     mp_part_count = str(len(full_key.multipart.partlist))  # type: ignore
                     response_headers["x-amz-mp-parts-count"] = mp_part_count
+                else:
+                    if part_number > 1:
+                        raise RangeNotSatisfiable
+                    response_headers["content-range"] = (
+                        f"bytes 0-{full_key.size -1}/{full_key.size}"  # type: ignore
+                    )
+                    return 206, response_headers, ""
 
             return 200, response_headers, ""
         else:


Test patch:
diff --git a/tests/test_s3/__init__.py b/tests/test_s3/__init__.py
index e89b1490c..55d0f0c7a 100644
--- a/tests/test_s3/__init__.py
+++ b/tests/test_s3/__init__.py
@@ -23,7 +23,7 @@ def s3_aws_verified(func):
     """
 
     @wraps(func)
-    def pagination_wrapper():
+    def pagination_wrapper(**kwargs):
         bucket_name = str(uuid4())
 
         allow_aws_request = (
@@ -32,13 +32,13 @@ def s3_aws_verified(func):
 
         if allow_aws_request:
             print(f"Test {func} will create {bucket_name}")  # noqa: T201
-            resp = create_bucket_and_test(bucket_name)
+            resp = create_bucket_and_test(bucket_name, **kwargs)
         else:
             with mock_aws():
-                resp = create_bucket_and_test(bucket_name)
+                resp = create_bucket_and_test(bucket_name, **kwargs)
         return resp
 
-    def create_bucket_and_test(bucket_name):
+    def create_bucket_and_test(bucket_name, **kwargs):
         client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
 
         client.create_bucket(Bucket=bucket_name)
@@ -47,7 +47,7 @@ def s3_aws_verified(func):
             Tagging={"TagSet": [{"Key": "environment", "Value": "moto_tests"}]},
         )
         try:
-            resp = func(bucket_name)
+            resp = func(**kwargs, bucket_name=bucket_name)
         finally:
             ### CLEANUP ###
 
diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py
index 8c1184e3a..00f43216e 100644
--- a/tests/test_s3/test_s3.py
+++ b/tests/test_s3/test_s3.py
@@ -1615,37 +1615,73 @@ def test_bucket_create_empty_bucket_configuration_should_return_malformed_xml_er
     assert exc.value.response["ResponseMetadata"]["HTTPStatusCode"] == 400
 
 
-@mock_aws
-def test_head_object():
+@s3_aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize("size", [10, 10000000])
+def test_head_object(size, bucket_name=None):
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
-    s3_resource.create_bucket(Bucket="blah")
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
 
-    s3_resource.Object("blah", "hello.txt").put(Body="some text")
+    s3_resource.Object(bucket_name, "hello.txt").put(Body="x" * size)
 
-    s3_resource.Object("blah", "hello.txt").meta.client.head_object(
-        Bucket="blah", Key="hello.txt"
+    resp = client.head_object(Bucket=bucket_name, Key="hello.txt")
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert "content-range" not in resp["ResponseMetadata"]["HTTPHeaders"]
+    assert resp["ContentLength"] == size
+    assert resp["AcceptRanges"] == "bytes"
+
+    resp = client.head_object(Bucket=bucket_name, Key="hello.txt", PartNumber=1)
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 206
+    assert (
+        resp["ResponseMetadata"]["HTTPHeaders"]["content-range"]
+        == f"bytes 0-{size-1}/{size}"
     )
+    assert resp["ContentLength"] == size
+    assert resp["AcceptRanges"] == "bytes"
 
     with pytest.raises(ClientError) as exc:
-        s3_resource.Object("blah", "hello2.txt").meta.client.head_object(
-            Bucket="blah", Key="hello_bad.txt"
-        )
+        client.head_object(Bucket=bucket_name, Key="hello.txt", PartNumber=2)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "416"
+    assert err["Message"] == "Requested Range Not Satisfiable"
+
+    with pytest.raises(ClientError) as exc:
+        client.head_object(Bucket=bucket_name, Key="hello_bad.txt")
     assert exc.value.response["Error"]["Code"] == "404"
 
 
-@mock_aws
-def test_get_object():
+@s3_aws_verified
+@pytest.mark.aws_verified
+@pytest.mark.parametrize("size", [10, 10000000])
+def test_get_object(size, bucket_name=None):
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
-    s3_resource.create_bucket(Bucket="blah")
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
 
-    s3_resource.Object("blah", "hello.txt").put(Body="some text")
+    s3_resource.Object(bucket_name, "hello.txt").put(Body="x" * size)
 
-    s3_resource.Object("blah", "hello.txt").meta.client.head_object(
-        Bucket="blah", Key="hello.txt"
+    resp = client.get_object(Bucket=bucket_name, Key="hello.txt")
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert "content-range" not in resp["ResponseMetadata"]["HTTPHeaders"]
+    assert resp["ContentLength"] == size
+    assert resp["AcceptRanges"] == "bytes"
+
+    resp = client.get_object(Bucket=bucket_name, Key="hello.txt", PartNumber=1)
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 206
+    assert (
+        resp["ResponseMetadata"]["HTTPHeaders"]["content-range"]
+        == f"bytes 0-{size - 1}/{size}"
     )
+    assert resp["ContentLength"] == size
+    assert resp["AcceptRanges"] == "bytes"
+
+    with pytest.raises(ClientError) as exc:
+        client.head_object(Bucket=bucket_name, Key="hello.txt", PartNumber=2)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "416"
+    assert err["Message"] == "Requested Range Not Satisfiable"
 
     with pytest.raises(ClientError) as exc:
-        s3_resource.Object("blah", "hello2.txt").get()
+        s3_resource.Object(bucket_name, "hello2.txt").get()
 
     assert exc.value.response["Error"]["Code"] == "NoSuchKey"
 
@@ -1766,27 +1802,27 @@ def test_delete_versioned_bucket():
 
 @pytest.mark.aws_verified
 @s3_aws_verified
-def test_delete_versioned_bucket_returns_metadata(name=None):
+def test_delete_versioned_bucket_returns_metadata(bucket_name=None):
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
     resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
-    bucket = resource.Bucket(name)
+    bucket = resource.Bucket(bucket_name)
     versions = bucket.object_versions
 
-    enable_versioning(name, client)
+    enable_versioning(bucket_name, client)
 
-    client.put_object(Bucket=name, Key="test1", Body=b"test1")
+    client.put_object(Bucket=bucket_name, Key="test1", Body=b"test1")
 
     # We now have zero delete markers
-    assert "DeleteMarkers" not in client.list_object_versions(Bucket=name)
+    assert "DeleteMarkers" not in client.list_object_versions(Bucket=bucket_name)
 
     # Delete the object
-    del_file = client.delete_object(Bucket=name, Key="test1")
+    del_file = client.delete_object(Bucket=bucket_name, Key="test1")
     deleted_version_id = del_file["VersionId"]
     assert del_file["DeleteMarker"] is True
     assert deleted_version_id is not None
 
     # We now have one DeleteMarker
-    assert len(client.list_object_versions(Bucket=name)["DeleteMarkers"]) == 1
+    assert len(client.list_object_versions(Bucket=bucket_name)["DeleteMarkers"]) == 1
 
     # list_object_versions returns the object itself, and a DeleteMarker
     # object.head() returns a 'x-amz-delete-marker' header
@@ -1808,7 +1844,7 @@ def test_delete_versioned_bucket_returns_metadata(name=None):
     # Note that delete_marker.head() returns a regular 404
     # i.e., without specifying the versionId
     with pytest.raises(ClientError) as exc:
-        client.head_object(Bucket=name, Key="test1")
+        client.head_object(Bucket=bucket_name, Key="test1")
     err = exc.value.response
     assert err["Error"] == {"Code": "404", "Message": "Not Found"}
     assert err["ResponseMetadata"]["HTTPStatusCode"] == 404
@@ -1818,16 +1854,16 @@ def test_delete_versioned_bucket_returns_metadata(name=None):
     )
 
     # Delete the same object gives a new version id
-    del_mrk1 = client.delete_object(Bucket=name, Key="test1")
+    del_mrk1 = client.delete_object(Bucket=bucket_name, Key="test1")
     assert del_mrk1["DeleteMarker"] is True
     assert del_mrk1["VersionId"] != del_file["VersionId"]
 
     # We now have two DeleteMarkers
-    assert len(client.list_object_versions(Bucket=name)["DeleteMarkers"]) == 2
+    assert len(client.list_object_versions(Bucket=bucket_name)["DeleteMarkers"]) == 2
 
     # Delete the delete marker
     del_mrk2 = client.delete_object(
-        Bucket=name, Key="test1", VersionId=del_mrk1["VersionId"]
+        Bucket=bucket_name, Key="test1", VersionId=del_mrk1["VersionId"]
     )
     assert del_mrk2["DeleteMarker"] is True
     assert del_mrk2["VersionId"] == del_mrk1["VersionId"]
@@ -1847,19 +1883,19 @@ def test_delete_versioned_bucket_returns_metadata(name=None):
             assert version.head()["ResponseMetadata"]["HTTPStatusCode"] == 200
 
     # We now have only one DeleteMarker
-    assert len(client.list_object_versions(Bucket=name)["DeleteMarkers"]) == 1
+    assert len(client.list_object_versions(Bucket=bucket_name)["DeleteMarkers"]) == 1
 
     # Delete the actual file
-    actual_version = client.list_object_versions(Bucket=name)["Versions"][0]
+    actual_version = client.list_object_versions(Bucket=bucket_name)["Versions"][0]
     del_mrk3 = client.delete_object(
-        Bucket=name, Key="test1", VersionId=actual_version["VersionId"]
+        Bucket=bucket_name, Key="test1", VersionId=actual_version["VersionId"]
     )
     assert "DeleteMarker" not in del_mrk3
     assert del_mrk3["VersionId"] == actual_version["VersionId"]
 
     # We still have one DeleteMarker, but zero objects
-    assert len(client.list_object_versions(Bucket=name)["DeleteMarkers"]) == 1
-    assert "Versions" not in client.list_object_versions(Bucket=name)
+    assert len(client.list_object_versions(Bucket=bucket_name)["DeleteMarkers"]) == 1
+    assert "Versions" not in client.list_object_versions(Bucket=bucket_name)
 
     # Because we only have DeleteMarkers, we can not call `head()` on any of othem
     for version in versions.filter(Prefix="test1"):
@@ -1870,7 +1906,7 @@ def test_delete_versioned_bucket_returns_metadata(name=None):
 
     # Delete the last marker
     del_mrk4 = client.delete_object(
-        Bucket=name, Key="test1", VersionId=del_mrk2["VersionId"]
+        Bucket=bucket_name, Key="test1", VersionId=del_mrk2["VersionId"]
     )
     assert "DeleteMarker" not in del_mrk4
     assert del_mrk4["VersionId"] == del_mrk2["VersionId"]
diff --git a/tests/test_s3/test_s3_copyobject.py b/tests/test_s3/test_s3_copyobject.py
index c734d22c2..1dd900f13 100644
--- a/tests/test_s3/test_s3_copyobject.py
+++ b/tests/test_s3/test_s3_copyobject.py
@@ -39,7 +39,7 @@ def test_copy_key_boto3(key_name):
 
 @pytest.mark.aws_verified
 @s3_aws_verified
-def test_copy_key_boto3_with_args(bucket=None):
+def test_copy_key_boto3_with_args(bucket_name=None):
     # Setup
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
@@ -47,13 +47,13 @@ def test_copy_key_boto3_with_args(bucket=None):
     new_key = "new_key"
     expected_hash = "qz0H8xacy9DtbEtF3iFRn5+TjHLSQSSZiquUnOg7tRs="
 
-    key = s3_resource.Object(bucket, key_name)
+    key = s3_resource.Object(bucket_name, key_name)
     key.put(Body=b"some value")
 
     # Execute
-    key2 = s3_resource.Object(bucket, new_key)
+    key2 = s3_resource.Object(bucket_name, new_key)
     key2.copy(
-        CopySource={"Bucket": bucket, "Key": key_name},
+        CopySource={"Bucket": bucket_name, "Key": key_name},
         ExtraArgs={
             "ChecksumAlgorithm": "SHA256",
             "WebsiteRedirectLocation": "http://getmoto.org/",
@@ -62,20 +62,20 @@ def test_copy_key_boto3_with_args(bucket=None):
 
     # Verify
     resp = client.get_object_attributes(
-        Bucket=bucket, Key=new_key, ObjectAttributes=["Checksum"]
+        Bucket=bucket_name, Key=new_key, ObjectAttributes=["Checksum"]
     )
 
     assert "Checksum" in resp
     assert "ChecksumSHA256" in resp["Checksum"]
     assert resp["Checksum"]["ChecksumSHA256"] == expected_hash
 
-    obj = client.get_object(Bucket=bucket, Key=new_key)
+    obj = client.get_object(Bucket=bucket_name, Key=new_key)
     assert obj["WebsiteRedirectLocation"] == "http://getmoto.org/"
 
     # Verify in place
     copy_in_place = client.copy_object(
-        Bucket=bucket,
-        CopySource=f"{bucket}/{new_key}",
+        Bucket=bucket_name,
+        CopySource=f"{bucket_name}/{new_key}",
         Key=new_key,
         ChecksumAlgorithm="SHA256",
         MetadataDirective="REPLACE",
@@ -87,7 +87,7 @@ def test_copy_key_boto3_with_args(bucket=None):
 
 @pytest.mark.aws_verified
 @s3_aws_verified
-def test_copy_key_boto3_with_args__using_multipart(bucket=None):
+def test_copy_key_boto3_with_args__using_multipart(bucket_name=None):
     # Setup
     s3_resource = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
     client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
@@ -95,13 +95,13 @@ def test_copy_key_boto3_with_args__using_multipart(bucket=None):
     new_key = "new_key"
     expected_hash = "DnKotDi4EtYGwNMDKmnR6SqH3bWVOlo2BC+tsz9rHqw="
 
-    key = s3_resource.Object(bucket, key_name)
+    key = s3_resource.Object(bucket_name, key_name)
     key.put(Body=b"some value")
 
     # Execute
-    key2 = s3_resource.Object(bucket, new_key)
+    key2 = s3_resource.Object(bucket_name, new_key)
     key2.copy(
-        CopySource={"Bucket": bucket, "Key": key_name},
+        CopySource={"Bucket": bucket_name, "Key": key_name},
         ExtraArgs={
             "ChecksumAlgorithm": "SHA256",
             "WebsiteRedirectLocation": "http://getmoto.org/",
@@ -111,14 +111,14 @@ def test_copy_key_boto3_with_args__using_multipart(bucket=None):
 
     # Verify
     resp = client.get_object_attributes(
-        Bucket=bucket, Key=new_key, ObjectAttributes=["Checksum"]
+        Bucket=bucket_name, Key=new_key, ObjectAttributes=["Checksum"]
     )
 
     assert "Checksum" in resp
     assert "ChecksumSHA256" in resp["Checksum"]
     assert resp["Checksum"]["ChecksumSHA256"] == expected_hash
 
-    obj = client.get_object(Bucket=bucket, Key=new_key)
+    obj = client.get_object(Bucket=bucket_name, Key=new_key)
     assert obj["WebsiteRedirectLocation"] == "http://getmoto.org/"
 
 
diff --git a/tests/test_s3/test_s3_list_objects.py b/tests/test_s3/test_s3_list_objects.py
index f436aa22f..fe6d2749b 100644
--- a/tests/test_s3/test_s3_list_objects.py
+++ b/tests/test_s3/test_s3_list_objects.py
@@ -10,7 +10,7 @@ from . import s3_aws_verified
 
 @pytest.mark.aws_verified
 @s3_aws_verified
-def test_s3_filter_with_manual_marker(name=None) -> None:
+def test_s3_filter_with_manual_marker(bucket_name=None) -> None:
     """
     Batch manually:
     1. Get list of all items, but only take first 5
@@ -22,7 +22,7 @@ def test_s3_filter_with_manual_marker(name=None) -> None:
 
     This test verifies that it is possible to pass an existing key-name as the Marker-attribute
     """
-    bucket = boto3.resource("s3", "us-east-1").Bucket(name)
+    bucket = boto3.resource("s3", "us-east-1").Bucket(bucket_name)
 
     def batch_from(marker: str, *, batch_size: int):
         first: str | None = None