self.assertStatus(200)
def test_safe_to_destroy(self):
- self._get('/api/osd/5/safe_to_destroy')
+ osd_dump = json.loads(self._ceph_cmd(['osd', 'dump', '-f', 'json']))
+ unused_osd_id = max(map(lambda e: e['osd'], osd_dump['osds'])) + 10
+ self._get('/api/osd/{}/safe_to_destroy'.format(unused_osd_id))
self.assertStatus(200)
self.assertJsonBody({'safe-to-destroy': True})
def get_destroy_status():
self._get('/api/osd/0/safe_to_destroy')
- return self.jsonBody()['safe-to-destroy']
+ if 'safe-to-destroy' in self.jsonBody():
+ return self.jsonBody()['safe-to-destroy']
+ return None
self.wait_until_equal(get_destroy_status, False, 10)
self.assertStatus(200)
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import re
from . import ApiController, RESTController, UpdatePermission
from .. import mgr, logger
from ..security import Scope
'mon', 'osd safe-to-destroy', ids=svc_id, target=('mgr', ''))
return {'safe-to-destroy': True}
except SendCommandError as e:
- match = re.match(
- r'OSD\(s\) (\d+) have (\d+) pgs currently mapped to them',
- e.message)
- if match:
- return {
- 'message': e.message,
- 'safe-to-destroy': False
- }
- else:
- raise e
+ return {
+ 'message': e.message,
+ 'safe-to-destroy': False,
+ }
@ApiController('/osd/flags', Scope.OSD)