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
|
from pcs_test.tools.command_env.mock_node_communicator import (
place_multinode_call,
)
class SbdShortcuts:
def __init__(self, calls):
self.__calls = calls
def enable_sbd(
self,
node_labels=None,
communication_list=None,
name="http.sbd.enable_sbd",
):
place_multinode_call(
self.__calls,
name,
node_labels,
communication_list,
action="remote/sbd_enable",
)
def disable_sbd(
self,
node_labels=None,
communication_list=None,
name="http.sbd.disable_sbd",
):
"""
Create a call for disabling sbd on nodes
node_labels list -- create success responses from these nodes
communication_list list -- create custom responses
name string -- the key of this call
"""
place_multinode_call(
self.__calls,
name,
node_labels,
communication_list,
action="remote/sbd_disable",
)
def check_sbd(self, communication_list=None, name="http.sbd.check_sbd"):
place_multinode_call(
self.__calls,
name,
communication_list=communication_list,
action="remote/check_sbd",
)
def set_sbd_config(
self,
config_generator=None,
node_labels=None,
communication_list=None,
name="http.sbd.set_sbd_config",
):
if bool(config_generator) == bool(communication_list):
raise AssertionError(
"Exactly one of 'config_generator', 'communication_list' "
"must be specified"
)
if config_generator and not node_labels:
raise AssertionError(
"'node_labels' has to be defined if 'config_generator' is used"
)
if communication_list is None:
communication_list = [
dict(
param_list=[("config", config_generator(node))],
label=node,
)
for node in node_labels
]
place_multinode_call(
self.__calls,
name,
None,
communication_list,
action="remote/set_sbd_config",
)
|
