for osd in osds:
self.assertIn('osd.{}'.format(osd['osd']), data)
+ def _validate_perf(self, srv_id, srv_type, data, allow_empty):
+ self.assertIsInstance(data, dict)
+ self.assertEqual(srv_type, data['service']['type'])
+ self.assertEqual(str(srv_id), data['service']['id'])
+ self.assertIsInstance(data['counters'], list)
+ if not allow_empty:
+ self.assertGreater(len(data['counters']), 0)
+ for counter in data['counters'][0:1]:
+ self.assertIsInstance(counter, dict)
+ self.assertIn('description', counter)
+ self.assertIn('name', counter)
+ self.assertIn('unit', counter)
+ self.assertIn('value', counter)
+
+
@authenticate
def test_perf_counters_mon_get(self):
mon = self.mons()[0]
data = self._get('/api/perf_counters/mon/{}'.format(mon))
self.assertStatus(200)
-
- self.assertIsInstance(data, dict)
- self.assertEqual('mon', data['service']['type'])
- self.assertEqual(mon, data['service']['id'])
- self.assertIsInstance(data['counters'], list)
- self.assertGreater(len(data['counters']), 0)
- counter = data['counters'][0]
- self.assertIsInstance(counter, dict)
- self.assertIn('description', counter)
- self.assertIn('name', counter)
- self.assertIn('unit', counter)
- self.assertIn('value', counter)
+ self._validate_perf(mon, 'mon', data, allow_empty=False)
@authenticate
def test_perf_counters_mgr_get(self):
mgr = self.mgr_cluster.mgr_ids[0]
data = self._get('/api/perf_counters/mgr/{}'.format(mgr))
self.assertStatus(200)
+ self._validate_perf(mgr, 'mgr', data, allow_empty=False)
- self.assertIsInstance(data, dict)
- self.assertEqual('mgr', data['service']['type'])
- self.assertEqual(mgr, data['service']['id'])
- self.assertIsInstance(data['counters'], list)
- self.assertGreater(len(data['counters']), 0)
- counter = data['counters'][0]
- self.assertIsInstance(counter, dict)
- self.assertIn('description', counter)
- self.assertIn('name', counter)
- self.assertIn('unit', counter)
- self.assertIn('value', counter)
+ @authenticate
+ def test_perf_counters_mds_get(self):
+ for mds in self.mds_cluster.mds_ids:
+ data = self._get('/api/perf_counters/mds/{}'.format(mds))
+ self.assertStatus(200)
+ self._validate_perf(mds, 'mds', data, allow_empty=True)
+
+ @authenticate
+ def test_perf_counters_osd_get(self):
+ for osd in self.ceph_cluster.mon_manager.get_osd_dump():
+ osd = osd['osd']
+ data = self._get('/api/perf_counters/osd/{}'.format(osd))
+ self.assertStatus(200)
+ self._validate_perf(osd, 'osd', data, allow_empty=False)