2
2
from copy import deepcopy
3
3
from typing import List
4
4
5
- from chaoslib .types import Activity , Experiment , Run
5
+ from chaoslib .types import Activity , Experiment , Run
6
6
7
7
__all__ = ["after_activity_control" ]
8
8
logger = logging .getLogger ("chaostoolkit" )
9
9
10
10
11
- def after_activity_control (context : Activity , experiment : Experiment ,
12
- state : Run , repeat_count : int = 0 ,
13
- break_if_previous_iteration_failed : bool = False
14
- ) -> None :
11
+ def after_activity_control (
12
+ context : Activity ,
13
+ experiment : Experiment ,
14
+ state : Run ,
15
+ repeat_count : int = 0 ,
16
+ break_if_previous_iteration_failed : bool = False ,
17
+ ) -> None :
15
18
"""
16
19
Repeat the activity a certain number of times.
17
20
@@ -72,28 +75,28 @@ def after_activity_control(context: Activity, experiment: Experiment,
72
75
if break_if_previous_iteration_failed and last_status != "succeeded" :
73
76
logger .debug (
74
77
"Last iteration failed so stopping our iterations "
75
- f"of '{ activity_name } '" )
78
+ f"of '{ activity_name } '"
79
+ )
76
80
return
77
81
78
- hypothesis_activities = experiment .get (
79
- "steady-state-hypothesis" , {}). get ( " probes" , [])
80
- repeat_activity (
81
- activity , hypothesis_activities , repeat_count = repeat_count )
82
+ hypothesis_activities = experiment .get ("steady-state-hypothesis" , {}). get (
83
+ "probes" , []
84
+ )
85
+ repeat_activity ( activity , hypothesis_activities , repeat_count = repeat_count )
82
86
83
87
method_activities = experiment .get ("method" , [])
84
- repeat_activity (
85
- activity , method_activities , repeat_count = repeat_count )
88
+ repeat_activity (activity , method_activities , repeat_count = repeat_count )
86
89
87
90
rollback_activities = experiment .get ("rollbacks" , [])
88
- repeat_activity (
89
- activity , rollback_activities , repeat_count = repeat_count )
91
+ repeat_activity (activity , rollback_activities , repeat_count = repeat_count )
90
92
91
93
92
94
###############################################################################
93
95
# Internals
94
96
###############################################################################
95
- def repeat_activity (activity : Activity , activities : List [Activity ],
96
- repeat_count : int = 0 ) -> None :
97
+ def repeat_activity (
98
+ activity : Activity , activities : List [Activity ], repeat_count : int = 0
99
+ ) -> None :
97
100
if not activities :
98
101
return
99
102
@@ -102,8 +105,8 @@ def repeat_activity(activity: Activity, activities: List[Activity],
102
105
103
106
for pos , a in enumerate (copy_activities ):
104
107
if a ["name" ] == activity_name :
105
- for index in range (1 , repeat_count + 1 ):
108
+ for index in range (1 , repeat_count + 1 ):
106
109
new_activity = deepcopy (activity )
107
110
new_activity ["iteration_index" ] = index
108
- activities .insert (pos + index , new_activity )
111
+ activities .insert (pos + index , new_activity )
109
112
break
0 commit comments