class PingTest(ControllerTestCase):
@classmethod
def setup_test(cls):
- Ping._cp_config['tools.authenticate.on'] = False
- cls.setup_controllers([Ping])
+ cp_config = {'tools.authenticate.on': True}
+ cls.setup_controllers([Ping], cp_config=cp_config)
def test_ping(self):
self._get("/api/ping")
The ``ControllerTestCase`` class starts by initializing a CherryPy webserver.
Then it will call the ``setup_test()`` class method where we can explicitly
load the controllers that we want to test. In the above example we are only
-loading the ``Ping`` controller. We can also disable authentication of a
-controller at this stage, as depicted in the example.
+loading the ``Ping`` controller. We can also provide ``cp_config`` in order to
+update the controller's cherrypy config (e.g. enable authentication as shown in the example).
How to update or create new dashboards in grafana?
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
import logging
import threading
import time
+from typing import Any, Dict
from unittest.mock import Mock
import cherrypy
_endpoints_cache = {}
@classmethod
- def setup_controllers(cls, ctrl_classes, base_url=''):
+ def setup_controllers(cls, ctrl_classes, base_url='', cp_config: Dict[str, Any] = None):
if not isinstance(ctrl_classes, list):
ctrl_classes = [ctrl_classes]
mapper = cherrypy.dispatch.RoutesDispatcher()
endpoint_list = []
for ctrl in ctrl_classes:
+ ctrl._cp_config = {
+ 'tools.dashboard_exception_handler.on': True,
+ 'tools.authenticate.on': False
+ }
+ if cp_config:
+ ctrl._cp_config.update(cp_config)
inst = ctrl()
# We need to cache the controller endpoints because
from ..services.access_control import SYSTEM_ROLES, AccessControlDB, \
PasswordPolicy, load_access_control_db, password_hash
from ..settings import Settings
-from . import CLICommandTestMixin, CmdException # pylint: disable=no-name-in-module
+from ..tests import CLICommandTestMixin, CmdException
class AccessControlTest(unittest.TestCase, CLICommandTestMixin):
from .. import mgr
from ..controllers import RESTController, Router
-from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase, KVStoreMockMixin
# pylint: disable=W0613
import unittest
+from unittest.mock import Mock, patch
from .. import mgr
+from ..controllers.auth import Auth
from ..services.auth import JwtManager
+from ..tests import ControllerTestCase
+
+mgr.get_module_option.return_value = JwtManager.JWT_TOKEN_TTL
+mgr.get_store.return_value = 'jwt_secret'
+mgr.ACCESS_CTRL_DB = Mock()
+mgr.ACCESS_CTRL_DB.get_attempt.return_value = 1
class JwtManagerTest(unittest.TestCase):
self.assertIsInstance(decoded_token, dict)
self.assertEqual(decoded_token['iss'], 'ceph-dashboard')
self.assertEqual(decoded_token['username'], 'my-username')
+
+
+class AuthTest(ControllerTestCase):
+
+ @classmethod
+ def setup_server(cls):
+ cls.setup_controllers([Auth])
+
+ def test_request_not_authorized(self):
+ self.setup_controllers([Auth], cp_config={'tools.authenticate.on': True})
+ self._post('/api/auth/logout')
+ self.assertStatus(401)
+
+ @patch('dashboard.controllers.auth.JwtManager.gen_token', Mock(return_value='my-token'))
+ @patch('dashboard.controllers.auth.AuthManager.authenticate', Mock(return_value={
+ 'permissions': {'read-only': ['read']},
+ 'pwdExpirationDate': 1000000,
+ 'pwdUpdateRequired': False
+ }))
+ def test_login(self):
+ self._post('/api/auth', {'username': 'my-user', 'password': 'my-pass'})
+ self.assertStatus(201)
+ self.assertJsonBody({
+ 'token': 'my-token',
+ 'username': 'my-user',
+ 'permissions': {'read-only': ['read']},
+ 'pwdExpirationDate': 1000000,
+ 'sso': False,
+ 'pwdUpdateRequired': False
+ })
+
+ @patch('dashboard.controllers.auth.JwtManager', Mock())
+ def test_logout(self):
+ self._post('/api/auth/logout')
+ self.assertStatus(200)
+ self.assertJsonBody({
+ 'redirect_url': '#/login'
+ })
from unittest.mock import patch, Mock
from ..controllers.cephfs import CephFS
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
class MetaDataMock(object):
from __future__ import absolute_import
from ..controllers import APIRouter, BaseController, Endpoint, RESTController, Router
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
@Router("/btest/{key}", base_url="/ui", secure=False)
from __future__ import absolute_import
from ..api.doc import SchemaType
-from ..controllers import APIDoc, APIRouter, Endpoint, EndpointDoc, RESTController
+from ..controllers import ENDPOINT_MAP, APIDoc, APIRouter, Endpoint, EndpointDoc, RESTController
from ..controllers._version import APIVersion
from ..controllers.docs import Docs
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
# Dummy controller and endpoint that can be assigned with @EndpointDoc and @GroupDoc
class DocsTest(ControllerTestCase):
@classmethod
def setup_server(cls):
+ ENDPOINT_MAP.clear()
cls.setup_controllers([DecoratedController, Docs], "/test")
def test_type_to_str(self):
self.assertTrue(any(base in key.split('/')[1] for base in ['api', 'ui-api']))
def test_gen_tags(self):
- outcome = Docs()._gen_tags(False)[0]
- self.assertEqual({'description': 'Group description', 'name': 'FooGroup'}, outcome)
+ outcome = Docs._gen_tags(False)
+ self.assertEqual([{'description': 'Group description', 'name': 'FooGroup'}], outcome)
from .. import mgr
from ..controllers.erasure_code_profile import ErasureCodeProfile
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
class ErasureCodeProfileTest(ControllerTestCase):
'fs_map': {'filesystems': []},
}[key]
- # pylint: disable=protected-access
- ErasureCodeProfile._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([ErasureCodeProfile])
def test_list(self):
from ..services.ceph_service import SendCommandError
from ..services.exception import handle_rados_error, \
handle_send_command_error, serialize_dashboard_exception
+from ..tests import ControllerTestCase
from ..tools import NotificationQueue, TaskManager, ViewCache
-from . import ControllerTestCase # pylint: disable=no-name-in-module
# pylint: disable=W0613
from unittest.mock import Mock, patch
from ..plugins.feature_toggles import Actions, Features, FeatureToggles
-from . import KVStoreMockMixin # pylint: disable=no-name-in-module
+from ..tests import KVStoreMockMixin
class SettingsTest(unittest.TestCase, KVStoreMockMixin):
from ..controllers.grafana import Grafana
from ..grafana import GrafanaRestClient
from ..settings import Settings
-from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase, KVStoreMockMixin
class GrafanaTest(ControllerTestCase, KVStoreMockMixin):
@classmethod
def setup_server(cls):
- # pylint: disable=protected-access
- Grafana._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Grafana])
def setUp(self):
from .. import mgr
from ..controllers.home import HomeController, LanguageMixin
-from . import ControllerTestCase, FakeFsMixin # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase, FakeFsMixin
logger = logging.getLogger()
from .. import mgr
from ..controllers._version import APIVersion
from ..controllers.host import Host, HostUi, get_device_osd_map, get_hosts, get_inventories
+from ..tests import ControllerTestCase
from ..tools import NotificationQueue, TaskManager
-from . import ControllerTestCase # pylint: disable=no-name-in-module
@contextlib.contextmanager
def setup_server(cls):
NotificationQueue.start_queue()
TaskManager.init()
- # pylint: disable=protected-access
- Host._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Host])
@classmethod
@classmethod
def setup_server(cls):
- # pylint: disable=protected-access
- HostUi._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([HostUi])
def test_labels(self):
from ..rest_client import RequestException
from ..services.iscsi_client import IscsiClient
from ..services.orchestrator import OrchClient
+from ..tests import CLICommandTestMixin, CmdException, ControllerTestCase, KVStoreMockMixin
from ..tools import NotificationQueue, TaskManager
-from . import CLICommandTestMixin # pylint: disable=no-name-in-module
-from . import CmdException # pylint: disable=no-name-in-module
-from . import ControllerTestCase # pylint: disable=no-name-in-module
-from . import KVStoreMockMixin # pylint: disable=no-name-in-module
class IscsiTestCli(unittest.TestCase, CLICommandTestMixin):
TaskManager.init()
OrchClient.instance().available = lambda: False
mgr.rados.side_effect = None
- # pylint: disable=protected-access
- Iscsi._cp_config['tools.authenticate.on'] = False
- IscsiTarget._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Iscsi, IscsiTarget])
@classmethod
from ..controllers.orchestrator import Orchestrator
from ..services.orchestrator import OrchFeature
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
class OrchestratorControllerTest(ControllerTestCase):
@classmethod
def setup_server(cls):
- # pylint: disable=protected-access
- Orchestrator._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Orchestrator])
@mock.patch('dashboard.controllers.orchestrator.OrchClient.instance')
from .. import mgr
from ..controllers.osd import Osd
+from ..tests import ControllerTestCase
from ..tools import NotificationQueue, TaskManager
-from . import ControllerTestCase # pylint: disable=no-name-in-module
from .helper import update_dict # pylint: disable=import-error
class OsdTest(ControllerTestCase):
@classmethod
def setup_server(cls):
- Osd._cp_config['tools.authenticate.on'] = False # pylint: disable=protected-access
cls.setup_controllers([Osd])
NotificationQueue.start_queue()
TaskManager.init()
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-from . import CLICommandTestMixin, ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import CLICommandTestMixin, ControllerTestCase
class TestPluginDebug(ControllerTestCase, CLICommandTestMixin):
from ..controllers.pool import Pool
from ..controllers.task import Task
+from ..tests import ControllerTestCase
from ..tools import NotificationQueue, TaskManager
-from . import ControllerTestCase # pylint: disable=no-name-in-module
class MockTask(object):
class PoolControllerTest(ControllerTestCase):
@classmethod
def setup_server(cls):
- Task._cp_config['tools.authenticate.on'] = False
- Pool._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Pool, Task])
@mock.patch('dashboard.services.progress.get_progress_tasks')
from .. import mgr
from ..controllers.prometheus import Prometheus, PrometheusNotifications, PrometheusReceiver
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
class PrometheusControllerTest(ControllerTestCase):
'PROMETHEUS_API_HOST': cls.prometheus_host
}
mgr.get_module_option.side_effect = settings.get
- Prometheus._cp_config['tools.authenticate.on'] = False
- PrometheusNotifications._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Prometheus, PrometheusNotifications, PrometheusReceiver])
def test_rules(self):
from ..controllers.rbd_mirroring import RbdMirroring, RbdMirroringPoolBootstrap, RbdMirroringSummary
from ..controllers.summary import Summary
from ..services import progress
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
mock_list_servers = [{
'hostname': 'ceph-host',
@classmethod
def setup_server(cls):
- # pylint: disable=protected-access
- RbdMirroring._cp_config['tools.authenticate.on'] = False
- # pylint: enable=protected-access
-
cls.setup_controllers([RbdMirroring])
@mock.patch('dashboard.controllers.rbd_mirroring.rbd.RBD')
@classmethod
def setup_server(cls):
- # pylint: disable=protected-access
- RbdMirroringPoolBootstrap._cp_config['tools.authenticate.on'] = False
- # pylint: enable=protected-access
-
cls.setup_controllers([RbdMirroringPoolBootstrap])
@mock.patch('dashboard.controllers.rbd_mirroring.rbd.RBD')
progress.get_progress_tasks = mock.MagicMock()
progress.get_progress_tasks.return_value = ([], [])
- # pylint: disable=protected-access
- RbdMirroringSummary._cp_config['tools.authenticate.on'] = False
- Summary._cp_config['tools.authenticate.on'] = False
- # pylint: enable=protected-access
-
cls.setup_controllers([RbdMirroringSummary, Summary], '/test')
@mock.patch('dashboard.controllers.rbd_mirroring.rbd.RBD')
from ..controllers import RESTController, Router, Task
from ..controllers.task import Task as TaskController
from ..services import progress
+from ..tests import ControllerTestCase
from ..tools import NotificationQueue, TaskManager
-from . import ControllerTestCase # pylint: disable=no-name-in-module
@Router('/test/task', secure=False)
NotificationQueue.start_queue()
TaskManager.init()
- TaskTest._cp_config['tools.authenticate.on'] = False
- TaskController._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([TaskTest, TaskController])
@classmethod
from ..controllers.rgw import Rgw, RgwDaemon, RgwUser
from ..rest_client import RequestException
from ..services.rgw_client import RgwClient
-from . import ControllerTestCase, RgwStub # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase, RgwStub
class RgwControllerTestCase(ControllerTestCase):
@classmethod
def setup_server(cls):
- Rgw._cp_config['tools.authenticate.on'] = False # pylint: disable=protected-access
cls.setup_controllers([Rgw], '/test')
def setUp(self) -> None:
class RgwDaemonControllerTestCase(ControllerTestCase):
@classmethod
def setup_server(cls):
- RgwDaemon._cp_config['tools.authenticate.on'] = False # pylint: disable=protected-access
cls.setup_controllers([RgwDaemon], '/test')
@patch('dashboard.services.rgw_client.RgwClient._get_user_id', Mock(
class RgwUserControllerTestCase(ControllerTestCase):
@classmethod
def setup_server(cls):
- RgwUser._cp_config['tools.authenticate.on'] = False # pylint: disable=protected-access
cls.setup_controllers([RgwUser], '/test')
@patch('dashboard.controllers.rgw.RgwRESTController.proxy')
from ..services.rgw_client import NoCredentialsException, \
NoRgwDaemonsException, RgwClient, _parse_frontend_config
from ..settings import Settings
-from . import CLICommandTestMixin, RgwStub # pylint: disable=no-name-in-module
+from ..tests import CLICommandTestMixin, RgwStub
@patch('dashboard.services.rgw_client.RgwClient._get_user_id', Mock(
from .. import settings
from ..controllers.settings import Settings as SettingsController
from ..settings import Settings, handle_option_command
-from . import ControllerTestCase, KVStoreMockMixin # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase, KVStoreMockMixin
class SettingsTest(unittest.TestCase, KVStoreMockMixin):
class SettingsControllerTest(ControllerTestCase, KVStoreMockMixin):
@classmethod
def setup_server(cls):
- # pylint: disable=protected-access
-
- SettingsController._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([SettingsController])
@classmethod
import unittest
from ..services.sso import load_sso_db
-from . import CLICommandTestMixin, CmdException # pylint: disable=no-name-in-module
+from ..tests import CLICommandTestMixin, CmdException
class AccessControlTest(unittest.TestCase, CLICommandTestMixin):
from ..controllers import APIRouter, BaseController, Proxy, RESTController, Router
from ..controllers._version import APIVersion
from ..services.exception import handle_rados_error
+from ..tests import ControllerTestCase
from ..tools import dict_contains_path, dict_get, json_str_to_object, \
merge_list_of_dicts_by_key, partial_dict
-from . import ControllerTestCase # pylint: disable=no-name-in-module
# pylint: disable=W0613
from ..controllers._api_router import APIRouter
from ..controllers._rest_controller import RESTController
from ..controllers._version import APIVersion
-from . import ControllerTestCase # pylint: disable=no-name-in-module
+from ..tests import ControllerTestCase
@APIRouter("/vtest", secure=False)
projects / ceph.git / commitdiff