12
12
from simvue .exception import ObjectNotFoundError
13
13
import simvue .run as sv_run
14
14
import simvue .api .objects as sv_api_obj
15
-
15
+ from simvue . api . objects . alert . base import AlertBase
16
16
17
17
@pytest .mark .dependency
18
18
@pytest .mark .client
@@ -24,25 +24,60 @@ def test_get_events(create_test_run: tuple[sv_run.Run, dict]) -> None:
24
24
@pytest .mark .dependency
25
25
@pytest .mark .client
26
26
@pytest .mark .parametrize (
27
- "from_run" , (True , False )
27
+ "from_run" , (True , False ), ids = ( "from_run" , "all_runs" )
28
28
)
29
- def test_get_alerts (create_test_run : tuple [sv_run .Run , dict ], from_run : bool ) -> None :
30
- time .sleep (1.0 )
29
+ @pytest .mark .parametrize (
30
+ "names_only" , (True , False ), ids = ("names_only" , "all_details" )
31
+ )
32
+ @pytest .mark .parametrize (
33
+ "critical_only" , (True , False ), ids = ("critical_only" , "all_states" )
34
+ )
35
+ def test_get_alerts (create_plain_run : tuple [sv_run .Run , dict ], from_run : bool , names_only : bool , critical_only : bool ) -> None :
36
+ run , run_data = create_plain_run
37
+ run_id = run .id
38
+ unique_id = f"{ uuid .uuid4 ()} " .split ("-" )[0 ]
39
+ _id_1 = run .create_user_alert (
40
+ name = f"user_alert_1_{ unique_id } " ,
41
+ )
42
+ _id_2 = run .create_user_alert (
43
+ name = f"user_alert_2_{ unique_id } " ,
44
+ )
45
+ _id_3 = run .create_user_alert (
46
+ name = f"user_alert_3_{ unique_id } " ,
47
+ attach_to_run = False
48
+ )
49
+ run .log_alert (identifier = _id_1 , state = "critical" )
50
+ time .sleep (2 )
51
+ run .close ()
52
+
31
53
client = svc .Client ()
32
- _ , run_data = create_test_run
33
- if from_run :
34
- triggered_alerts_full = client .get_alerts (run_id = create_test_run [1 ]["run_id" ], critical_only = False , names_only = False )
35
- assert len (triggered_alerts_full ) == 7
36
- for alert in triggered_alerts_full :
37
- if alert .name == "value_above_1" :
38
- assert alert ["alert" ]["status" ]["current" ] == "critical"
54
+
55
+ if critical_only and not from_run :
56
+ with pytest .raises (RuntimeError ) as e :
57
+ _alerts = client .get_alerts (run_id = run_id if from_run else None , critical_only = critical_only , names_only = names_only )
58
+ assert "critical_only is ambiguous when returning alerts with no run ID specified." in str (e .value )
39
59
else :
40
- assert (triggered_alerts_full := client .get_alerts (names_only = True , critical_only = False ))
41
-
42
- for alert in run_data ["created_alerts" ]:
43
- assert alert in triggered_alerts_full , f"Alert '{ alert } ' was not triggered"
44
-
45
-
60
+ _alerts = client .get_alerts (run_id = run_id if from_run else None , critical_only = critical_only , names_only = names_only )
61
+
62
+ if names_only :
63
+ assert all (isinstance (item , str ) for item in _alerts )
64
+ else :
65
+ assert all (isinstance (item , AlertBase ) for item in _alerts )
66
+ _alerts = [alert .name for alert in _alerts ]
67
+
68
+ assert f"user_alert_1_{ unique_id } " in _alerts
69
+
70
+ if not from_run :
71
+ assert len (_alerts ) > 2
72
+ assert f"user_alert_3_{ unique_id } " in _alerts
73
+ else :
74
+ assert f"user_alert_3_{ unique_id } " not in _alerts
75
+ if critical_only :
76
+ assert len (_alerts ) == 1
77
+ else :
78
+ assert len (_alerts ) == 2
79
+ assert f"user_alert_2_{ unique_id } " in _alerts
80
+
46
81
@pytest .mark .dependency
47
82
@pytest .mark .client
48
83
def test_get_run_id_from_name (create_test_run : tuple [sv_run .Run , dict ]) -> None :
0 commit comments