prop = 'pg_num'
pgp_prop = 'pg_placement_num'
health = lambda: self._get('/api/health/minimal')['health']['status'] == 'HEALTH_OK'
- t = 0;
+ t = 0
while (int(value) != pool[pgp_prop] or not health()) and t < 180:
time.sleep(2)
t += 2
@classmethod
def tearDownClass(cls):
super(PoolTest, cls).tearDownClass()
- for name in ['dashboard_pool1', 'dashboard_pool2', 'dashboard_pool3', 'dashboard_pool_update1']:
+ for name in ['dashboard_pool1', 'dashboard_pool2', 'dashboard_pool3',
+ 'dashboard_pool_update1']:
cls._ceph_cmd(['osd', 'pool', 'delete', name, name, '--yes-i-really-really-mean-it'])
cls._ceph_cmd(['osd', 'erasure-code-profile', 'rm', 'ecprofile'])
update = {
'compression_mode': None,
'compression_algorithm': None,
- 'compression_mode': None,
'compression_max_blob_size': None,
'compression_required_ratio': None,
}
img = cls._get('/api/block/image/{}/{}'.format(pool, name))
cls._task_post("/api/block/image/{}/{}/move_trash".format(pool, name),
- {'delay': delay})
+ {'delay': delay})
return img['id']
@classmethod
def remove_trash(cls, pool, image_id, image_name, force=False):
- return cls._task_delete('/api/block/image/trash/{}/{}/?image_name={}&force={}'.format('rbd', image_id, image_name, force))
+ return cls._task_delete('/api/block/image/trash/{}/{}/?image_name={}&force={}'.format(
+ 'rbd', image_id, image_name, force))
@classmethod
def get_trash(cls, pool, image_id):
def test_default_features(self):
default_features = self._get('/api/block/image/default_features')
- self.assertEqual(default_features, ['deep-flatten', 'exclusive-lock',
- 'fast-diff', 'layering',
- 'object-map'])
+ self.assertEqual(default_features, [
+ 'deep-flatten', 'exclusive-lock', 'fast-diff', 'layering', 'object-map'])
def test_image_with_special_name(self):
rbd_name = 'test/rbd'
def test_restore_trash(self):
id = self.create_image_in_trash('rbd', 'test_rbd')
- self._task_post('/api/block/image/trash/{}/{}/restore'.format('rbd', id), {'new_image_name': 'test_rbd'})
+ self._task_post('/api/block/image/trash/{}/{}/restore'.format('rbd', id),
+ {'new_image_name': 'test_rbd'})
self._get('/api/block/image/rbd/test_rbd')
self.assertStatus(200)
from __future__ import absolute_import
-import time
-
-from .helper import DashboardTestCase, JObj, JLeaf, JList
+from .helper import DashboardTestCase
class RbdMirroringTest(DashboardTestCase):
def update_pool(cls, pool, mirror_mode):
data = {'mirror_mode': mirror_mode}
return cls._task_put('/api/block/mirroring/pool/{}'.format(pool),
- data)
+ data)
@classmethod
def list_peers(cls, pool):
data = {'cluster_name': cluster_name, 'client_id': client_id}
data.update(kwargs)
return cls._task_post('/api/block/mirroring/pool/{}/peer'.format(pool),
- data)
+ data)
@classmethod
def update_peer(cls, pool, peer_uuid, **kwargs):
return cls._task_put('/api/block/mirroring/pool/{}/peer/{}'.format(pool, peer_uuid),
- kwargs)
+ kwargs)
@classmethod
def delete_peer(cls, pool, peer_uuid):
import logging
import urllib
-import six
from .helper import DashboardTestCase, JObj, JList, JLeaf
from __future__ import absolute_import
from .helper import DashboardTestCase, JList, JObj, JAny
-from pprint import pformat
class SettingsTest(DashboardTestCase):
self.assertIsNotNone(data['health_status'])
self.assertIsNotNone(data['mgr_id'])
self.assertIsNotNone(data['have_mon_connection'])
-