summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorSri Vignesh <sselvan@redhat.com>2019-09-25 14:12:51 +0530
committerSri Vignesh <sselvan@redhat.com>2019-11-21 12:49:19 +0530
commit746e741ea2e728383b8591356bac1d7d23140ca1 (patch)
tree0d9465b6ef5c34568331da22ef78c3e1adfca2fd
parente7264123e6463f418df8c6d7b75f91e504294ecb (diff)
Modify and move 'topology_volumes_with_bricks' func to 'heketi_ops.py'
Move 'topology_volumes_with_bricks' func to the "heketi_ops" module to be able to use it in other test cases. Change-Id: I7cca884a4f3fb34ec15bb947d3c39d9226e294d0 Signed-off-by: Sri Vignesh <sselvan@redhat.com>
-rw-r--r--openshift-storage-libs/openshiftstoragelibs/heketi_ops.py32
-rw-r--r--tests/functional/heketi/test_heketi_create_volume.py61
2 files changed, 54 insertions, 39 deletions
diff --git a/openshift-storage-libs/openshiftstoragelibs/heketi_ops.py b/openshift-storage-libs/openshiftstoragelibs/heketi_ops.py
index 7cb0478..f2c8196 100644
--- a/openshift-storage-libs/openshiftstoragelibs/heketi_ops.py
+++ b/openshift-storage-libs/openshiftstoragelibs/heketi_ops.py
@@ -1778,3 +1778,35 @@ def heketi_volume_endpoint_patch(
out = heketi_cmd_run(heketi_client_node, cmd)
return json.loads(out)
+
+
+def get_heketi_volume_and_brick_count_list(
+ heketi_client_node, heketi_server_url, **kwargs):
+ """Calculate amount of volumes and bricks.
+
+ Args:
+ heketi_client_node (str): Node on which cmd has to be executed.
+ heketi_server_url (str): Heketi server url
+
+ Kwargs:
+ The keys, values in kwargs are:
+ - secret : (str)|None
+ - user : (str)|None
+
+ Returns:
+ list of tuples containing volume name and brick count
+
+ example:
+ [('heketidbstorage', 3), ('vol_dcedb64fae938d8a72d0749c2159fcdb', 6)]
+
+ Raises:
+ AssertionError: if command fails.
+
+ """
+ topology_info = heketi_topology_info(
+ heketi_client_node, heketi_server_url, json=True, **kwargs)
+ volume_name_brick_count = []
+ for c in topology_info['clusters']:
+ volume_name_brick_count = [
+ (v['name'], len(v['bricks'])) for v in c['volumes']]
+ return volume_name_brick_count
diff --git a/tests/functional/heketi/test_heketi_create_volume.py b/tests/functional/heketi/test_heketi_create_volume.py
index 1afabcf..56e72db 100644
--- a/tests/functional/heketi/test_heketi_create_volume.py
+++ b/tests/functional/heketi/test_heketi_create_volume.py
@@ -1,16 +1,10 @@
-try:
- # py2/3
- import simplejson as json
-except ImportError:
- # py2
- import json
-
from glusto.core import Glusto as g
from glustolibs.gluster.volume_ops import get_volume_list, get_volume_info
import six
from openshiftstoragelibs.baseclass import BaseClass
from openshiftstoragelibs.heketi_ops import (
+ get_heketi_volume_and_brick_count_list,
heketi_blockvolume_create,
heketi_blockvolume_delete,
heketi_cluster_delete,
@@ -18,7 +12,6 @@ from openshiftstoragelibs.heketi_ops import (
heketi_node_delete,
heketi_node_info,
heketi_node_list,
- heketi_topology_info,
heketi_volume_create,
heketi_volume_delete,
heketi_volume_expand,
@@ -112,19 +105,6 @@ class TestHeketiVolume(BaseClass):
self.assertTrue(vol_info, "Failed to get volume info %s" % name)
g.log.info("Successfully got the volume info %s" % name)
- def topology_volumes_with_bricks(self):
- g.log.info("Retrieving heketi topology info")
- topology_info_out = heketi_topology_info(self.heketi_client_node,
- self.heketi_server_url,
- json=True)
- indented_topology = json.dumps(topology_info_out, indent=4)
- g.log.info("Successfully got the heketi topology info")
- topology_volumes = dict()
- for c in topology_info_out['clusters']:
- for v in c['volumes']:
- topology_volumes.update({v['name']: len(v['bricks'])})
- return topology_volumes, indented_topology
-
def test_create_vol_and_retrieve_topology_info(self):
volume_names = []
volume_ids = []
@@ -145,35 +125,38 @@ class TestHeketiVolume(BaseClass):
self.heketi_server_url, volume_ids[1], 1)
# Check if volume is shown in the heketi topology
- topology_volumes, indented_topology = (
- self.topology_volumes_with_bricks())
- for vol_name in volume_names:
- self.assertIn(vol_name, topology_volumes.keys(), (
- "volume %s not found in the heketi topology\n Topology "
- "info:\n%s" % (vol_name, indented_topology)))
- bricks = 6 if vol_name == volume_names[1] else 3
+ topology_volumes = get_heketi_volume_and_brick_count_list(
+ self.heketi_client_node, self.heketi_server_url)
+ existing_volumes = [v for v, _ in topology_volumes]
+ for v in volume_names:
+ self.assertIn(v, existing_volumes)
+ for v, b_count in topology_volumes:
+ expected_bricks_count = 6 if v == volume_names[1] else 3
self.assertGreaterEqual(
- topology_volumes[vol_name], bricks, 'Bricks of the volume:'
- '%s are %s and it should be more or equal to %s' %
- (vol_name, topology_volumes[vol_name], bricks))
+ b_count, expected_bricks_count,
+ 'Bricks number of the %s volume is %s and it is expected '
+ 'to be greater or equal to %s' % (
+ v, b_count, expected_bricks_count))
# Delete first 2 volumes and verify their deletion in the topology
for vol_id in volume_ids[:2]:
g.log.info("Deleting volume %s" % vol_id)
heketi_volume_delete(self.heketi_client_node,
self.heketi_server_url, vol_id)
- topology_volumes.clear()
- topology_volumes, indented_topology = (
- self.topology_volumes_with_bricks())
+ topology_volumes = get_heketi_volume_and_brick_count_list(
+ self.heketi_client_node, self.heketi_server_url)
+ existing_volumes = [v for v, _ in topology_volumes]
for vol_name in volume_names[:2]:
- self.assertNotIn(vol_name, topology_volumes.keys(), (
+ self.assertNotIn(vol_name, existing_volumes, (
"volume %s shown in the heketi topology after deletion"
- "\nTopology info:\n%s" % (vol_name, indented_topology)))
+ "\nTopology info:\n%s" % (
+ vol_name, existing_volumes)))
# Check the existence of third volume
- self.assertIn(volume_names[2], topology_volumes.keys(), ("volume %s "
- "not shown in the heketi topology\nTopology info\n%s" % (
- volume_ids[2], indented_topology)))
+ self.assertIn(
+ volume_names[2], existing_volumes, "volume %s not "
+ "shown in the heketi topology\nTopology info"
+ "\n%s" % (volume_ids[2], existing_volumes))
g.log.info("Sucessfully verified the topology info")
def test_to_check_deletion_of_cluster(self):