def __init__(self):
logger.info('Initializing controller: %s -> %s',
self.__class__.__name__, self._cp_path_)
+ super(BaseController, self).__init__()
def _has_permissions(self, permissions, scope=None):
if not self._cp_config['tools.authenticate.on']:
from .. import mgr, logger
-LANGUAGES = {f for f in os.listdir(mgr.get_frontend_path())
- if os.path.isdir(os.path.join(mgr.get_frontend_path(), f))}
-LANGUAGES_PATH_MAP = {f.lower(): {'lang': f, 'path': os.path.join(mgr.get_frontend_path(), f)}
- for f in LANGUAGES}
-# pre-populating with the primary language subtag
-for _lang in list(LANGUAGES_PATH_MAP.keys()):
- if '-' in _lang:
- LANGUAGES_PATH_MAP[_lang.split('-')[0]] = {
- 'lang': LANGUAGES_PATH_MAP[_lang]['lang'], 'path': LANGUAGES_PATH_MAP[_lang]['path']}
-
-
-def _get_default_language():
- with open("{}/../package.json".format(mgr.get_frontend_path()), "r") as f:
- config = json.load(f)
- return config['config']['locale']
-
-
-DEFAULT_LANGUAGE = _get_default_language()
-DEFAULT_LANGUAGE_PATH = os.path.join(mgr.get_frontend_path(), DEFAULT_LANGUAGE)
+class LanguageMixin(object):
+ def __init__(self):
+ self.LANGUAGES = {
+ f
+ for f in os.listdir(mgr.get_frontend_path())
+ if os.path.isdir(os.path.join(mgr.get_frontend_path(), f))
+ }
+ self.LANGUAGES_PATH_MAP = {
+ f.lower(): {
+ 'lang': f,
+ 'path': os.path.join(mgr.get_frontend_path(), f)
+ }
+ for f in self.LANGUAGES
+ }
+ # pre-populating with the primary language subtag.
+ for lang in list(self.LANGUAGES_PATH_MAP.keys()):
+ if '-' in lang:
+ self.LANGUAGES_PATH_MAP[lang.split('-')[0]] = {
+ 'lang': self.LANGUAGES_PATH_MAP[lang]['lang'],
+ 'path': self.LANGUAGES_PATH_MAP[lang]['path']
+ }
+ with open("{}/../package.json".format(mgr.get_frontend_path()),
+ "r") as f:
+ config = json.load(f)
+ self.DEFAULT_LANGUAGE = config['config']['locale']
+ self.DEFAULT_LANGUAGE_PATH = os.path.join(mgr.get_frontend_path(),
+ self.DEFAULT_LANGUAGE)
+ super(LanguageMixin, self).__init__()
@Controller("/", secure=False)
-class HomeController(BaseController):
+class HomeController(BaseController, LanguageMixin):
LANG_TAG_SEQ_RE = re.compile(r'\s*([^,]+)\s*,?\s*')
LANG_TAG_RE = re.compile(
r'^(?P<locale>[a-zA-Z]{1,8}(-[a-zA-Z0-9]{1,8})?)(;q=(?P<weight>[01]\.\d{0,3}))?$')
def _language_dir(self, langs):
for lang in langs:
- if lang in LANGUAGES_PATH_MAP:
+ if lang in self.LANGUAGES_PATH_MAP:
logger.debug("found directory for language '%s'", lang)
- cherrypy.response.headers['Content-Language'] = LANGUAGES_PATH_MAP[lang]['lang']
- return LANGUAGES_PATH_MAP[lang]['path']
+ cherrypy.response.headers[
+ 'Content-Language'] = self.LANGUAGES_PATH_MAP[lang]['lang']
+ return self.LANGUAGES_PATH_MAP[lang]['path']
- logger.debug("Languages '%s' not available, falling back to %s", langs, DEFAULT_LANGUAGE)
- cherrypy.response.headers['Content-Language'] = DEFAULT_LANGUAGE
- return DEFAULT_LANGUAGE_PATH
+ logger.debug("Languages '%s' not available, falling back to %s",
+ langs, self.DEFAULT_LANGUAGE)
+ cherrypy.response.headers['Content-Language'] = self.DEFAULT_LANGUAGE
+ return self.DEFAULT_LANGUAGE_PATH
@Proxy()
def __call__(self, path, **params):
accept_lang_header = cherrypy.request.headers['Accept-Language']
langs = self._parse_accept_language(accept_lang_header)
else:
- langs = [DEFAULT_LANGUAGE.lower()]
+ langs = [self.DEFAULT_LANGUAGE.lower()]
logger.debug("frontend language from headers: %s", langs)
base_dir = self._language_dir(langs)
@UiApiController("/langs", secure=False)
-class LangsController(BaseController):
+class LangsController(BaseController, LanguageMixin):
@Endpoint('GET')
def __call__(self):
- return list(LANGUAGES)
+ return list(self.LANGUAGES)
virtualenv==15.1.0
werkzeug==0.14.1
wrapt==1.10.11
+pyfakefs
import json
import threading
+import sys
import time
import cherrypy
from cherrypy._cptools import HandlerWrapperTool
from cherrypy.test import helper
+from pyfakefs import fake_filesystem
from mgr_module import CLICommand, MgrModule
return exec_dashboard_cmd(None, cmd, **kwargs)
+class FakeFsMixin(object):
+ fs = fake_filesystem.FakeFilesystem()
+ f_open = fake_filesystem.FakeFileOpen(fs)
+ f_os = fake_filesystem.FakeOsModule(fs)
+
+ if sys.version_info > (3, 0):
+ builtins_open = 'builtins.open'
+ else:
+ builtins_open = '__builtin__.open'
+
+
class ControllerTestCase(helper.CPWebCase):
_endpoints_cache = {}
from ..controllers.grafana import Grafana
from ..grafana import GrafanaRestClient
from ..settings import Settings
-from .. import mgr
-class GrafanaTest(ControllerTestCase):
+class GrafanaTest(ControllerTestCase, KVStoreMockMixin):
@classmethod
def setup_server(cls):
- cls.server_settings()
# pylint: disable=protected-access
Grafana._cp_config['tools.authenticate.on'] = False
cls.setup_controllers([Grafana])
- @classmethod
+ def setUp(self):
+ self.mock_kv_store()
+
+ @staticmethod
def server_settings(
- cls,
url='http://localhost:3000',
user='admin',
password='admin',
):
- settings = dict()
if url is not None:
- settings['GRAFANA_API_URL'] = url
+ Settings.GRAFANA_API_URL = url
if user is not None:
- settings['GRAFANA_API_USERNAME'] = user
+ Settings.GRAFANA_API_USERNAME = user
if password is not None:
- settings['GRAFANA_API_PASSWORD'] = password
- mgr.get_module_option.side_effect = settings.get
+ Settings.GRAFANA_API_PASSWORD = password
def test_url(self):
self.server_settings()
self._get('/api/grafana/validation/foo')
self.assertStatus(500)
- def test_dashboards(self):
+ def test_dashboards_unavailable_no_url(self):
self.server_settings(url=None)
self._post('/api/grafana/dashboards')
self.assertStatus(500)
+
+ def test_dashboards_unavailable_no_user(self):
self.server_settings(user=None)
self._post('/api/grafana/dashboards')
self.assertStatus(500)
+
+ def test_dashboards_unavailable_no_password(self):
self.server_settings(password=None)
self._post('/api/grafana/dashboards')
self.assertStatus(500)
from __future__ import absolute_import
import logging
+import os
-from . import ControllerTestCase
-from ..controllers.home import HomeController
+try:
+ import mock
+except ImportError:
+ import unittest.mock as mock
+from . import ControllerTestCase, FakeFsMixin
+from .. import mgr
+
+from ..controllers.home import HomeController, LanguageMixin
logger = logging.getLogger()
-class HomeTest(ControllerTestCase):
+class HomeTest(ControllerTestCase, FakeFsMixin):
@classmethod
def setup_server(cls):
+ frontend_path = mgr.get_frontend_path()
+ cls.fs.reset()
+ cls.fs.create_dir(frontend_path)
+ cls.fs.create_file(
+ os.path.join(frontend_path, '..', 'package.json'),
+ contents='{"config":{"locale": "en-US"}}')
+ with mock.patch(cls.builtins_open, new=cls.f_open),\
+ mock.patch('os.listdir', new=cls.f_os.listdir):
+ lang = LanguageMixin()
+ cls.fs.create_file(
+ os.path.join(lang.DEFAULT_LANGUAGE_PATH, 'index.html'),
+ contents='<!doctype html><html lang="en"><body></body></html>')
cls.setup_controllers([HomeController])
+ @mock.patch(FakeFsMixin.builtins_open, new=FakeFsMixin.f_open)
+ @mock.patch('os.stat', new=FakeFsMixin.f_os.stat)
+ @mock.patch('os.listdir', new=FakeFsMixin.f_os.listdir)
def test_home_default_lang(self):
self._get('/')
self.assertStatus(200)
logger.info(self.body)
self.assertIn('<html lang="en">', self.body.decode('utf-8'))
+ @mock.patch(FakeFsMixin.builtins_open, new=FakeFsMixin.f_open)
+ @mock.patch('os.stat', new=FakeFsMixin.f_os.stat)
+ @mock.patch('os.listdir', new=FakeFsMixin.f_os.listdir)
def test_home_uplevel_check(self):
self._get('/../../../../../../etc/shadow')
self.assertStatus(403)
+ @mock.patch(FakeFsMixin.builtins_open, new=FakeFsMixin.f_open)
+ @mock.patch('os.stat', new=FakeFsMixin.f_os.stat)
+ @mock.patch('os.listdir', new=FakeFsMixin.f_os.listdir)
def test_home_en_us(self):
self._get('/', headers=[('Accept-Language', 'en-US')])
self.assertStatus(200)
logger.info(self.body)
self.assertIn('<html lang="en">', self.body.decode('utf-8'))
+ @mock.patch(FakeFsMixin.builtins_open, new=FakeFsMixin.f_open)
+ @mock.patch('os.stat', new=FakeFsMixin.f_os.stat)
+ @mock.patch('os.listdir', new=FakeFsMixin.f_os.listdir)
def test_home_non_supported_lang(self):
self._get('/', headers=[('Accept-Language', 'NO-NO')])
self.assertStatus(200)
# -*- coding: utf-8 -*-
import unittest
-from .. import mgr
from ..services.rgw_client import RgwClient
+from ..settings import Settings
from . import KVStoreMockMixin
})
def test_ssl_verify(self):
- mgr.set_module_option('RGW_API_SSL_VERIFY', True)
+ Settings.RGW_API_SSL_VERIFY = True
instance = RgwClient.admin_instance()
self.assertTrue(instance.session.verify)
def test_no_ssl_verify(self):
- mgr.set_module_option('RGW_API_SSL_VERIFY', False)
+ Settings.RGW_API_SSL_VERIFY = False
instance = RgwClient.admin_instance()
self.assertFalse(instance.session.verify)
projects / ceph.git / commitdiff