@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}
+ 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"}
"protocols": [4],
"transports": ["TCP"],
"clients": [{
- "addresses":["10.0.0.0/8"],
+ "addresses": ["10.0.0.0/8"],
"access_type": "RO",
"squash": "root"
}]
self._task_delete("/api/nfs-ganesha/export/{}/{}"
.format(exp['cluster_id'], exp['export_id']))
- def test_create_export(self):
+ def _test_create_export(self, cephfs_path):
exports = self._get("/api/nfs-ganesha/export")
self.assertEqual(len(exports), 0)
- data = self.create_export("/foo", 'cluster1', ['node1', 'node2'], 'CEPH', "security.selinux")
+ 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()
+ export = self._test_create_export('/foo')
export['access_type'] = 'RO'
export['daemons'] = ['node1', 'node3']
export['security_label'] = True
self.assertEqual(exports[0]['security_label'], True)
def test_delete_export(self):
- export = self.test_create_export()
+ export = self._test_create_export('/foo')
self._task_delete("/api/nfs-ganesha/export/{}/{}"
.format(export['cluster_id'], export['export_id']))
self.assertStatus(204)