Zakaria
500 lines
20 KiB
Python
500 lines
20 KiB
Python
"""S3 bucket and object integration tests."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import random
|
|
import zlib
|
|
|
|
import pytest
|
|
from boto3.s3.transfer import TransferConfig
|
|
from botocore.exceptions import ClientError
|
|
|
|
MIB = 1024 * 1024
|
|
|
|
|
|
class TestS3Bucket:
|
|
"""Test S3 bucket operations."""
|
|
|
|
def test_create_bucket(self, s3_client, unique_name):
|
|
"""Test CreateBucket creates a bucket."""
|
|
bucket_name = f"pytest-s3-{unique_name}"
|
|
|
|
try:
|
|
s3_client.create_bucket(Bucket=bucket_name)
|
|
# Verify bucket exists
|
|
s3_client.head_bucket(Bucket=bucket_name)
|
|
finally:
|
|
s3_client.delete_bucket(Bucket=bucket_name)
|
|
|
|
def test_create_bucket_with_location_constraint(self, s3_client, unique_name):
|
|
"""Test CreateBucket with LocationConstraint (regression: issue #11)."""
|
|
bucket_name = f"pytest-s3-eu-{unique_name}"
|
|
|
|
try:
|
|
s3_client.create_bucket(
|
|
Bucket=bucket_name,
|
|
CreateBucketConfiguration={"LocationConstraint": "eu-central-1"},
|
|
)
|
|
|
|
response = s3_client.get_bucket_location(Bucket=bucket_name)
|
|
assert response.get("LocationConstraint") == "eu-central-1"
|
|
finally:
|
|
s3_client.delete_bucket(Bucket=bucket_name)
|
|
|
|
def test_list_buckets(self, s3_client, unique_name):
|
|
"""Test ListBuckets returns created bucket."""
|
|
bucket_name = f"pytest-s3-{unique_name}"
|
|
|
|
s3_client.create_bucket(Bucket=bucket_name)
|
|
try:
|
|
response = s3_client.list_buckets()
|
|
assert any(b["Name"] == bucket_name for b in response["Buckets"])
|
|
finally:
|
|
s3_client.delete_bucket(Bucket=bucket_name)
|
|
|
|
def test_head_bucket(self, s3_client, test_bucket):
|
|
"""Test HeadBucket succeeds for existing bucket."""
|
|
s3_client.head_bucket(Bucket=test_bucket)
|
|
# If no exception, test passes
|
|
|
|
def test_head_bucket_non_existent(self, s3_client):
|
|
"""Test HeadBucket returns 404 for non-existent bucket."""
|
|
with pytest.raises(ClientError) as exc_info:
|
|
s3_client.head_bucket(Bucket="non-existent-bucket-pytest-xyz")
|
|
assert exc_info.value.response["ResponseMetadata"]["HTTPStatusCode"] == 404
|
|
|
|
def test_get_bucket_location(self, s3_client, test_bucket):
|
|
"""Test GetBucketLocation returns location constraint."""
|
|
response = s3_client.get_bucket_location(Bucket=test_bucket)
|
|
assert "LocationConstraint" in response
|
|
|
|
def test_delete_bucket(self, s3_client, unique_name):
|
|
"""Test DeleteBucket removes bucket."""
|
|
bucket_name = f"pytest-s3-{unique_name}"
|
|
|
|
s3_client.create_bucket(Bucket=bucket_name)
|
|
s3_client.delete_bucket(Bucket=bucket_name)
|
|
|
|
with pytest.raises(ClientError):
|
|
s3_client.head_bucket(Bucket=bucket_name)
|
|
|
|
|
|
class TestS3Object:
|
|
"""Test S3 object operations."""
|
|
|
|
def test_put_object(self, s3_client, test_bucket):
|
|
"""Test PutObject uploads object."""
|
|
key = "test-file.txt"
|
|
content = b"Hello from pytest!"
|
|
|
|
s3_client.put_object(
|
|
Bucket=test_bucket, Key=key, Body=content, ContentType="text/plain"
|
|
)
|
|
|
|
# Verify object exists
|
|
response = s3_client.head_object(Bucket=test_bucket, Key=key)
|
|
assert response["ContentLength"] == len(content)
|
|
|
|
def test_list_objects(self, s3_client, test_bucket):
|
|
"""Test ListObjectsV2 returns uploaded objects."""
|
|
key = "test-file.txt"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=b"content")
|
|
|
|
response = s3_client.list_objects_v2(Bucket=test_bucket)
|
|
assert any(o["Key"] == key for o in response.get("Contents", []))
|
|
|
|
def test_get_object(self, s3_client, test_bucket):
|
|
"""Test GetObject retrieves correct content."""
|
|
key = "test-file.txt"
|
|
content = b"Hello from pytest!"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=content)
|
|
|
|
response = s3_client.get_object(Bucket=test_bucket, Key=key)
|
|
data = response["Body"].read()
|
|
assert data == content
|
|
|
|
def test_head_object(self, s3_client, test_bucket):
|
|
"""Test HeadObject returns metadata."""
|
|
key = "test-file.txt"
|
|
content = b"Hello from pytest!"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=content)
|
|
|
|
response = s3_client.head_object(Bucket=test_bucket, Key=key)
|
|
assert response["ContentLength"] == len(content)
|
|
# LastModified should have second precision (microsecond == 0)
|
|
assert response["LastModified"].microsecond == 0
|
|
|
|
def test_delete_object(self, s3_client, test_bucket):
|
|
"""Test DeleteObject removes object."""
|
|
key = "test-file.txt"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=b"content")
|
|
|
|
s3_client.delete_object(Bucket=test_bucket, Key=key)
|
|
|
|
response = s3_client.list_objects_v2(Bucket=test_bucket)
|
|
assert not any(o["Key"] == key for o in response.get("Contents", []))
|
|
|
|
def test_delete_objects_batch(self, s3_client, test_bucket):
|
|
"""Test DeleteObjects batch deletes multiple objects."""
|
|
for i in range(1, 4):
|
|
s3_client.put_object(
|
|
Bucket=test_bucket, Key=f"batch-{i}.txt", Body=f"batch {i}".encode()
|
|
)
|
|
|
|
response = s3_client.delete_objects(
|
|
Bucket=test_bucket,
|
|
Delete={"Objects": [{"Key": f"batch-{i}.txt"} for i in range(1, 4)]},
|
|
)
|
|
assert len(response.get("Deleted", [])) == 3
|
|
|
|
|
|
class TestS3CopyObject:
|
|
"""Test S3 copy operations."""
|
|
|
|
def test_copy_object_same_bucket(self, s3_client, test_bucket):
|
|
"""Test CopyObject within same bucket."""
|
|
src_key = "src-file.txt"
|
|
dst_key = "dst-file.txt"
|
|
content = b"content to copy"
|
|
|
|
s3_client.put_object(Bucket=test_bucket, Key=src_key, Body=content)
|
|
|
|
response = s3_client.copy_object(
|
|
CopySource={"Bucket": test_bucket, "Key": src_key},
|
|
Bucket=test_bucket,
|
|
Key=dst_key,
|
|
)
|
|
assert response.get("CopyObjectResult")
|
|
|
|
# Verify copy
|
|
get_response = s3_client.get_object(Bucket=test_bucket, Key=dst_key)
|
|
assert get_response["Body"].read() == content
|
|
|
|
def test_copy_object_cross_bucket(self, s3_client, test_bucket, unique_name):
|
|
"""Test CopyObject across buckets."""
|
|
dest_bucket = f"pytest-s3-copy-{unique_name}"
|
|
src_key = "src-file.txt"
|
|
dst_key = "dst-file.txt"
|
|
content = b"content to copy"
|
|
|
|
s3_client.put_object(Bucket=test_bucket, Key=src_key, Body=content)
|
|
s3_client.create_bucket(Bucket=dest_bucket)
|
|
|
|
try:
|
|
response = s3_client.copy_object(
|
|
CopySource={"Bucket": test_bucket, "Key": src_key},
|
|
Bucket=dest_bucket,
|
|
Key=dst_key,
|
|
)
|
|
assert response.get("CopyObjectResult")
|
|
|
|
# Verify copy
|
|
get_response = s3_client.get_object(Bucket=dest_bucket, Key=dst_key)
|
|
assert get_response["Body"].read() == content
|
|
finally:
|
|
s3_client.delete_object(Bucket=dest_bucket, Key=dst_key)
|
|
s3_client.delete_bucket(Bucket=dest_bucket)
|
|
|
|
def test_copy_object_non_ascii_key(self, s3_client, test_bucket):
|
|
"""Test CopyObject with non-ASCII characters in key."""
|
|
src_key = "src/テスト画像.png"
|
|
dst_key = "dst/テスト画像.png"
|
|
content = b"non-ascii content"
|
|
|
|
s3_client.put_object(Bucket=test_bucket, Key=src_key, Body=content)
|
|
|
|
response = s3_client.copy_object(
|
|
CopySource={"Bucket": test_bucket, "Key": src_key},
|
|
Bucket=test_bucket,
|
|
Key=dst_key,
|
|
)
|
|
assert response.get("CopyObjectResult")
|
|
|
|
# Verify copy
|
|
get_response = s3_client.get_object(Bucket=test_bucket, Key=dst_key)
|
|
assert get_response["Body"].read() == content
|
|
|
|
|
|
class TestS3ObjectTagging:
|
|
"""Test S3 object tagging operations."""
|
|
|
|
def test_put_object_tagging(self, s3_client, test_bucket):
|
|
"""Test PutObjectTagging adds tags to object."""
|
|
key = "tagged-file.txt"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=b"content")
|
|
|
|
s3_client.put_object_tagging(
|
|
Bucket=test_bucket,
|
|
Key=key,
|
|
Tagging={
|
|
"TagSet": [
|
|
{"Key": "env", "Value": "test"},
|
|
{"Key": "project", "Value": "floci"},
|
|
]
|
|
},
|
|
)
|
|
# If no exception, test passes
|
|
|
|
def test_get_object_tagging(self, s3_client, test_bucket):
|
|
"""Test GetObjectTagging returns tags."""
|
|
key = "tagged-file.txt"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=b"content")
|
|
s3_client.put_object_tagging(
|
|
Bucket=test_bucket,
|
|
Key=key,
|
|
Tagging={
|
|
"TagSet": [
|
|
{"Key": "env", "Value": "test"},
|
|
{"Key": "project", "Value": "floci"},
|
|
]
|
|
},
|
|
)
|
|
|
|
response = s3_client.get_object_tagging(Bucket=test_bucket, Key=key)
|
|
tags = {t["Key"]: t["Value"] for t in response["TagSet"]}
|
|
assert tags.get("env") == "test"
|
|
assert tags.get("project") == "floci"
|
|
|
|
def test_delete_object_tagging(self, s3_client, test_bucket):
|
|
"""Test DeleteObjectTagging removes tags."""
|
|
key = "tagged-file.txt"
|
|
s3_client.put_object(Bucket=test_bucket, Key=key, Body=b"content")
|
|
s3_client.put_object_tagging(
|
|
Bucket=test_bucket,
|
|
Key=key,
|
|
Tagging={"TagSet": [{"Key": "env", "Value": "test"}]},
|
|
)
|
|
|
|
s3_client.delete_object_tagging(Bucket=test_bucket, Key=key)
|
|
|
|
response = s3_client.get_object_tagging(Bucket=test_bucket, Key=key)
|
|
assert len(response["TagSet"]) == 0
|
|
|
|
|
|
class TestS3BucketTagging:
|
|
"""Test S3 bucket tagging operations."""
|
|
|
|
def test_put_bucket_tagging(self, s3_client, test_bucket):
|
|
"""Test PutBucketTagging adds tags to bucket."""
|
|
s3_client.put_bucket_tagging(
|
|
Bucket=test_bucket,
|
|
Tagging={
|
|
"TagSet": [
|
|
{"Key": "team", "Value": "backend"},
|
|
{"Key": "cost-center", "Value": "123"},
|
|
]
|
|
},
|
|
)
|
|
# If no exception, test passes
|
|
|
|
def test_get_bucket_tagging(self, s3_client, test_bucket):
|
|
"""Test GetBucketTagging returns tags."""
|
|
s3_client.put_bucket_tagging(
|
|
Bucket=test_bucket,
|
|
Tagging={
|
|
"TagSet": [
|
|
{"Key": "team", "Value": "backend"},
|
|
{"Key": "cost-center", "Value": "123"},
|
|
]
|
|
},
|
|
)
|
|
|
|
response = s3_client.get_bucket_tagging(Bucket=test_bucket)
|
|
tags = {t["Key"]: t["Value"] for t in response["TagSet"]}
|
|
assert tags.get("team") == "backend"
|
|
assert tags.get("cost-center") == "123"
|
|
|
|
def test_delete_bucket_tagging(self, s3_client, test_bucket):
|
|
"""Test DeleteBucketTagging removes tags."""
|
|
s3_client.put_bucket_tagging(
|
|
Bucket=test_bucket,
|
|
Tagging={"TagSet": [{"Key": "team", "Value": "backend"}]},
|
|
)
|
|
|
|
s3_client.delete_bucket_tagging(Bucket=test_bucket)
|
|
|
|
# Either empty tags or NoSuchTagSet error
|
|
try:
|
|
response = s3_client.get_bucket_tagging(Bucket=test_bucket)
|
|
assert len(response.get("TagSet", [])) == 0
|
|
except ClientError:
|
|
pass # NoSuchTagSet is also acceptable
|
|
|
|
|
|
class TestS3LargeObject:
|
|
"""Test S3 large object operations."""
|
|
|
|
def test_put_object_25mb(self, s3_client, test_bucket):
|
|
"""Test PutObject with 25 MB payload."""
|
|
key = "large-object-25mb.bin"
|
|
large_payload = b"\x00" * (25 * 1024 * 1024)
|
|
|
|
s3_client.put_object(
|
|
Bucket=test_bucket,
|
|
Key=key,
|
|
Body=large_payload,
|
|
ContentType="application/octet-stream",
|
|
)
|
|
|
|
response = s3_client.head_object(Bucket=test_bucket, Key=key)
|
|
assert response["ContentLength"] == 25 * 1024 * 1024
|
|
|
|
|
|
class TestS3MultipartChecksums:
|
|
"""Multipart uploads carry composite checksums, computed as on AWS.
|
|
|
|
AWS combines the part-level checksums of a multipart upload into a single object-level value:
|
|
the algorithm applied to the concatenated binary part checksums, Base64 encoded, followed by
|
|
``-<part count>``. HeadObject/GetObject return that suffixed value, GetObjectAttributes returns
|
|
it without the suffix. SHA1/SHA256 uploads are always COMPOSITE, and so are CRC32 uploads
|
|
unless FULL_OBJECT is requested on CreateMultipartUpload.
|
|
"""
|
|
|
|
PART_SIZE = 5 * MIB
|
|
TRANSFER_CONFIG = TransferConfig(multipart_threshold=5 * MIB, multipart_chunksize=5 * MIB)
|
|
|
|
@staticmethod
|
|
def _payload(size=12 * MIB):
|
|
return random.Random(20260903).randbytes(size)
|
|
|
|
@staticmethod
|
|
def _parts(payload, part_size):
|
|
return [payload[i:i + part_size] for i in range(0, len(payload), part_size)]
|
|
|
|
@staticmethod
|
|
def _sha256(data):
|
|
return hashlib.sha256(data).digest()
|
|
|
|
@staticmethod
|
|
def _crc32(data):
|
|
return (zlib.crc32(data) & 0xFFFFFFFF).to_bytes(4, "big")
|
|
|
|
@classmethod
|
|
def _composite(cls, digest, parts):
|
|
combined = digest(b"".join(digest(part) for part in parts))
|
|
return base64.b64encode(combined).decode() + f"-{len(parts)}"
|
|
|
|
def _upload(self, s3_client, bucket, key, payload, tmp_path, **extra_args):
|
|
source = tmp_path / key
|
|
source.write_bytes(payload)
|
|
s3_client.upload_file(
|
|
str(source), bucket, key,
|
|
ExtraArgs=extra_args or None,
|
|
Config=self.TRANSFER_CONFIG,
|
|
)
|
|
|
|
def test_multipart_upload_sha256_reports_composite_checksum(self, s3_client, test_bucket, tmp_path):
|
|
"""Multipart upload with SHA256 returns the composite checksum, stable across reads."""
|
|
key = "multipart-sha256.bin"
|
|
payload = self._payload()
|
|
parts = self._parts(payload, self.PART_SIZE)
|
|
composite = self._composite(self._sha256, parts)
|
|
assert composite.endswith("-3")
|
|
|
|
self._upload(s3_client, test_bucket, key, payload, tmp_path, ChecksumAlgorithm="SHA256")
|
|
|
|
head = s3_client.head_object(Bucket=test_bucket, Key=key, ChecksumMode="ENABLED")
|
|
assert head["ChecksumSHA256"] == composite
|
|
assert head["ChecksumType"] == "COMPOSITE"
|
|
assert head["ETag"].endswith('-3"')
|
|
|
|
# The same value on every read: a client comparing checksums must not see drift
|
|
again = s3_client.head_object(Bucket=test_bucket, Key=key, ChecksumMode="ENABLED")
|
|
assert again["ChecksumSHA256"] == composite
|
|
|
|
attributes = s3_client.get_object_attributes(
|
|
Bucket=test_bucket, Key=key, ObjectAttributes=["Checksum", "ObjectParts"]
|
|
)
|
|
assert attributes["Checksum"]["ChecksumSHA256"] == composite.rsplit("-", 1)[0]
|
|
assert attributes["Checksum"]["ChecksumType"] == "COMPOSITE"
|
|
assert attributes["ObjectParts"]["TotalPartsCount"] == 3
|
|
assert [p["ChecksumSHA256"] for p in attributes["ObjectParts"]["Parts"]] == [
|
|
base64.b64encode(self._sha256(part)).decode() for part in parts
|
|
]
|
|
|
|
body = s3_client.get_object(Bucket=test_bucket, Key=key)["Body"].read()
|
|
assert body == payload
|
|
|
|
def test_multipart_upload_default_checksum_is_composite_crc32(self, s3_client, test_bucket, tmp_path):
|
|
"""Without an explicit algorithm the SDK declares CRC32, which AWS stores as COMPOSITE."""
|
|
key = "multipart-default.bin"
|
|
payload = self._payload()
|
|
parts = self._parts(payload, self.PART_SIZE)
|
|
|
|
self._upload(s3_client, test_bucket, key, payload, tmp_path)
|
|
|
|
head = s3_client.head_object(Bucket=test_bucket, Key=key, ChecksumMode="ENABLED")
|
|
assert head["ChecksumType"] == "COMPOSITE"
|
|
assert head["ChecksumCRC32"] == self._composite(self._crc32, parts)
|
|
|
|
attributes = s3_client.get_object_attributes(
|
|
Bucket=test_bucket, Key=key, ObjectAttributes=["Checksum", "ObjectParts"]
|
|
)
|
|
assert attributes["Checksum"]["ChecksumCRC32"] == head["ChecksumCRC32"].rsplit("-", 1)[0]
|
|
assert [p["ChecksumCRC32"] for p in attributes["ObjectParts"]["Parts"]] == [
|
|
base64.b64encode(self._crc32(part)).decode() for part in parts
|
|
]
|
|
|
|
def test_single_part_upload_keeps_full_object_checksum(self, s3_client, test_bucket, tmp_path):
|
|
"""A file below the multipart threshold is a plain PutObject: FULL_OBJECT, no suffix."""
|
|
key = "single-part-sha256.bin"
|
|
payload = self._payload(1 * MIB)
|
|
|
|
self._upload(s3_client, test_bucket, key, payload, tmp_path, ChecksumAlgorithm="SHA256")
|
|
|
|
head = s3_client.head_object(Bucket=test_bucket, Key=key, ChecksumMode="ENABLED")
|
|
assert head["ChecksumSHA256"] == base64.b64encode(self._sha256(payload)).decode()
|
|
assert head["ChecksumType"] == "FULL_OBJECT"
|
|
assert "-" not in head["ETag"]
|
|
|
|
def test_copy_of_multipart_object_is_a_single_full_object(self, s3_client, test_bucket, tmp_path):
|
|
"""CopyObject writes one object: FULL_OBJECT checksum, an MD5 ETag without part count, no ObjectParts."""
|
|
key, copy_key = "multipart-copy-source.bin", "multipart-copy.bin"
|
|
payload = self._payload()
|
|
self._upload(s3_client, test_bucket, key, payload, tmp_path, ChecksumAlgorithm="SHA256")
|
|
source = s3_client.head_object(Bucket=test_bucket, Key=key, ChecksumMode="ENABLED")
|
|
assert source["ChecksumType"] == "COMPOSITE"
|
|
|
|
result = s3_client.copy_object(
|
|
Bucket=test_bucket, Key=copy_key, CopySource={"Bucket": test_bucket, "Key": key}
|
|
)["CopyObjectResult"]
|
|
full = base64.b64encode(self._sha256(payload)).decode()
|
|
assert result["ChecksumSHA256"] == full
|
|
assert result["ChecksumType"] == "FULL_OBJECT"
|
|
assert result["ETag"] == f'"{hashlib.md5(payload).hexdigest()}"'
|
|
assert result["ETag"] != source["ETag"]
|
|
|
|
head = s3_client.head_object(Bucket=test_bucket, Key=copy_key, ChecksumMode="ENABLED")
|
|
assert (head["ChecksumSHA256"], head["ChecksumType"], head["ETag"]) == (full, "FULL_OBJECT", result["ETag"])
|
|
attributes = s3_client.get_object_attributes(
|
|
Bucket=test_bucket, Key=copy_key, ObjectAttributes=["Checksum", "ObjectParts"]
|
|
)
|
|
assert attributes["Checksum"]["ChecksumType"] == "FULL_OBJECT"
|
|
assert "ObjectParts" not in attributes
|
|
|
|
def test_get_object_attributes_etag_has_no_quotes(self, s3_client, test_bucket, tmp_path):
|
|
"""S3 quotes the ETag everywhere except in GetObjectAttributes."""
|
|
key = "etag-quotes.bin"
|
|
self._upload(s3_client, test_bucket, key, self._payload(), tmp_path, ChecksumAlgorithm="SHA256")
|
|
|
|
head_etag = s3_client.head_object(Bucket=test_bucket, Key=key)["ETag"]
|
|
attributes = s3_client.get_object_attributes(Bucket=test_bucket, Key=key, ObjectAttributes=["ETag"])
|
|
|
|
assert head_etag.startswith('"') and head_etag.endswith('-3"')
|
|
assert attributes["ETag"] == head_etag.strip('"')
|
|
|
|
def test_download_with_checksum_mode_works_for_composite_and_full_object(self, s3_client, test_bucket, tmp_path):
|
|
"""botocore validates a full-object checksum on GetObject and skips a composite one (the -N suffix)."""
|
|
payload, small = self._payload(), self._payload(1 * MIB)
|
|
self._upload(s3_client, test_bucket, "download-composite.bin", payload, tmp_path, ChecksumAlgorithm="SHA256")
|
|
self._upload(s3_client, test_bucket, "download-full.bin", small, tmp_path, ChecksumAlgorithm="SHA256")
|
|
|
|
composite = s3_client.get_object(Bucket=test_bucket, Key="download-composite.bin", ChecksumMode="ENABLED")
|
|
assert composite["ChecksumSHA256"].endswith("-3")
|
|
assert composite["ChecksumType"] == "COMPOSITE"
|
|
assert composite["Body"].read() == payload
|
|
|
|
full = s3_client.get_object(Bucket=test_bucket, Key="download-full.bin", ChecksumMode="ENABLED")
|
|
assert full["ChecksumType"] == "FULL_OBJECT"
|
|
assert full["Body"].read() == small
|