forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_s3_notification.py
145 lines (116 loc) · 7.01 KB
/
test_s3_notification.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import json
from datetime import datetime
import pytest
from aws_lambda_powertools.utilities.parser import ValidationError
from aws_lambda_powertools.utilities.parser.models import (
S3EventNotificationEventBridgeModel,
S3SqsEventNotificationModel,
)
from tests.functional.utils import load_event
def test_s3_eventbridge_notification_object_created_event():
raw_event = load_event("s3EventBridgeNotificationObjectCreatedEvent.json")
model = S3EventNotificationEventBridgeModel(**raw_event)
assert model.version == raw_event["version"]
assert model.id == raw_event["id"]
assert model.detail_type == raw_event["detail-type"]
assert model.source == raw_event["source"]
assert model.account == raw_event["account"]
assert model.time == datetime.fromisoformat(raw_event["time"].replace("Z", "+00:00"))
assert model.region == raw_event["region"]
assert model.resources == raw_event["resources"]
assert model.detail.version == raw_event["detail"]["version"]
assert model.detail.bucket.name == raw_event["detail"]["bucket"]["name"]
assert model.detail.object.key == raw_event["detail"]["object"]["key"]
assert model.detail.object.size == raw_event["detail"]["object"]["size"]
assert model.detail.object.etag == raw_event["detail"]["object"]["etag"]
assert model.detail.object.sequencer == raw_event["detail"]["object"]["sequencer"]
assert model.detail.request_id == raw_event["detail"]["request-id"]
assert model.detail.requester == raw_event["detail"]["requester"]
assert model.detail.source_ip_address == raw_event["detail"]["source-ip-address"]
assert model.detail.reason == raw_event["detail"]["reason"]
def test_s3_eventbridge_notification_object_deleted_event():
raw_event = load_event("s3EventBridgeNotificationObjectDeletedEvent.json")
model = S3EventNotificationEventBridgeModel(**raw_event)
assert model.version == raw_event["version"]
assert model.id == raw_event["id"]
assert model.detail_type == raw_event["detail-type"]
assert model.source == raw_event["source"]
assert model.account == raw_event["account"]
assert model.time == datetime.fromisoformat(raw_event["time"].replace("Z", "+00:00"))
assert model.region == raw_event["region"]
assert model.resources == raw_event["resources"]
assert model.detail.version == raw_event["detail"]["version"]
assert model.detail.bucket.name == raw_event["detail"]["bucket"]["name"]
assert model.detail.object.key == raw_event["detail"]["object"]["key"]
assert model.detail.object.size == raw_event["detail"]["object"].get("size")
assert model.detail.object.etag == raw_event["detail"]["object"].get("etag", "")
assert model.detail.object.sequencer == raw_event["detail"]["object"]["sequencer"]
assert model.detail.request_id == raw_event["detail"]["request-id"]
assert model.detail.requester == raw_event["detail"]["requester"]
assert model.detail.source_ip_address == raw_event["detail"]["source-ip-address"]
assert model.detail.reason == raw_event["detail"]["reason"]
assert model.detail.deletion_type == raw_event["detail"]["deletion-type"]
def test_s3_eventbridge_notification_object_expired_event():
raw_event = load_event("s3EventBridgeNotificationObjectExpiredEvent.json")
model = S3EventNotificationEventBridgeModel(**raw_event)
assert model.version == raw_event["version"]
assert model.id == raw_event["id"]
assert model.detail_type == raw_event["detail-type"]
assert model.source == raw_event["source"]
assert model.account == raw_event["account"]
assert model.time == datetime.fromisoformat(raw_event["time"].replace("Z", "+00:00"))
assert model.region == raw_event["region"]
assert model.resources == raw_event["resources"]
assert model.detail.version == raw_event["detail"]["version"]
assert model.detail.bucket.name == raw_event["detail"]["bucket"]["name"]
assert model.detail.object.key == raw_event["detail"]["object"]["key"]
assert model.detail.object.size == raw_event["detail"]["object"]["size"]
assert model.detail.object.etag == raw_event["detail"]["object"]["etag"]
assert model.detail.object.sequencer == raw_event["detail"]["object"]["sequencer"]
assert model.detail.request_id == raw_event["detail"]["request-id"]
assert model.detail.requester == raw_event["detail"]["requester"]
assert model.detail.reason == raw_event["detail"]["reason"]
assert model.detail.deletion_type == raw_event["detail"]["deletion-type"]
def test_s3_eventbridge_notification_object_restore_completed_event():
raw_event = load_event("s3EventBridgeNotificationObjectRestoreCompletedEvent.json")
model = S3EventNotificationEventBridgeModel(**raw_event)
assert model.version == raw_event["version"]
assert model.id == raw_event["id"]
assert model.detail_type == raw_event["detail-type"]
assert model.source == raw_event["source"]
assert model.account == raw_event["account"]
assert model.time == datetime.fromisoformat(raw_event["time"].replace("Z", "+00:00"))
assert model.region == raw_event["region"]
assert model.resources == raw_event["resources"]
assert model.detail.version == raw_event["detail"]["version"]
assert model.detail.bucket.name == raw_event["detail"]["bucket"]["name"]
assert model.detail.object.key == raw_event["detail"]["object"]["key"]
assert model.detail.object.size == raw_event["detail"]["object"]["size"]
assert model.detail.object.etag == raw_event["detail"]["object"]["etag"]
assert model.detail.object.sequencer == raw_event["detail"]["object"]["sequencer"]
assert model.detail.request_id == raw_event["detail"]["request-id"]
assert model.detail.requester == raw_event["detail"]["requester"]
assert model.detail.restore_expiry_time == raw_event["detail"]["restore-expiry-time"]
assert model.detail.source_storage_class == raw_event["detail"]["source-storage-class"]
def test_s3_sqs_event_notification():
raw_event = load_event("s3SqsEvent.json")
model = S3SqsEventNotificationModel(**raw_event)
body = json.loads(raw_event["Records"][0]["body"])
assert model.Records[0].body.Records[0].eventVersion == body["Records"][0]["eventVersion"]
assert model.Records[0].body.Records[0].eventSource == body["Records"][0]["eventSource"]
assert model.Records[0].body.Records[0].eventTime == datetime.fromisoformat(
body["Records"][0]["eventTime"].replace("Z", "+00:00"),
)
assert model.Records[0].body.Records[0].eventName == body["Records"][0]["eventName"]
def test_s3_sqs_event_notification_body_invalid_json():
raw_event = load_event("s3SqsEvent.json")
for record in raw_event["Records"]:
record["body"] = "invalid body"
with pytest.raises(ValidationError):
S3SqsEventNotificationModel(**raw_event)
def test_s3_sqs_event_notification_body_containing_arbitrary_json():
raw_event = load_event("s3SqsEvent.json")
for record in raw_event["Records"]:
record["body"] = {"foo": "bar"}
with pytest.raises(ValidationError):
S3SqsEventNotificationModel(**raw_event)