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
|
from glustolibs.gluster.snap_ops import (
snap_create,
snap_delete,
snap_list,
)
import pytest
from openshiftstoragelibs.baseclass import BaseClass
from openshiftstoragelibs.command import cmd_run
from openshiftstoragelibs.heketi_ops import (
heketi_node_info,
heketi_volume_create,
heketi_volume_delete,
heketi_volume_expand,
heketi_volume_info,
)
from openshiftstoragelibs.openshift_ops import cmd_run_on_gluster_pod_or_node
from openshiftstoragelibs import podcmd
from openshiftstoragelibs import utils
class TestHeketiVolumeOperations(BaseClass):
"""
Class to test heketi volume operations - create, expand
"""
@classmethod
def setUpClass(cls):
super(TestHeketiVolumeOperations, cls).setUpClass()
cls.volume_size = 1
@pytest.mark.tier1
def test_heketi_with_default_options(self):
"""
Test to create volume with default options.
"""
vol_info = heketi_volume_create(self.heketi_client_node,
self.heketi_server_url,
self.volume_size, json=True)
self.assertTrue(vol_info, ("Failed to create heketi volume of size %s"
% self.volume_size))
self.addCleanup(
heketi_volume_delete,
self.heketi_client_node, self.heketi_server_url, vol_info['id'])
self.assertEqual(vol_info['size'], self.volume_size,
("Failed to create volume with default options."
"Expected Size: %s, Actual Size: %s"
% (self.volume_size, vol_info['size'])))
@pytest.mark.tier1
def test_heketi_with_expand_volume(self):
"""
Test volume expand and size if updated correctly in heketi-cli info
"""
vol_info = heketi_volume_create(self.heketi_client_node,
self.heketi_server_url,
self.volume_size, json=True)
self.assertTrue(vol_info, ("Failed to create heketi volume of size %s"
% self.volume_size))
self.addCleanup(
heketi_volume_delete,
self.heketi_client_node, self.heketi_server_url, vol_info['id'])
self.assertEqual(vol_info['size'], self.volume_size,
("Failed to create volume."
"Expected Size: %s, Actual Size: %s"
% (self.volume_size, vol_info['size'])))
volume_id = vol_info["id"]
expand_size = 2
ret = heketi_volume_expand(self.heketi_client_node,
self.heketi_server_url, volume_id,
expand_size)
self.assertTrue(ret, ("Failed to expand heketi volume of id %s"
% volume_id))
volume_info = heketi_volume_info(self.heketi_client_node,
self.heketi_server_url,
volume_id, json=True)
expected_size = self.volume_size + expand_size
self.assertEqual(volume_info['size'], expected_size,
("Volume Expansion failed Expected Size: %s, Actual "
"Size: %s" % (str(expected_size),
str(volume_info['size']))))
@pytest.mark.tier1
def test_heketi_volume_mount(self):
self.node = self.ocp_master_node[0]
try:
cmd_run('rpm -q glusterfs-fuse', self.node)
except AssertionError:
self.skipTest(
"gluster-fuse package is not present on Node %s" % self.node)
h_node, h_url = self.heketi_client_node, self.heketi_server_url
# Create volume
vol_info = heketi_volume_create(h_node, h_url, 2, json=True)
self.addCleanup(heketi_volume_delete, h_node, h_url, vol_info['id'])
mount_point = vol_info['mount']['glusterfs']['device']
mount_dir = '/mnt/dir-%s' % utils.get_random_str()
mount_cmd = 'mount -t glusterfs %s %s' % (mount_point, mount_dir)
# Create directory to mount volume
cmd_run('mkdir %s' % mount_dir, self.node)
self.addCleanup(cmd_run, 'rm -rf %s' % mount_dir, self.node)
# Mount volume
cmd_run(mount_cmd, self.node)
self.addCleanup(cmd_run, 'umount %s' % mount_dir, self.node)
# Run I/O to make sure Mount point works
_file = 'file'
run_io_cmd = (
'dd if=/dev/urandom of=%s/%s bs=4k count=1000' % (
mount_dir, _file))
# Verify size of volume
cmd_run(run_io_cmd, self.node)
size = cmd_run(
'df -kh --output=size %s | tail -1' % mount_dir, self.node).strip()
self.assertEqual('2.0G', size)
# Verify file on gluster vol bricks
for brick in vol_info['bricks']:
node_id = brick['node']
node_info = heketi_node_info(h_node, h_url, node_id, json=True)
brick_host = node_info['hostnames']['storage'][0]
cmd_run_on_gluster_pod_or_node(self.node, 'ls %s/%s' % (
brick['path'], _file), brick_host)
@pytest.mark.tier1
@podcmd.GlustoPod()
def test_heketi_volume_snapshot_create(self):
"""Test heketi volume snapshot create operation"""
h_volume_size = 1
snap_name = 'snap_test_heketi_volume_snapshot_create_1'
h_node, h_url = self.heketi_client_node, self.heketi_server_url
h_volume_info = heketi_volume_create(
h_node, h_url, h_volume_size, json=True)
self.addCleanup(
heketi_volume_delete, h_node, h_url, h_volume_info["id"])
h_volume_name = h_volume_info["name"]
ret, _, _ = snap_create(
'auto_get_gluster_endpoint',
h_volume_name, snap_name, timestamp=False)
self.addCleanup(
podcmd.GlustoPod()(snap_delete),
"auto_get_gluster_endpoint", snap_name)
self.assertEqual(
ret, 0, "Failed to create snapshot {} for heketi volume {}"
.format(snap_name, h_volume_name))
ret, out, _ = snap_list('auto_get_gluster_endpoint')
self.assertEqual(
ret, 0, "Failed to list snapshot {} for heketi volume"
.format(snap_name))
self.assertIn(
snap_name, out, "Heketi volume snapshot {} not found in {}"
.format(snap_name, out))
|
