ceph/qa/tasks/mgr/dashboard/test_cephfs.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 from .helper import DashboardTestCase
   5
   6
   7 class CephfsTest(DashboardTestCase):
   8     CEPHFS = True
   9
  10     AUTH_ROLES = ['cephfs-manager']
  11
  12     @DashboardTestCase.RunAs('test', 'test', ['block-manager'])
  13     def test_access_permissions(self):
  14         fs_id = self.fs.get_namespace_id()
  15         self._get("/api/cephfs/{}/clients".format(fs_id))
  16         self.assertStatus(403)
  17         self._get("/api/cephfs/{}".format(fs_id))
  18         self.assertStatus(403)
  19         self._get("/api/cephfs/{}/mds_counters".format(fs_id))
  20         self.assertStatus(403)
  21
  22     def test_cephfs_clients(self):
  23         fs_id = self.fs.get_namespace_id()
  24         data = self._get("/api/cephfs/{}/clients".format(fs_id))
  25         self.assertStatus(200)
  26
  27         self.assertIn('status', data)
  28         self.assertIn('data', data)
  29
  30     def test_cephfs_get(self):
  31         fs_id = self.fs.get_namespace_id()
  32         data = self._get("/api/cephfs/{}/".format(fs_id))
  33         self.assertStatus(200)
  34
  35         self.assertIn('cephfs', data)
  36         self.assertIn('standbys', data)
  37         self.assertIn('versions', data)
  38         self.assertIsNotNone(data['cephfs'])
  39         self.assertIsNotNone(data['standbys'])
  40         self.assertIsNotNone(data['versions'])
  41
  42     def test_cephfs_mds_counters(self):
  43         fs_id = self.fs.get_namespace_id()
  44         data = self._get("/api/cephfs/{}/mds_counters".format(fs_id))
  45         self.assertStatus(200)
  46
  47         self.assertIsInstance(data, dict)
  48         self.assertIsNotNone(data)
  49
  50     def test_cephfs_mds_counters_wrong(self):
  51         self._get("/api/cephfs/baadbaad/mds_counters")
  52         self.assertStatus(400)
  53         self.assertJsonBody({
  54             "component": 'cephfs',
  55             "code": "invalid_cephfs_id",
  56             "detail": "Invalid cephfs ID baadbaad"
  57         })
  58
  59     def test_cephfs_list(self):
  60         data = self._get("/api/cephfs/")
  61         self.assertStatus(200)
  62         self.assertIsInstance(data, list)
  63
  64         cephfs = data[0]
  65         self.assertIn('id', cephfs)
  66         self.assertIn('mdsmap', cephfs)
  67         self.assertIsNotNone(cephfs['id'])
  68         self.assertIsNotNone(cephfs['mdsmap'])