@@ -23,44 +23,48 @@ def to_dict(self) -> Dict[str, str]:
23
23
24
24
25
25
class ActionRequest :
26
- def __init__ (self , action_type : ActionType , parameters : Union [Dict [str , Any ], RestartDmcParams ]):
26
+ def __init__ (
27
+ self ,
28
+ action_type : ActionType ,
29
+ parameters : Union [Dict [str , Any ], RestartDmcParams ],
30
+ ):
27
31
self .type = action_type
28
32
self .parameters = parameters
29
33
30
34
def to_dict (self ) -> Dict [str , Any ]:
31
35
return {
32
36
"type" : self .type .value , # Use the string value of the enum
33
- "parameters" : self .parameters .to_dict () if isinstance (self .parameters ,
34
- RestartDmcParams ) else self .parameters
37
+ "parameters" : self .parameters .to_dict ()
38
+ if isinstance (self .parameters , RestartDmcParams )
39
+ else self .parameters ,
35
40
}
36
41
37
42
38
43
class FaultInjectorClient :
39
44
def __init__ (self , base_url : str ):
40
- self .base_url = base_url .rstrip ('/' )
45
+ self .base_url = base_url .rstrip ("/" )
41
46
42
- def _make_request (self , method : str , path : str , data : Optional [Dict ] = None ) -> Dict [str , Any ]:
47
+ def _make_request (
48
+ self , method : str , path : str , data : Optional [Dict ] = None
49
+ ) -> Dict [str , Any ]:
43
50
url = f"{ self .base_url } { path } "
44
51
headers = {"Content-Type" : "application/json" } if data else {}
45
52
46
53
request_data = None
47
54
if data :
48
- request_data = json .dumps (data ).encode (' utf-8' )
55
+ request_data = json .dumps (data ).encode (" utf-8" )
49
56
print (f"JSON payload being sent: { request_data .decode ('utf-8' )} " )
50
57
51
58
request = urllib .request .Request (
52
- url ,
53
- method = method ,
54
- data = request_data ,
55
- headers = headers
59
+ url , method = method , data = request_data , headers = headers
56
60
)
57
61
58
62
try :
59
63
with urllib .request .urlopen (request ) as response :
60
- return json .loads (response .read ().decode (' utf-8' ))
64
+ return json .loads (response .read ().decode (" utf-8" ))
61
65
except urllib .error .HTTPError as e :
62
66
if e .code == 422 :
63
- error_body = json .loads (e .read ().decode (' utf-8' ))
67
+ error_body = json .loads (e .read ().decode (" utf-8" ))
64
68
raise ValueError (f"Validation Error: { error_body } " )
65
69
raise
66
70
@@ -77,32 +81,31 @@ def trigger_action(self, action_request: ActionRequest) -> Dict[str, Any]:
77
81
def get_action_status (self , action_id : str ) -> Dict [str , Any ]:
78
82
"""Get the status of a specific action"""
79
83
return self ._make_request ("GET" , f"/action/{ action_id } " )
80
-
81
- def execute_rladmin_command (self , command : str , bdb_id : str = None ) -> Dict [str , Any ]:
84
+
85
+ def execute_rladmin_command (
86
+ self , command : str , bdb_id : str = None
87
+ ) -> Dict [str , Any ]:
82
88
"""Execute rladmin command directly as string"""
83
89
url = f"{ self .base_url } /rladmin"
84
-
90
+
85
91
# The fault injector expects the raw command string
86
92
command_string = f"rladmin { command } "
87
93
if bdb_id :
88
94
command_string = f"rladmin -b { bdb_id } { command } "
89
-
95
+
90
96
print (f"Sending rladmin command: { command_string } " )
91
-
97
+
92
98
headers = {"Content-Type" : "text/plain" }
93
-
99
+
94
100
request = urllib .request .Request (
95
- url ,
96
- method = "POST" ,
97
- data = command_string .encode ('utf-8' ),
98
- headers = headers
101
+ url , method = "POST" , data = command_string .encode ("utf-8" ), headers = headers
99
102
)
100
-
103
+
101
104
try :
102
105
with urllib .request .urlopen (request ) as response :
103
- return json .loads (response .read ().decode (' utf-8' ))
106
+ return json .loads (response .read ().decode (" utf-8" ))
104
107
except urllib .error .HTTPError as e :
105
108
if e .code == 422 :
106
- error_body = json .loads (e .read ().decode (' utf-8' ))
109
+ error_body = json .loads (e .read ().decode (" utf-8" ))
107
110
raise ValueError (f"Validation Error: { error_body } " )
108
- raise
111
+ raise
0 commit comments