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
158
159
160
161
162
|
# Copyright (C) 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.
"""
Test Description:
Tests to validate auth.allow and auth.reject on a volume
"""
from glusto.core import Glusto as g
from glustolibs.gluster.gluster_base_class import GlusterBaseClass, runs_on
from glustolibs.gluster.volume_ops import (set_volume_options,
volume_reset)
from glustolibs.gluster.exceptions import ExecutionError
from glustolibs.gluster.mount_ops import (mount_volume, umount_volume,
is_mounted)
@runs_on([['distributed', 'replicated', 'distributed-replicated',
'dispersed', 'distributed-dispersed', 'arbiter',
'distributed-arbiter'], ['glusterfs']])
class TestValidateAuthAllowAndAuthReject(GlusterBaseClass):
def setUp(self):
# calling GlusterBaseClass setUp
self.get_super_method(self, 'setUp')()
ret = self.setup_volume()
if not ret:
raise ExecutionError("Volume creation failed: %s"
% self.volname)
g.log.info("Volume created successfully : %s", self.volname)
def tearDown(self):
# Cleanup the volume
ret = self.cleanup_volume()
if not ret:
raise ExecutionError("Failed to cleanup the volume %s"
% self.volname)
g.log.info("Volume deleted successfully: %s", self.volname)
self.get_super_method(self, 'tearDown')()
def _set_option_and_mount_and_unmount_volumes(self, option="",
is_allowed=True):
"""
Setting volume option and then mounting and unmounting the volume
"""
# Check if an option is passed
if option:
# Setting the option passed as an argument
ret = set_volume_options(self.mnode, self.volname,
{option: self.mounts[0].client_system})
self.assertTrue(ret, "Failed to set %s option in volume: %s"
% (option, self.volname))
g.log.info("Successfully set %s option in volume: %s", option,
self.volname)
# Mounting a volume
ret, _, _ = mount_volume(self.volname, mtype=self.mount_type,
mpoint=self.mounts[0].mountpoint,
mserver=self.mnode,
mclient=self.mounts[0].client_system)
# Checking if volume was successfully mounted or not
ret = is_mounted(self.volname, mtype=self.mount_type,
mpoint=self.mounts[0].mountpoint,
mserver=self.mnode,
mclient=self.mounts[0].client_system)
if is_allowed:
self.assertTrue(ret, "Failed to mount the volume: %s"
% self.volname)
else:
self.assertFalse(ret, "Unexpected: Mounting"
" the volume %s was successful" % self.volname)
# Unmount only if the volume is supposed to be mounted
if is_allowed:
ret, _, _ = umount_volume(self.mounts[0].client_system,
self.mounts[0].mountpoint,
mtype=self.mount_type)
self.assertEqual(ret, 0, "Failed to unmount the volume: %s"
% self.volname)
def _reset_the_volume(self):
"""
Resetting the volume
"""
ret = volume_reset(self.mnode, self.volname)
self.assertTrue(ret, "Failed to reset volume: %s" % self.volname)
g.log.info("Reseting volume %s was successful", self.volname)
def _check_validate_test(self):
"""
Checking volume mounting and unmounting with auth.allow
and auth.reject option set for it
"""
# Setting auth.allow option and then mounting and unmounting volume
self._set_option_and_mount_and_unmount_volumes("auth.allow")
g.log.info("Successfully performed the set, mounting and unmounting"
" operation as expected on volume: %s", self.volname)
# Reseting the volume options
self._reset_the_volume()
# Setting auth.reject option and then checking mounting of volume
self._set_option_and_mount_and_unmount_volumes("auth.reject", False)
g.log.info("Successfully performed the set and mounting operation"
"as expected on volume: %s", self.volname)
# Reseting the volume options
self._reset_the_volume()
# Check mounting and unmounting of volume without setting any options
self._set_option_and_mount_and_unmount_volumes()
g.log.info("Successfully mounted and unmounted the volume: %s",
self.volname)
def test_validate_auth_allow_and_auth_reject(self):
"""
Test Case:
1. Create and start a volume
2. Disable brick mutliplex
2. Set auth.allow option on volume for the client address on which
volume is to be mounted
3. Mount the volume on client and then unmmount it.
4. Reset the volume
5. Set auth.reject option on volume for the client address on which
volume is to be mounted
6. Mounting the volume should fail
7. Reset the volume and mount it on client.
8. Repeat the steps 2-7 with brick multiplex enabled
"""
# Setting cluster.brick-multiplex to disable
ret = set_volume_options(self.mnode, 'all',
{'cluster.brick-multiplex': 'disable'})
self.assertTrue(ret, "Failed to set brick-multiplex to enable.")
g.log.info("Successfully set brick-multiplex to disable.")
# Checking auth options with brick multiplex disabled
self._check_validate_test()
# Setting cluster.brick-multiplex to enable
ret = set_volume_options(self.mnode, 'all',
{'cluster.brick-multiplex': 'enable'})
self.assertTrue(ret, "Failed to set brick-multiplex to enable.")
g.log.info("Successfully set brick-multiplex to enable.")
# Checking auth options with brick multiplex enabled
self._check_validate_test()
|
