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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
|
# Copyright (C) 2019-2020 Red Hat, Inc. <http://www.redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
from random import choice
import string
from glusto.core import Glusto as g
from glustolibs.gluster.gluster_base_class import (GlusterBaseClass, runs_on)
from glustolibs.gluster.exceptions import ExecutionError
from glustolibs.gluster.brickmux_ops import (disable_brick_mux,
is_brick_mux_enabled,
get_brick_mux_status)
from glustolibs.gluster.lib_utils import search_pattern_in_file
from glustolibs.gluster.volume_ops import set_volume_options
@runs_on([['replicated'],
['glusterfs']])
class TestBrickMultiplexing(GlusterBaseClass):
"""
Tests for brick multiplexing statuses
"""
def tearDown(self):
# Disable brick multiplexing
g.log.info("Checking for brick multiplexing status...")
if is_brick_mux_enabled(self.mnode):
g.log.info("Disabling brick multiplexing...")
if not disable_brick_mux(self.mnode):
raise ExecutionError("Failed to disable brick multiplexing")
g.log.info("Disabled brick multiplexing successfully")
# Calling GlusterBaseClass teardown
self.get_super_method(self, 'tearDown')()
@staticmethod
def get_random_string(chars, str_len=4):
return ''.join((choice(chars) for _ in range(str_len)))
def test_enabling_brick_mux(self):
"""
Test case:
- check if brick multiplex is disable by default
- check for warning message triggering by setting brick-multiplex and
choosing 'n' in y/n
- check if brick multiplex is disabled after triggering warning message
- check brick multiplex for all possible statuses
(positive and negative)
- check for brick multiplex status in /var/lib/glusterd/options file
"""
# Check if brickmux is disabled by default
g.log.info('Checking if brick multiplex operation is disabled...')
self.assertFalse(is_brick_mux_enabled(self.mnode),
"Brick multiplex is not disabled by default")
# Check for warning message while changing status
warning_message = ("Brick-multiplexing is supported only for "
"OCS converged or independent mode. Also it is "
"advised to make sure that either all volumes are "
"in stopped state or no bricks are running before "
"this option is modified."
"Do you still want to continue? (y/n)")
g.log.info('Triggering warning message...')
cmd = "gluster v set all cluster.brick-multiplex enable"
_, out, _ = g.run(self.mnode, cmd)
g.log.info('Checking for warning message in output...')
if "volume set: success" not in out:
self.assertIn(warning_message, out,
'There is no warning message in '
'output or message is incorrect.')
g.log.info('Warning message is correct.')
else:
g.log.info('Skipped warning message check.')
# If brick-mux is enabled then disabling it.
if is_brick_mux_enabled(self.mnode):
if not disable_brick_mux(self.mnode):
g.log.info("Disabling brick multiplexing as it"
" was enabled due to no warning message.")
# Check if brickmux is still disabled
g.log.info('Checking if brick multiplex is still disabled')
self.assertFalse(is_brick_mux_enabled(self.mnode),
"Brick multiplex operation is not disabled")
# Enable brick multiplex with all possible statuses
statuses = ['on', 'enable', '1', 'true',
'off', 'disable', '0', 'false']
for status in statuses:
g.log.info('Enabling brick multiplex with %s status...',
status)
cmd = ("yes | gluster v set all cluster.brick-multiplex %s"
% status)
_, out, _ = g.run(self.mnode, cmd)
self.assertIn('success', out,
'Failed on enabling brick multiplexing')
# Check if brick multiplex status is correct
g.log.info('Checking if brick multiplexing status is correct...')
gluster_status = get_brick_mux_status(self.mnode)
self.assertEqual(status, gluster_status,
"Brick multiplex status is not correct")
g.log.info('Brick multiplex status "%s" is correct',
status)
# Check for brick multiplexing status in file 'options'
g.log.info("Checking for brick multiplexing status '%s' in file "
"'/var/lib/glusterd/options'...", status)
search_pattern = 'cluster.brick-multiplex=%s' % status
self.assertTrue(search_pattern_in_file(self.mnode, search_pattern,
'/var/lib/glusterd/options',
'', ''))
g.log.info("Brick multiplexing status '%s' in file "
"'/var/lib/glusterd/options' is correct", status)
# Check brick multiplex with incorrect status
g.log.info('Checking brick multiplex with incorrect status...')
cmd = "yes | gluster v set all cluster.brick-multiplex incorrect"
ret, _, _ = g.run(self.mnode, cmd)
self.assertEqual(ret, 1, 'Incorrect status has passed')
def test_enabling_brick_mux_with_wrong_values(self):
"""
Test Case:
- Create a gluster cluster
- Set cluster.brick-multiplex value to random string(Must fail)
- Set cluster.brick-multiplex value to random int(Must fail)
- Set cluster.brick-multiplex value to random
special characters(Must fail)
"""
# Creation of random data for cluster.brick-multiplex option
# Data has: alphabets, numbers, punctuations and their combinations
key = 'cluster.brick-multiplex'
for char_type in (string.ascii_letters, string.punctuation,
string.printable, string.digits):
temp_val = self.get_random_string(char_type)
value = "{}".format(temp_val)
ret = set_volume_options(self.mnode, self.volname, {key: value})
self.assertFalse(ret, "Unexpected: Erroneous value {}, to option "
"{} should result in failure".format(value, key))
g.log.info("Expected: Erroneous value %s, to option "
"%s resulted in failure", value, key)
|
