1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
import json
from pcs.common import communication
from pcs.common.interface.dto import to_dict
from pcs_test.tools.command_env.mock_node_communicator import (
place_communication,
)
class ScsiShortcuts:
def __init__(self, calls):
self.__calls = calls
def unfence_node(
self,
original_devices=(),
updated_devices=(),
node_labels=None,
communication_list=None,
name="http.scsi.unfence_node",
):
"""
Create a calls for node unfencing
list original_devices -- list of scsi devices before an update
list updated_devices -- list of scsi devices after an update
list node_labels -- create success responses from these nodes
list communication_list -- use these custom responses
string name -- the key of this call
"""
if (node_labels is None and communication_list is None) or (
node_labels and communication_list
):
raise AssertionError(
"Exactly one of 'node_labels', 'communication_list' "
"must be specified"
)
if node_labels:
communication_list = [
dict(
label=node,
raw_data=json.dumps(
dict(
node=node,
original_devices=original_devices,
updated_devices=updated_devices,
)
),
)
for node in node_labels
]
place_communication(
self.__calls,
name,
communication_list,
action="api/v1/scsi-unfence-node/v2",
output=json.dumps(
to_dict(
communication.dto.InternalCommunicationResultDto(
status=communication.const.COM_STATUS_SUCCESS,
status_msg=None,
report_list=[],
data=None,
)
)
),
)
def unfence_node_mpath(
self,
node_key_map,
original_devices=(),
updated_devices=(),
node_labels=None,
communication_list=None,
name="http.scsi.unfence_node",
):
"""
Create a calls for node unfencing
dict node_key_map -- map of node name to its registration key
list original_devices -- list of scsi devices before an update
list updated_devices -- list of scsi devices after an update
list node_labels -- create success responses from these nodes
list communication_list -- use these custom responses
string name -- the key of this call
"""
if (node_labels is None and communication_list is None) or (
node_labels and communication_list
):
raise AssertionError(
"Exactly one of 'node_labels', 'communication_list' "
"must be specified"
)
if node_labels:
communication_list = [
dict(
label=node,
raw_data=json.dumps(
dict(
key=node_key_map[node],
original_devices=original_devices,
updated_devices=updated_devices,
)
),
)
for node in node_labels
]
place_communication(
self.__calls,
name,
communication_list,
action="api/v1/scsi-unfence-node-mpath/v1",
output=json.dumps(
to_dict(
communication.dto.InternalCommunicationResultDto(
status=communication.const.COM_STATUS_SUCCESS,
status_msg=None,
report_list=[],
data=None,
)
)
),
)
|
