catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Enhancement/3821 (#3822)

Browse source 
* Add _does_event_match_pattern() to EventsBackend and use when determining whether to archive an event

* Add comment to _does_event_item_match_pattern_item()

* Expand test case for Archive EventFilter

* Apply black formatting

Co-authored-by: Tom Noble <tom.noble@bjss.com>
This commit is contained in:
Tom Noble 2021-04-01 10:31:10 +01:00 • committed by GitHub
commit 1440709e4c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 76 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

53
tests/test_events/test_events.py
View file

@ -1488,6 +1488,59 @@ def test_archive_event_with_bus_arn():
response["SizeBytes"].should.be.greater_than(0)
@mock_events
def test_event_not_routed_to_archive_when_detail_does_not_match_pattern():
# given
client = boto3.client("events", "eu-central-1")
event_bus_arn = "arn:aws:events:eu-central-1:{}:event-bus/default".format(
ACCOUNT_ID
)
client.create_archive(
ArchiveName="archive-with-dict-filter",
EventSourceArn=event_bus_arn,
EventPattern=json.dumps({"detail": {"foo": ["bar"]}}),
)
client.create_archive(
ArchiveName="archive-with-list-filter",
EventSourceArn=event_bus_arn,
EventPattern=json.dumps({"source": ["foo", "bar"]}),
)
# when
event_matching_detail = {
"Source": "source",
"DetailType": "type",
"Detail": '{"foo": "bar"}',
}
event_not_matching_detail = {
"Source": "source",
"DetailType": "type",
"Detail": '{"foo": "baz"}',
}
event_matching_source_foo = {"Source": "foo", "DetailType": "type", "Detail": "{}"}
event_matching_source_bar = {"Source": "bar", "DetailType": "type", "Detail": "{}"}
event_not_matching_source = {"Source": "baz", "DetailType": "type", "Detail": "{}"}
client.put_events(
Entries=[
event_matching_detail,
event_not_matching_detail,
event_matching_source_foo,
event_matching_source_bar,
event_not_matching_source,
]
)
# then
response = client.describe_archive(ArchiveName="archive-with-dict-filter")
response["EventCount"].should.equal(1)
response["SizeBytes"].should.be.greater_than(0)
response = client.describe_archive(ArchiveName="archive-with-list-filter")
response["EventCount"].should.equal(2)
response["SizeBytes"].should.be.greater_than(0)
@mock_events
def test_start_replay():
# given