# -*- coding: utf-8 -*- # pylint: disable=too-many-public-methods from __future__ import absolute_import from .helper import DashboardTestCase, JList, JObj class GaneshaTest(DashboardTestCase): CEPHFS = True AUTH_ROLES = ['pool-manager', 'ganesha-manager'] @classmethod def setUpClass(cls): super(GaneshaTest, cls).setUpClass() cls.create_pool('ganesha', 2**2, 'replicated') cls._rados_cmd(['-p', 'ganesha', '-N', 'ganesha1', 'create', 'conf-node1']) cls._rados_cmd(['-p', 'ganesha', '-N', 'ganesha1', 'create', 'conf-node2']) cls._rados_cmd(['-p', 'ganesha', '-N', 'ganesha1', 'create', 'conf-node3']) cls._rados_cmd(['-p', 'ganesha', '-N', 'ganesha2', 'create', 'conf-node1']) cls._rados_cmd(['-p', 'ganesha', '-N', 'ganesha2', 'create', 'conf-node2']) cls._rados_cmd(['-p', 'ganesha', '-N', 'ganesha2', 'create', 'conf-node3']) cls._ceph_cmd(['dashboard', 'set-ganesha-clusters-rados-pool-namespace', 'cluster1:ganesha/ganesha1,cluster2:ganesha/ganesha2']) # RGW setup cls._radosgw_admin_cmd([ 'user', 'create', '--uid', 'admin', '--display-name', 'admin', '--system', '--access-key', 'admin', '--secret', 'admin' ]) cls._ceph_cmd_with_secret(['dashboard', 'set-rgw-api-secret-key'], 'admin') cls._ceph_cmd_with_secret(['dashboard', 'set-rgw-api-access-key'], 'admin') @classmethod def tearDownClass(cls): super(GaneshaTest, cls).tearDownClass() cls._radosgw_admin_cmd(['user', 'rm', '--uid', 'admin', '--purge-data']) cls._ceph_cmd(['osd', 'pool', 'delete', 'ganesha', 'ganesha', '--yes-i-really-really-mean-it']) @DashboardTestCase.RunAs('test', 'test', [{'rbd-image': ['create', 'update', 'delete']}]) def test_read_access_permissions(self): self._get('/api/nfs-ganesha/export') self.assertStatus(403) def test_list_daemons(self): daemons = self._get("/api/nfs-ganesha/daemon") self.assertEqual(len(daemons), 6) daemons = [(d['daemon_id'], d['cluster_id']) for d in daemons] self.assertIn(('node1', 'cluster1'), daemons) self.assertIn(('node2', 'cluster1'), daemons) self.assertIn(('node3', 'cluster1'), daemons) self.assertIn(('node1', 'cluster2'), daemons) self.assertIn(('node2', 'cluster2'), daemons) self.assertIn(('node3', 'cluster2'), daemons) @classmethod def create_export(cls, path, cluster_id, daemons, fsal, sec_label_xattr=None): if fsal == 'CEPH': fsal = {"name": "CEPH", "user_id": "admin", "fs_name": None, "sec_label_xattr": sec_label_xattr} pseudo = "/cephfs{}".format(path) else: fsal = {"name": "RGW", "rgw_user_id": "admin"} pseudo = "/rgw/{}".format(path if path[0] != '/' else "") ex_json = { "path": path, "fsal": fsal, "cluster_id": cluster_id, "daemons": daemons, "pseudo": pseudo, "tag": None, "access_type": "RW", "squash": "no_root_squash", "security_label": sec_label_xattr is not None, "protocols": [4], "transports": ["TCP"], "clients": [{ "addresses": ["10.0.0.0/8"], "access_type": "RO", "squash": "root" }] } return cls._task_post('/api/nfs-ganesha/export', ex_json) def tearDown(self): super(GaneshaTest, self).tearDown() exports = self._get("/api/nfs-ganesha/export") if self._resp.status_code != 200: return self.assertIsInstance(exports, list) for exp in exports: self._task_delete("/api/nfs-ganesha/export/{}/{}" .format(exp['cluster_id'], exp['export_id'])) def _test_create_export(self, cephfs_path): exports = self._get("/api/nfs-ganesha/export") self.assertEqual(len(exports), 0) data = self.create_export(cephfs_path, 'cluster1', ['node1', 'node2'], 'CEPH', "security.selinux") exports = self._get("/api/nfs-ganesha/export") self.assertEqual(len(exports), 1) self.assertDictEqual(exports[0], data) return data def test_create_export(self): self._test_create_export('/foo') def test_create_export_for_cephfs_root(self): self._test_create_export('/') def test_update_export(self): export = self._test_create_export('/foo') export['access_type'] = 'RO' export['daemons'] = ['node1', 'node3'] export['security_label'] = True data = self._task_put('/api/nfs-ganesha/export/{}/{}' .format(export['cluster_id'], export['export_id']), export) exports = self._get("/api/nfs-ganesha/export") self.assertEqual(len(exports), 1) self.assertDictEqual(exports[0], data) self.assertEqual(exports[0]['daemons'], ['node1', 'node3']) self.assertEqual(exports[0]['security_label'], True) def test_delete_export(self): export = self._test_create_export('/foo') self._task_delete("/api/nfs-ganesha/export/{}/{}" .format(export['cluster_id'], export['export_id'])) self.assertStatus(204) def test_get_export(self): exports = self._get("/api/nfs-ganesha/export") self.assertEqual(len(exports), 0) data1 = self.create_export("/foo", 'cluster2', ['node1', 'node2'], 'CEPH') data2 = self.create_export("mybucket", 'cluster2', ['node2', 'node3'], 'RGW') export1 = self._get("/api/nfs-ganesha/export/cluster2/1") self.assertDictEqual(export1, data1) export2 = self._get("/api/nfs-ganesha/export/cluster2/2") self.assertDictEqual(export2, data2) def test_invalid_status(self): self._ceph_cmd(['dashboard', 'set-ganesha-clusters-rados-pool-namespace', '']) data = self._get('/api/nfs-ganesha/status') self.assertStatus(200) self.assertIn('available', data) self.assertIn('message', data) self.assertFalse(data['available']) self.assertIn(("NFS-Ganesha cluster is not detected. " "Please set the GANESHA_RADOS_POOL_NAMESPACE " "setting or deploy an NFS-Ganesha cluster with the Orchestrator."), data['message']) self._ceph_cmd(['dashboard', 'set-ganesha-clusters-rados-pool-namespace', 'cluster1:ganesha/ganesha1,cluster2:ganesha/ganesha2']) def test_valid_status(self): data = self._get('/api/nfs-ganesha/status') self.assertStatus(200) self.assertIn('available', data) self.assertIn('message', data) self.assertTrue(data['available']) def test_ganesha_fsals(self): data = self._get('/ui-api/nfs-ganesha/fsals') self.assertStatus(200) self.assertIn('CEPH', data) def test_ganesha_filesystems(self): data = self._get('/ui-api/nfs-ganesha/cephfs/filesystems') self.assertStatus(200) self.assertSchema(data, JList(JObj({ 'id': int, 'name': str }))) def test_ganesha_lsdir(self): fss = self._get('/ui-api/nfs-ganesha/cephfs/filesystems') self.assertStatus(200) for fs in fss: data = self._get('/ui-api/nfs-ganesha/lsdir/{}'.format(fs['name'])) self.assertStatus(200) self.assertSchema(data, JObj({'paths': JList(str)})) self.assertEqual(data['paths'][0], '/') def test_ganesha_buckets(self): data = self._get('/ui-api/nfs-ganesha/rgw/buckets') self.assertStatus(200) schema = JList(str) self.assertSchema(data, schema) def test_ganesha_clusters(self): data = self._get('/ui-api/nfs-ganesha/clusters') self.assertStatus(200) schema = JList(str) self.assertSchema(data, schema) def test_ganesha_cephx_clients(self): data = self._get('/ui-api/nfs-ganesha/cephx/clients') self.assertStatus(200) schema = JList(str) self.assertSchema(data, schema)