forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_eventbridge.py
More file actions
53 lines (40 loc) · 2.03 KB
/
test_eventbridge.py
File metadata and controls
53 lines (40 loc) · 2.03 KB
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
import pytest
from aws_lambda_powertools.utilities.parser import ValidationError, envelopes, parse
from tests.functional.utils import load_event
from tests.unit.parser.schemas import (
MyAdvancedEventbridgeBusiness,
MyEventbridgeBusiness,
)
def test_handle_eventbridge_trigger_event():
raw_event = load_event("eventBridgeEvent.json")
parsed_event: MyEventbridgeBusiness = parse(
event=raw_event,
model=MyEventbridgeBusiness,
envelope=envelopes.EventBridgeEnvelope,
)
assert parsed_event.instance_id == raw_event["detail"]["instance_id"]
assert parsed_event.state == raw_event["detail"]["state"]
def test_validate_event_does_not_conform_with_user_dict_model():
raw_event = load_event("eventBridgeEvent.json")
raw_event.pop("version")
with pytest.raises(ValidationError):
parse(event=raw_event, model=MyEventbridgeBusiness, envelope=envelopes.EventBridgeEnvelope)
def test_handle_eventbridge_trigger_event_no_envelope():
raw_event = load_event("eventBridgeEvent.json")
parsed_event: MyAdvancedEventbridgeBusiness = MyAdvancedEventbridgeBusiness(**raw_event)
assert parsed_event.detail.instance_id == raw_event["detail"]["instance_id"]
assert parsed_event.detail.state == raw_event["detail"]["state"]
assert parsed_event.id == raw_event["id"]
assert parsed_event.version == raw_event["version"]
assert parsed_event.account == raw_event["account"]
time_str = parsed_event.time.strftime("%Y-%m-%dT%H:%M:%SZ")
assert time_str == raw_event["time"]
assert parsed_event.region == raw_event["region"]
assert parsed_event.resources == raw_event["resources"]
assert parsed_event.source == raw_event["source"]
assert parsed_event.detail_type == raw_event["detail-type"]
assert parsed_event.replay_name == raw_event["replay-name"]
def test_handle_invalid_event_with_eventbridge_envelope():
empty_event = {}
with pytest.raises(ValidationError):
parse(event=empty_event, model=MyEventbridgeBusiness, envelope=envelopes.EventBridgeEnvelope)