# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from functools import partial
import logging
+import os
+from functools import partial
import cherrypy
import cephfs
from ..security import Scope
from ..services.cephfs import CephFS
from ..services.cephx import CephX
-from ..services.exception import serialize_dashboard_exception
+from ..services.exception import DashboardException, serialize_dashboard_exception
from ..services.ganesha import Ganesha, GaneshaConf, NFSException
from ..services.rgw_client import RgwClient
@Endpoint('GET', '/lsdir')
@ReadPermission
- def lsdir(self, root_dir=None, depth=1): # pragma: no cover
+ def lsdir(self, fs_name, root_dir=None, depth=1): # pragma: no cover
if root_dir is None:
root_dir = "/"
- depth = int(depth)
- if depth > 5:
- logger.warning("Limiting depth to maximum value of 5: "
- "input depth=%s", depth)
- depth = 5
- root_dir = '{}{}'.format(root_dir.rstrip('/'), '/')
+ if not root_dir.startswith('/'):
+ root_dir = '/{}'.format(root_dir)
+ root_dir = os.path.normpath(root_dir)
+
+ try:
+ depth = int(depth)
+ error_msg = ''
+ if depth < 0:
+ error_msg = '`depth` must be greater or equal to 0.'
+ if depth > 5:
+ logger.warning("Limiting depth to maximum value of 5: "
+ "input depth=%s", depth)
+ depth = 5
+ except ValueError:
+ error_msg = '`depth` must be an integer.'
+ finally:
+ if error_msg:
+ raise DashboardException(code=400,
+ component='nfsganesha',
+ msg=error_msg)
+
try:
- cfs = CephFS()
- root_dir = root_dir.encode()
- paths = cfs.ls_dir(root_dir, depth)
- # Convert (bytes => string) and prettify paths (strip slashes).
- paths = [p.decode().rstrip('/') for p in paths if p != root_dir]
+ cfs = CephFS(fs_name)
+ paths = [root_dir]
+ paths.extend([p['path'].rstrip('/')
+ for p in cfs.ls_dir(root_dir, depth)])
except (cephfs.ObjectNotFound, cephfs.PermissionError):
paths = []
return {'paths': paths}
from __future__ import absolute_import
import unittest
+from urllib.parse import urlencode
try:
- from mock import MagicMock, Mock
+ from mock import MagicMock, Mock, patch
except ImportError:
- from unittest.mock import MagicMock, Mock
+ # pylint: disable=ungrouped-imports
+ from unittest.mock import MagicMock, Mock, patch
import orchestrator
-from . import KVStoreMockMixin
+
from .. import mgr
-from ..settings import Settings
+from ..controllers.nfsganesha import NFSGaneshaUi
from ..services import ganesha
-from ..services.ganesha import GaneshaConf, Export, GaneshaConfParser
+from ..services.ganesha import Export, GaneshaConf, GaneshaConfParser
+from ..settings import Settings
+from . import ControllerTestCase # pylint: disable=no-name-in-module
+from . import KVStoreMockMixin # pylint: disable=no-name-in-module
class GaneshaConfTest(unittest.TestCase, KVStoreMockMixin):
self.assertEqual(export.cluster_id, '_default_')
self.assertEqual(export.attr_expiration_time, 0)
self.assertEqual(export.security_label, True)
+
+
+class NFSGaneshaUiControllerTest(ControllerTestCase):
+ @classmethod
+ def setup_server(cls):
+ # pylint: disable=protected-access
+ NFSGaneshaUi._cp_config['tools.authenticate.on'] = False
+ cls.setup_controllers([NFSGaneshaUi])
+
+ @classmethod
+ def _create_ls_dir_url(cls, fs_name, query_params):
+ api_url = '/ui-api/nfs-ganesha/lsdir/{}'.format(fs_name)
+ if query_params is not None:
+ return '{}?{}'.format(api_url, urlencode(query_params))
+ return api_url
+
+ @patch('dashboard.controllers.nfsganesha.CephFS')
+ def test_lsdir(self, cephfs_class):
+ cephfs_class.return_value.ls_dir.return_value = [
+ {'path': '/foo'},
+ {'path': '/foo/bar'}
+ ]
+ mocked_ls_dir = cephfs_class.return_value.ls_dir
+
+ reqs = [
+ {
+ 'params': None,
+ 'cephfs_ls_dir_args': ['/', 1],
+ 'path0': '/',
+ 'status': 200
+ },
+ {
+ 'params': {'root_dir': '/', 'depth': '1'},
+ 'cephfs_ls_dir_args': ['/', 1],
+ 'path0': '/',
+ 'status': 200
+ },
+ {
+ 'params': {'root_dir': '', 'depth': '1'},
+ 'cephfs_ls_dir_args': ['/', 1],
+ 'path0': '/',
+ 'status': 200
+ },
+ {
+ 'params': {'root_dir': '/foo', 'depth': '3'},
+ 'cephfs_ls_dir_args': ['/foo', 3],
+ 'path0': '/foo',
+ 'status': 200
+ },
+ {
+ 'params': {'root_dir': 'foo', 'depth': '6'},
+ 'cephfs_ls_dir_args': ['/foo', 5],
+ 'path0': '/foo',
+ 'status': 200
+ },
+ {
+ 'params': {'root_dir': '/', 'depth': '-1'},
+ 'status': 400
+ },
+ {
+ 'params': {'root_dir': '/', 'depth': 'abc'},
+ 'status': 400
+ }
+ ]
+
+ for req in reqs:
+ self._get(self._create_ls_dir_url('a', req['params']))
+ self.assertStatus(req['status'])
+
+ # Returned paths should contain root_dir as first element
+ if req['status'] == 200:
+ paths = self.json_body()['paths']
+ self.assertEqual(paths[0], req['path0'])
+ cephfs_class.assert_called_once_with('a')
+
+ # Check the arguments passed to `CephFS.ls_dir`.
+ if req.get('cephfs_ls_dir_args'):
+ mocked_ls_dir.assert_called_once_with(*req['cephfs_ls_dir_args'])
+ else:
+ mocked_ls_dir.assert_not_called()
+ mocked_ls_dir.reset_mock()
+ cephfs_class.reset_mock()
+
+ @patch('dashboard.controllers.nfsganesha.cephfs')
+ @patch('dashboard.controllers.nfsganesha.CephFS')
+ def test_lsdir_non_existed_dir(self, cephfs_class, cephfs):
+ cephfs.ObjectNotFound = Exception
+ cephfs.PermissionError = Exception
+ cephfs_class.return_value.ls_dir.side_effect = cephfs.ObjectNotFound()
+ self._get(self._create_ls_dir_url('a', {'root_dir': '/foo', 'depth': '3'}))
+ cephfs_class.assert_called_once_with('a')
+ cephfs_class.return_value.ls_dir.assert_called_once_with('/foo', 3)
+ self.assertStatus(200)
+ self.assertJsonBody({'paths': []})
projects / ceph.git / commitdiff