Enhancement/3837 (#3847)
* Move event pattern validation into EventPattern class and apply enhanced pattern logic to all Rules * Fix exists filtering logic to only match leaf nodes in event * Apply black formatting * Replace JSONDecodeError with ValueError for Python2 compatibility * Update unit test names * Move event pattern tests into test_event_pattern.py * Apply black formatting Co-authored-by: TSNoble <tom.noble@bjss.com>
This commit is contained in:
Tom Noble
GitHub
parent
5eb99da75a
commit
3942613bf4
4 changed files with 124 additions and 135 deletions
90
tests/test_events/test_event_pattern.py
Normal file
90
tests/test_events/test_event_pattern.py
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from moto.events.models import EventPattern
|
||||
|
||||
|
||||
def test_event_pattern_with_allowed_values_event_filter():
|
||||
pattern = EventPattern(json.dumps({"source": ["foo", "bar"]}))
|
||||
assert pattern.matches_event({"source": "foo"})
|
||||
assert pattern.matches_event({"source": "bar"})
|
||||
assert not pattern.matches_event({"source": "baz"})
|
||||
|
||||
|
||||
def test_event_pattern_with_nested_event_filter():
|
||||
pattern = EventPattern(json.dumps({"detail": {"foo": ["bar"]}}))
|
||||
assert pattern.matches_event({"detail": {"foo": "bar"}})
|
||||
assert not pattern.matches_event({"detail": {"foo": "baz"}})
|
||||
|
||||
|
||||
def test_event_pattern_with_exists_event_filter():
|
||||
foo_exists = EventPattern(json.dumps({"detail": {"foo": [{"exists": True}]}}))
|
||||
assert foo_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert not foo_exists.matches_event({"detail": {}})
|
||||
# exists filters only match leaf nodes of an event
|
||||
assert not foo_exists.matches_event({"detail": {"foo": {"bar": "baz"}}})
|
||||
|
||||
foo_not_exists = EventPattern(json.dumps({"detail": {"foo": [{"exists": False}]}}))
|
||||
assert not foo_not_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert foo_not_exists.matches_event({"detail": {}})
|
||||
assert foo_not_exists.matches_event({"detail": {"foo": {"bar": "baz"}}})
|
||||
|
||||
bar_exists = EventPattern(json.dumps({"detail": {"bar": [{"exists": True}]}}))
|
||||
assert not bar_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert not bar_exists.matches_event({"detail": {}})
|
||||
|
||||
bar_not_exists = EventPattern(json.dumps({"detail": {"bar": [{"exists": False}]}}))
|
||||
assert bar_not_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert bar_not_exists.matches_event({"detail": {}})
|
||||
|
||||
|
||||
def test_event_pattern_with_prefix_event_filter():
|
||||
pattern = EventPattern(json.dumps({"detail": {"foo": [{"prefix": "bar"}]}}))
|
||||
assert pattern.matches_event({"detail": {"foo": "bar"}})
|
||||
assert pattern.matches_event({"detail": {"foo": "bar!"}})
|
||||
assert not pattern.matches_event({"detail": {"foo": "ba"}})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"operator, compare_to, should_match, should_not_match",
|
||||
[
|
||||
("<", 1, [0], [1, 2]),
|
||||
("<=", 1, [0, 1], [2]),
|
||||
("=", 1, [1], [0, 2]),
|
||||
(">", 1, [2], [0, 1]),
|
||||
(">=", 1, [1, 2], [0]),
|
||||
],
|
||||
)
|
||||
def test_event_pattern_with_single_numeric_event_filter(
|
||||
operator, compare_to, should_match, should_not_match
|
||||
):
|
||||
pattern = EventPattern(
|
||||
json.dumps({"detail": {"foo": [{"numeric": [operator, compare_to]}]}})
|
||||
)
|
||||
for number in should_match:
|
||||
assert pattern.matches_event({"detail": {"foo": number}})
|
||||
for number in should_not_match:
|
||||
assert not pattern.matches_event({"detail": {"foo": number}})
|
||||
|
||||
|
||||
def test_event_pattern_with_multi_numeric_event_filter():
|
||||
events = [{"detail": {"foo": number}} for number in range(5)]
|
||||
|
||||
one_or_two = EventPattern(
|
||||
json.dumps({"detail": {"foo": [{"numeric": [">=", 1, "<", 3]}]}})
|
||||
)
|
||||
assert not one_or_two.matches_event(events[0])
|
||||
assert one_or_two.matches_event(events[1])
|
||||
assert one_or_two.matches_event(events[2])
|
||||
assert not one_or_two.matches_event(events[3])
|
||||
assert not one_or_two.matches_event(events[4])
|
||||
|
||||
two_or_three = EventPattern(
|
||||
json.dumps({"detail": {"foo": [{"numeric": [">", 1, "<=", 3]}]}})
|
||||
)
|
||||
assert not two_or_three.matches_event(events[0])
|
||||
assert not two_or_three.matches_event(events[1])
|
||||
assert two_or_three.matches_event(events[2])
|
||||
assert two_or_three.matches_event(events[3])
|
||||
assert not two_or_three.matches_event(events[4])
|
||||
|
|
@ -14,7 +14,6 @@ from moto import mock_logs
|
|||
from moto.core import ACCOUNT_ID
|
||||
from moto.core.utils import iso_8601_datetime_without_milliseconds
|
||||
from moto.events import mock_events
|
||||
from moto.events.models import EventPattern
|
||||
|
||||
RULES = [
|
||||
{"Name": "test1", "ScheduleExpression": "rate(5 minutes)"},
|
||||
|
|
@ -1518,88 +1517,6 @@ def test_archive_event_with_bus_arn():
|
|||
response["SizeBytes"].should.be.greater_than(0)
|
||||
|
||||
|
||||
def test_archive_with_allowed_values_event_filter():
|
||||
pattern = EventPattern(json.dumps({"source": ["foo", "bar"]}))
|
||||
assert pattern.matches_event({"source": "foo"})
|
||||
assert pattern.matches_event({"source": "bar"})
|
||||
assert not pattern.matches_event({"source": "baz"})
|
||||
|
||||
|
||||
def test_archive_with_nested_event_filter():
|
||||
pattern = EventPattern(json.dumps({"detail": {"foo": ["bar"]}}))
|
||||
assert pattern.matches_event({"detail": {"foo": "bar"}})
|
||||
assert not pattern.matches_event({"detail": {"foo": "baz"}})
|
||||
|
||||
|
||||
def test_archive_with_exists_event_filter():
|
||||
foo_exists = EventPattern(json.dumps({"detail": {"foo": [{"exists": True}]}}))
|
||||
assert foo_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert not foo_exists.matches_event({"detail": {}})
|
||||
|
||||
foo_not_exists = EventPattern(json.dumps({"detail": {"foo": [{"exists": False}]}}))
|
||||
assert not foo_not_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert foo_not_exists.matches_event({"detail": {}})
|
||||
|
||||
bar_exists = EventPattern(json.dumps({"detail": {"bar": [{"exists": True}]}}))
|
||||
assert not bar_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert not bar_exists.matches_event({"detail": {}})
|
||||
|
||||
bar_not_exists = EventPattern(json.dumps({"detail": {"bar": [{"exists": False}]}}))
|
||||
assert bar_not_exists.matches_event({"detail": {"foo": "bar"}})
|
||||
assert bar_not_exists.matches_event({"detail": {}})
|
||||
|
||||
|
||||
def test_archive_with_prefix_event_filter():
|
||||
pattern = EventPattern(json.dumps({"detail": {"foo": [{"prefix": "bar"}]}}))
|
||||
assert pattern.matches_event({"detail": {"foo": "bar"}})
|
||||
assert pattern.matches_event({"detail": {"foo": "bar!"}})
|
||||
assert not pattern.matches_event({"detail": {"foo": "ba"}})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"operator, compare_to, should_match, should_not_match",
|
||||
[
|
||||
("<", 1, [0], [1, 2]),
|
||||
("<=", 1, [0, 1], [2]),
|
||||
("=", 1, [1], [0, 2]),
|
||||
(">", 1, [2], [0, 1]),
|
||||
(">=", 1, [1, 2], [0]),
|
||||
],
|
||||
)
|
||||
def test_archive_with_single_numeric_event_filter(
|
||||
operator, compare_to, should_match, should_not_match
|
||||
):
|
||||
pattern = EventPattern(
|
||||
json.dumps({"detail": {"foo": [{"numeric": [operator, compare_to]}]}})
|
||||
)
|
||||
for number in should_match:
|
||||
assert pattern.matches_event({"detail": {"foo": number}})
|
||||
for number in should_not_match:
|
||||
assert not pattern.matches_event({"detail": {"foo": number}})
|
||||
|
||||
|
||||
def test_archive_with_multi_numeric_event_filter():
|
||||
events = [{"detail": {"foo": number}} for number in range(5)]
|
||||
|
||||
one_or_two = EventPattern(
|
||||
json.dumps({"detail": {"foo": [{"numeric": [">=", 1, "<", 3]}]}})
|
||||
)
|
||||
assert not one_or_two.matches_event(events[0])
|
||||
assert one_or_two.matches_event(events[1])
|
||||
assert one_or_two.matches_event(events[2])
|
||||
assert not one_or_two.matches_event(events[3])
|
||||
assert not one_or_two.matches_event(events[4])
|
||||
|
||||
two_or_three = EventPattern(
|
||||
json.dumps({"detail": {"foo": [{"numeric": [">", 1, "<=", 3]}]}})
|
||||
)
|
||||
assert not two_or_three.matches_event(events[0])
|
||||
assert not two_or_three.matches_event(events[1])
|
||||
assert two_or_three.matches_event(events[2])
|
||||
assert two_or_three.matches_event(events[3])
|
||||
assert not two_or_three.matches_event(events[4])
|
||||
|
||||
|
||||
@mock_events
|
||||
def test_start_replay():
|
||||
# given
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue