Skip to content

Commit

Permalink
Enhance alert filtering tests in by adding comprehensive scenarios fo…
Browse files Browse the repository at this point in the history
…r 'IS_NOT' filter types across tags, owners, models, and statuses.
  • Loading branch information
MikaKerman committed Dec 23, 2024
1 parent 4f25fa5 commit 255bdb4
Showing 1 changed file with 228 additions and 1 deletion.
229 changes: 228 additions & 1 deletion tests/unit/monitor/api/alerts/test_alert_filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -340,7 +340,9 @@ def initial_alerts():
def test_filter_alerts_by_tags():
test_alerts, model_alerts, _ = initial_alerts()

filter = FiltersSchema(tags=[FilterSchema(values=["one"], type=FilterType.IS)])
filter = FiltersSchema(
tags=[FilterSchema(values=["one"], type=FilterType.IS)], statuses=[]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
Expand All @@ -349,6 +351,18 @@ def test_filter_alerts_by_tags():
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_1"

filter = FiltersSchema(
tags=[FilterSchema(values=["one"], type=FilterType.IS_NOT)], statuses=[]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_2"
assert filter_test_alerts[1].id == "test_alert_4"
assert len(filter_model_alerts) == 2
assert filter_model_alerts[0].id == "model_alert_2"
assert filter_model_alerts[1].id == "model_alert_3"

filter = FiltersSchema(
tags=[FilterSchema(values=["three"], type=FilterType.IS)], statuses=[]
)
Expand All @@ -361,6 +375,17 @@ def test_filter_alerts_by_tags():
assert filter_model_alerts[0].id == "model_alert_2"
assert filter_model_alerts[1].id == "model_alert_3"

filter = FiltersSchema(
tags=[FilterSchema(values=["three"], type=FilterType.IS_NOT)], statuses=[]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_1"
assert filter_test_alerts[1].id == "test_alert_3"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_1"

filter = FiltersSchema(
tags=[FilterSchema(values=["four"], type=FilterType.IS)], statuses=[]
)
Expand All @@ -371,6 +396,23 @@ def test_filter_alerts_by_tags():
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_3"

filter = FiltersSchema(
tags=[FilterSchema(values=["four"], type=FilterType.IS_NOT)], statuses=[]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 3
assert sorted([alert.id for alert in filter_test_alerts]) == [
"test_alert_1",
"test_alert_2",
"test_alert_3",
]
assert len(filter_model_alerts) == 2
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_1",
"model_alert_2",
]

filter = FiltersSchema(
tags=[
FilterSchema(values=["one"], type=FilterType.IS),
Expand All @@ -385,6 +427,26 @@ def test_filter_alerts_by_tags():
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_1"

filter = FiltersSchema(
tags=[
FilterSchema(values=["one"], type=FilterType.IS_NOT),
FilterSchema(values=["two"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert sorted([alert.id for alert in filter_test_alerts]) == [
"test_alert_2",
"test_alert_4",
]
assert len(filter_model_alerts) == 2
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_2",
"model_alert_3",
]

filter = FiltersSchema(
tags=[
FilterSchema(values=["one"], type=FilterType.IS),
Expand All @@ -397,6 +459,20 @@ def test_filter_alerts_by_tags():
assert len(filter_test_alerts) == 0
assert len(filter_model_alerts) == 0

filter = FiltersSchema(
tags=[
FilterSchema(values=["one"], type=FilterType.IS_NOT),
FilterSchema(values=["four"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_2"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_2"

filter = FiltersSchema(
tags=[
FilterSchema(values=["one", "four"], type=FilterType.IS),
Expand All @@ -417,6 +493,53 @@ def test_filter_alerts_by_tags():
"model_alert_3",
]

filter = FiltersSchema(
tags=[
FilterSchema(values=["one", "four"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_2"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_2"

filter = FiltersSchema(
tags=[
FilterSchema(values=["one"], type=FilterType.IS),
FilterSchema(values=["three"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert sorted([alert.id for alert in filter_test_alerts]) == [
"test_alert_1",
"test_alert_3",
]
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_1"

filter = FiltersSchema(
tags=[
FilterSchema(values=["one", "two"], type=FilterType.IS),
FilterSchema(values=["three", "four"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert sorted([alert.id for alert in filter_test_alerts]) == [
"test_alert_1",
"test_alert_3",
]
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_1"


def test_filter_alerts_by_owners():
test_alerts, model_alerts, _ = initial_alerts()
Expand All @@ -434,6 +557,16 @@ def test_filter_alerts_by_owners():
assert filter_model_alerts[0].id == "model_alert_1"
assert filter_model_alerts[1].id == "model_alert_3"

filter = FiltersSchema(
owners=[FilterSchema(values=["jeff"], type=FilterType.IS_NOT)], statuses=[]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_3"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_2"

filter = FiltersSchema(
owners=[FilterSchema(values=["john"], type=FilterType.IS)], statuses=[]
)
Expand All @@ -447,6 +580,53 @@ def test_filter_alerts_by_owners():
assert filter_model_alerts[0].id == "model_alert_1"
assert filter_model_alerts[1].id == "model_alert_2"

filter = FiltersSchema(
owners=[FilterSchema(values=["john"], type=FilterType.IS_NOT)], statuses=[]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_4"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_3"

filter = FiltersSchema(
owners=[
FilterSchema(values=["jeff"], type=FilterType.IS),
FilterSchema(values=["john"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 1
assert filter_test_alerts[0].id == "test_alert_4"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_3"

filter = FiltersSchema(
owners=[
FilterSchema(values=["jeff", "john"], type=FilterType.IS),
FilterSchema(values=["fake"], type=FilterType.IS_NOT),
],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 4
assert sorted([alert.id for alert in filter_test_alerts]) == [
"test_alert_1",
"test_alert_2",
"test_alert_3",
"test_alert_4",
]
assert len(filter_model_alerts) == 3
assert sorted([alert.id for alert in filter_model_alerts]) == [
"model_alert_1",
"model_alert_2",
"model_alert_3",
]


def test_filter_alerts_by_model():
test_alerts, model_alerts, _ = initial_alerts()
Expand All @@ -463,6 +643,18 @@ def test_filter_alerts_by_model():
assert filter_model_alerts[0].id == "model_alert_1"
assert filter_model_alerts[1].id == "model_alert_2"

filter = FiltersSchema(
models=[FilterSchema(values=["model_id_1"], type=FilterType.IS_NOT)],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_3"
assert filter_test_alerts[1].id == "test_alert_4"
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_3"

filter = FiltersSchema(
models=[FilterSchema(values=["model_id_2"], type=FilterType.IS)], statuses=[]
)
Expand All @@ -474,6 +666,19 @@ def test_filter_alerts_by_model():
assert len(filter_model_alerts) == 1
assert filter_model_alerts[0].id == "model_alert_3"

filter = FiltersSchema(
models=[FilterSchema(values=["model_id_2"], type=FilterType.IS_NOT)],
statuses=[],
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 2
assert filter_test_alerts[0].id == "test_alert_1"
assert filter_test_alerts[1].id == "test_alert_2"
assert len(filter_model_alerts) == 2
assert filter_model_alerts[0].id == "model_alert_1"
assert filter_model_alerts[1].id == "model_alert_2"

filter = FiltersSchema(
models=[FilterSchema(values=["model_id_1", "model_id_2"], type=FilterType.IS)],
statuses=[],
Expand Down Expand Up @@ -574,6 +779,19 @@ def test_filter_alerts_by_statuses():
assert len(filter_model_alerts) == 0
assert len(filter_source_freshness_alerts) == 2

filter = FiltersSchema(
statuses=[
StatusFilterSchema(
values=[Status.FAIL, Status.WARN, Status.ERROR],
type=FilterType.IS_NOT,
)
]
)
filter_test_alerts = filter_alerts(test_alerts, filter)
filter_model_alerts = filter_alerts(model_alerts, filter)
assert len(filter_test_alerts) == 0
assert len(filter_model_alerts) == 1


def test_filter_alerts_by_resource_types():
test_alerts, model_alerts, _ = initial_alerts()
Expand All @@ -597,6 +815,15 @@ def test_filter_alerts_by_resource_types():
filter_test_alerts = filter_alerts(all_alerts, filter)
assert filter_test_alerts == model_alerts

filter = FiltersSchema(
resource_types=[
ResourceTypeFilterSchema(values=[ResourceType.TEST], type=FilterType.IS_NOT)
],
statuses=[],
)
filter_test_alerts = filter_alerts(all_alerts, filter)
assert filter_test_alerts == model_alerts


def test_filter_alerts():
test_alerts, model_alerts, _ = initial_alerts()
Expand Down

0 comments on commit 255bdb4

Please sign in to comment.