Skip to content

Commit 20ac8bc

Browse files
committed
test: add comprehensive tests for S3 IntelligentTiering events
Relates to #7443 Add test event and comprehensive test coverage for both parser models and data classes handling of S3 IntelligentTiering events. Changes: - Add s3EventIntelligentTiering.json test event - Add parser model tests (2 tests) - Add data classes tests (3 tests) - Tests verify get_object field handling and intelligentTieringEventData
1 parent 8334b2b commit 20ac8bc

File tree

3 files changed

+182
-0
lines changed

3 files changed

+182
-0
lines changed
Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
{
2+
"Records": [
3+
{
4+
"eventVersion": "2.3",
5+
"eventSource": "aws:s3",
6+
"awsRegion": "ap-southeast-2",
7+
"eventTime": "2025-09-29T00:47:23.967Z",
8+
"eventName": "IntelligentTiering",
9+
"userIdentity": {
10+
"principalId": "s3.amazonaws.com"
11+
},
12+
"requestParameters": {
13+
"sourceIPAddress": "s3.amazonaws.com"
14+
},
15+
"responseElements": {
16+
"x-amz-request-id": "4C05BE601E268FCD",
17+
"x-amz-id-2": "Wt0dgzBP/wQ47PWP5BqoD4nAZC08NOjfdZj2prx5gVPBCI+YQtrx/UKVlnT63Dj+vse4x5koWp61HaoVyz6fUFe1yLkhNpK3"
18+
},
19+
"s3": {
20+
"s3SchemaVersion": "1.0",
21+
"configurationId": "ZDM3MGVjY2MtZTlmNC00ZTk0LThiNjItMWJiNTMwOTYwY2Rl",
22+
"bucket": {
23+
"name": "mybucket",
24+
"ownerIdentity": {
25+
"principalId": "ANY597T3BWFY2"
26+
},
27+
"arn": "arn:aws:s3:::mybucket"
28+
},
29+
"get_object": {
30+
"key": "myobject",
31+
"size": 252294,
32+
"eTag": "4e9270240d7d62d5ee8dbfcb7a7a3279",
33+
"versionId": "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn",
34+
"sequencer": "0066A8D0E77DE42BC5"
35+
}
36+
},
37+
"intelligentTieringEventData": {
38+
"destinationAccessTier": "ARCHIVE_ACCESS"
39+
}
40+
}
41+
]
42+
}
Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
from aws_lambda_powertools.utilities.data_classes import S3Event
2+
from tests.functional.utils import load_event
3+
4+
5+
def test_s3_intelligent_tiering_event():
6+
"""Test S3 IntelligentTiering event with get_object field"""
7+
event = S3Event(load_event("s3EventIntelligentTiering.json"))
8+
9+
# Test first record
10+
record = event.record
11+
assert record.event_name == "IntelligentTiering"
12+
assert record.event_version == "2.3"
13+
assert record.event_source == "aws:s3"
14+
assert record.aws_region == "ap-southeast-2"
15+
16+
# Test user identity
17+
assert record.user_identity.principal_id == "s3.amazonaws.com"
18+
19+
# Test S3 object via get_object property (handles both 'object' and 'get_object' keys)
20+
s3_object = record.s3.get_object
21+
assert s3_object.key == "myobject"
22+
assert s3_object.size == 252294
23+
assert s3_object.etag == "4e9270240d7d62d5ee8dbfcb7a7a3279"
24+
assert s3_object.version_id == "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn"
25+
assert s3_object.sequencer == "0066A8D0E77DE42BC5"
26+
27+
# Test bucket
28+
assert record.s3.bucket.name == "mybucket"
29+
assert record.s3.bucket.arn == "arn:aws:s3:::mybucket"
30+
31+
# Test intelligentTieringEventData
32+
assert record.intelligent_tiering_event_data is not None
33+
assert record.intelligent_tiering_event_data.destination_access_tier == "ARCHIVE_ACCESS"
34+
35+
# Verify glacierEventData is None
36+
assert record.glacier_event_data is None
37+
38+
# Test convenience properties
39+
assert event.bucket_name == "mybucket"
40+
assert event.object_key == "myobject"
41+
42+
43+
def test_s3_intelligent_tiering_event_iteration():
44+
"""Test iterating through multiple IntelligentTiering records"""
45+
event = S3Event(load_event("s3EventIntelligentTiering.json"))
46+
47+
records = list(event.records)
48+
assert len(records) == 1
49+
50+
for record in event.records:
51+
assert record.event_name == "IntelligentTiering"
52+
assert record.s3.get_object.key == "myobject"
53+
assert record.intelligent_tiering_event_data.destination_access_tier == "ARCHIVE_ACCESS"
54+
55+
56+
def test_s3_intelligent_tiering_deep_archive_access():
57+
"""Test IntelligentTiering event with DEEP_ARCHIVE_ACCESS tier"""
58+
raw_event = load_event("s3EventIntelligentTiering.json")
59+
raw_event["Records"][0]["intelligentTieringEventData"]["destinationAccessTier"] = "DEEP_ARCHIVE_ACCESS"
60+
61+
event = S3Event(raw_event)
62+
record = event.record
63+
64+
assert record.intelligent_tiering_event_data.destination_access_tier == "DEEP_ARCHIVE_ACCESS"
Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
from aws_lambda_powertools.utilities.parser.models import S3Model, S3RecordModel
2+
from tests.functional.utils import load_event
3+
4+
5+
def test_s3_intelligent_tiering_event():
6+
"""Test parsing of S3 IntelligentTiering events with get_object field"""
7+
raw_event = load_event("s3EventIntelligentTiering.json")
8+
parsed_event: S3Model = S3Model(**raw_event)
9+
10+
records = list(parsed_event.Records)
11+
assert len(records) == 1
12+
13+
record: S3RecordModel = records[0]
14+
raw_record = raw_event["Records"][0]
15+
16+
# Verify basic event properties
17+
assert record.eventVersion == "2.3"
18+
assert record.eventSource == "aws:s3"
19+
assert record.awsRegion == "ap-southeast-2"
20+
assert record.eventName == "IntelligentTiering"
21+
22+
# Verify user identity
23+
user_identity = record.userIdentity
24+
assert user_identity.principalId == "s3.amazonaws.com"
25+
26+
# Verify request parameters
27+
request_parameters = record.requestParameters
28+
# Note: sourceIPAddress is "s3.amazonaws.com" for IntelligentTiering events, not an IP
29+
assert str(request_parameters.sourceIPAddress) == "s3.amazonaws.com"
30+
31+
# Verify response elements
32+
assert record.responseElements.x_amz_request_id == raw_record["responseElements"]["x-amz-request-id"]
33+
assert record.responseElements.x_amz_id_2 == raw_record["responseElements"]["x-amz-id-2"]
34+
35+
# Verify S3 message
36+
s3 = record.s3
37+
assert s3.s3SchemaVersion == raw_record["s3"]["s3SchemaVersion"]
38+
assert s3.configurationId == raw_record["s3"]["configurationId"]
39+
40+
# Verify bucket
41+
bucket = s3.bucket
42+
raw_bucket = raw_record["s3"]["bucket"]
43+
assert bucket.name == "mybucket"
44+
assert bucket.ownerIdentity.principalId == raw_bucket["ownerIdentity"]["principalId"]
45+
assert bucket.arn == "arn:aws:s3:::mybucket"
46+
47+
# Verify get_object field (IntelligentTiering uses 'get_object' instead of 'object')
48+
assert s3.get_object is not None
49+
assert s3.get_object.key == "myobject"
50+
assert s3.get_object.size == 252294
51+
assert s3.get_object.eTag == "4e9270240d7d62d5ee8dbfcb7a7a3279"
52+
assert s3.get_object.versionId == "tiogA9Ga7Xi49yfJ6lkeTxPYx7ZK75yn"
53+
assert s3.get_object.sequencer == "0066A8D0E77DE42BC5"
54+
55+
# Verify intelligentTieringEventData
56+
assert record.intelligentTieringEventData is not None
57+
assert record.intelligentTieringEventData.destinationAccessTier == "ARCHIVE_ACCESS"
58+
59+
# Verify glacierEventData is None for IntelligentTiering events
60+
assert record.glacierEventData is None
61+
62+
63+
def test_s3_intelligent_tiering_event_access_tiers():
64+
"""Test different access tier values for IntelligentTiering events"""
65+
raw_event = load_event("s3EventIntelligentTiering.json")
66+
67+
# Test ARCHIVE_ACCESS tier (from the test event)
68+
parsed_event: S3Model = S3Model(**raw_event)
69+
record = list(parsed_event.Records)[0]
70+
assert record.intelligentTieringEventData.destinationAccessTier == "ARCHIVE_ACCESS"
71+
72+
# Test DEEP_ARCHIVE_ACCESS tier
73+
raw_event["Records"][0]["intelligentTieringEventData"]["destinationAccessTier"] = "DEEP_ARCHIVE_ACCESS"
74+
parsed_event: S3Model = S3Model(**raw_event)
75+
record = list(parsed_event.Records)[0]
76+
assert record.intelligentTieringEventData.destinationAccessTier == "DEEP_ARCHIVE_ACCESS"

0 commit comments

Comments
 (0)