class TestVolumes(TestVolumesHelper):
"""Tests for FS volume operations."""
- def test_volume_create(self):
- """
- That the volume can be created and then cleans up
- """
- volname = self._gen_vol_name()
- self._fs_cmd("volume", "create", volname)
- volumels = json.loads(self._fs_cmd("volume", "ls"))
-
- if not (volname in ([volume['name'] for volume in volumels])):
- raise RuntimeError("Error creating volume '{0}'".format(volname))
-
- # check that the pools were created with the correct config
- pool_details = json.loads(self._raw_cmd("osd", "pool", "ls", "detail", "--format=json"))
- pool_flags = {}
- for pool in pool_details:
- pool_flags[pool["pool_id"]] = pool["flags_names"].split(",")
-
- volume_details = json.loads(self._fs_cmd("get", volname, "--format=json"))
- for data_pool_id in volume_details['mdsmap']['data_pools']:
- self.assertIn("bulk", pool_flags[data_pool_id])
- meta_pool_id = volume_details['mdsmap']['metadata_pool']
- self.assertNotIn("bulk", pool_flags[meta_pool_id])
-
- # clean up
- self._fs_cmd("volume", "rm", volname, "--yes-i-really-mean-it")
-
def test_volume_ls(self):
"""
That the existing and the newly created volumes can be listed and
" of subvolumegroup")
+class TestVolumeCreate(TestVolumesHelper):
+ '''
+ Contains test for "ceph fs volume create" command.
+ '''
+
+ def test_volume_create(self):
+ """
+ That the volume can be created and then cleans up
+ """
+ volname = self._gen_vol_name()
+ self._fs_cmd("volume", "create", volname)
+ volumels = json.loads(self._fs_cmd("volume", "ls"))
+
+ if not (volname in ([volume['name'] for volume in volumels])):
+ raise RuntimeError("Error creating volume '{0}'".format(volname))
+
+ # check that the pools were created with the correct config
+ pool_details = json.loads(self._raw_cmd("osd", "pool", "ls", "detail", "--format=json"))
+ pool_flags = {}
+ for pool in pool_details:
+ pool_flags[pool["pool_id"]] = pool["flags_names"].split(",")
+
+ volume_details = json.loads(self._fs_cmd("get", volname, "--format=json"))
+ for data_pool_id in volume_details['mdsmap']['data_pools']:
+ self.assertIn("bulk", pool_flags[data_pool_id])
+ meta_pool_id = volume_details['mdsmap']['metadata_pool']
+ self.assertNotIn("bulk", pool_flags[meta_pool_id])
+
+ # clean up
+ self._fs_cmd("volume", "rm", volname, "--yes-i-really-mean-it")
+
class TestRenameCmd(TestVolumesHelper):
def test_volume_rename(self):