# pylint: disable=W0212
from __future__ import absolute_import
-import os
-import unittest
+import json
-import requests
+import cherrypy
+from cherrypy.test import helper
+from ..controllers.auth import Auth
+from ..tools import json_error_page, SessionExpireAtBrowserCloseTool
-class ControllerTestCase(unittest.TestCase):
- DASHBOARD_HOST = os.environ.get('DASHBOARD_V2_HOST', "localhost")
- DASHBOARD_PORT = os.environ.get('DASHBOARD_V2_PORT', 8080)
+class ControllerTestCase(helper.CPWebCase):
def __init__(self, *args, **kwargs):
- self.dashboard_host = kwargs.pop('dashboard_host') \
- if 'dashboard_host' in kwargs else self.DASHBOARD_HOST
- self.dashboard_port = kwargs.pop('dashboard_port') \
- if 'dashboard_port' in kwargs else self.DASHBOARD_PORT
+ cherrypy.tools.authenticate = cherrypy.Tool('before_handler', Auth.check_auth)
+ cherrypy.tools.session_expire_at_browser_close = SessionExpireAtBrowserCloseTool()
+ cherrypy.config.update({'error_page.default': json_error_page})
super(ControllerTestCase, self).__init__(*args, **kwargs)
- self._session = requests.Session()
- self._resp = None
def _request(self, url, method, data=None):
- url = "http://{}:{}{}".format(self.dashboard_host, self.dashboard_port,
- url)
- if method == 'GET':
- self._resp = self._session.get(url)
- return self._resp.json()
- elif method == 'POST':
- self._resp = self._session.post(url, json=data)
- elif method == 'DELETE':
- self._resp = self._session.delete(url, json=data)
- elif method == 'PUT':
- self._resp = self._session.put(url, json=data)
- return None
+ if not data:
+ b = None
+ h = None
+ else:
+ b = json.dumps(data)
+ h = [('Content-Type', 'application/json'),
+ ('Content-Length', str(len(b)))]
+ self.getPage(url, method=method, body=b, headers=h)
def _get(self, url):
- return self._request(url, 'GET')
+ self._request(url, 'GET')
def _post(self, url, data=None):
self._request(url, 'POST', data)
def _put(self, url, data=None):
self._request(url, 'PUT', data)
- def cookies(self):
- return self._resp.cookies
-
def jsonBody(self):
- return self._resp.json()
-
- def reset_session(self):
- self._session = requests.Session()
-
- def assertJsonBody(self, data):
- body = self._resp.json()
- self.assertEqual(body, data)
-
- def assertBody(self, body):
- self.assertEqual(self._resp.text, body)
-
- def assertStatus(self, status):
- self.assertEqual(self._resp.status_code, status)
+ body_str = self.body.decode('utf-8') if isinstance(self.body, bytes) else self.body
+ return json.loads(body_str)
+
+ def assertJsonBody(self, data, msg=None):
+ """Fail if value != self.body."""
+ json_body = self.jsonBody()
+ if data != json_body:
+ if msg is None:
+ msg = 'expected body:\n%r\n\nactual body:\n%r' % (
+ data, json_body)
+ self._handlewebError(msg)
+from __future__ import absolute_import
+
import json
import mock
import cherrypy
-from cherrypy.test.helper import CPWebCase
from .. import mgr
-from ..controllers.auth import Auth
from ..controllers.summary import Summary
from ..controllers.rbd_mirroring import RbdMirror
-from ..tools import SessionExpireAtBrowserCloseTool
from .helper import ControllerTestCase
}
-class RbdMirroringControllerTest(ControllerTestCase, CPWebCase):
+class RbdMirroringControllerTest(ControllerTestCase):
@classmethod
def setup_server(cls):
- # Initialize custom handlers.
- cherrypy.tools.authenticate = cherrypy.Tool('before_handler', Auth.check_auth)
- cherrypy.tools.session_expire_at_browser_close = SessionExpireAtBrowserCloseTool()
-
- cls._mgr_module = mock.Mock()
- cls.setup_test()
-
- @classmethod
- def setup_test(cls):
mgr.list_servers.return_value = mock_list_servers
mgr.get_metadata.return_value = mock_get_metadata
mgr.get_daemon_status.return_value = mock_get_daemon_status
cherrypy.tree.mount(RbdMirror(), '/api/test/rbdmirror')
cherrypy.tree.mount(Summary(), '/api/test/summary')
- def __init__(self, *args, **kwargs):
- super(RbdMirroringControllerTest, self).__init__(*args, dashboard_port=54583, **kwargs)
-
@mock.patch('dashboard_v2.controllers.rbd_mirroring.rbd')
def test_default(self, rbd_mock): # pylint: disable=W0613
self._get('/api/test/rbdmirror')
@mock.patch('dashboard_v2.controllers.rbd_mirroring.rbd')
def test_summary(self, rbd_mock): # pylint: disable=W0613
"""We're also testing `summary`, as it also uses code from `rbd_mirroring.py`"""
- data = self._get('/api/test/summary')
+ self._get('/api/test/summary')
self.assertStatus(200)
- summary = data['rbd_mirroring']
+
+ summary = self.jsonBody()['rbd_mirroring']
self.assertEqual(summary, {'errors': 0, 'warnings': 1})
-from cherrypy.test.helper import CPWebCase
+from __future__ import absolute_import
+
import cherrypy
-import mock
from .. import mgr
-from ..controllers.auth import Auth
-from ..tools import SessionExpireAtBrowserCloseTool
from ..controllers.tcmu_iscsi import TcmuIscsi
from .helper import ControllerTestCase
mocked_get_rate = 47
-class TcmuIscsiControllerTest(ControllerTestCase, CPWebCase):
+class TcmuIscsiControllerTest(ControllerTestCase):
@classmethod
def setup_server(cls):
- # Initialize custom handlers.
- cherrypy.tools.authenticate = cherrypy.Tool('before_handler', Auth.check_auth)
- cherrypy.tools.session_expire_at_browser_close = SessionExpireAtBrowserCloseTool()
-
- cls._mgr_module = mock.Mock()
- cls.setup_test()
-
- @classmethod
- def setup_test(cls):
mgr.list_servers.return_value = mocked_servers
mgr.get_metadata.return_value = mocked_metadata
mgr.get_daemon_status.return_value = mocked_get_daemon_status
cherrypy.tree.mount(TcmuIscsi(), "/api/test/tcmu")
- def __init__(self, *args, **kwargs):
- super(TcmuIscsiControllerTest, self).__init__(*args, dashboard_port=54583, **kwargs)
-
def test_list(self):
- self._post("/api/auth", {'username': 'admin', 'password': 'admin'})
self._get('/api/test/tcmu')
self.assertStatus(200)
self.assertJsonBody({
# -*- coding: utf-8 -*-
from __future__ import absolute_import
-import json
-
import cherrypy
from cherrypy.lib.sessions import RamSession
-from cherrypy.test import helper
from mock import patch
-from ..tools import RESTController, detail_route
+from .helper import ControllerTestCase
+from ..tools import RESTController
# pylint: disable=W0613
fooargs = FooArgs()
-class RESTControllerTest(helper.CPWebCase):
- def _request(self, url, method, data=None):
- if not data:
- b = None
- h = None
- else:
- b = json.dumps(data)
- h = [('Content-Type', 'application/json'),
- ('Content-Length', str(len(b)))]
- self.getPage(url, method=method, body=b, headers=h)
-
- def _get(self, url):
- self._request(url, 'GET')
-
- def _post(self, url, data=None):
- self._request(url, 'POST', data)
-
- def _delete(self, url, data=None):
- self._request(url, 'DELETE', data)
-
- def _put(self, url, data=None):
- self._request(url, 'PUT', data)
-
- def assertJsonBody(self, data, msg=None):
- """Fail if value != self.body."""
- body_str = self.body.decode('utf-8') if isinstance(self.body, bytes) else self.body
- json_body = json.loads(body_str)
- if data != json_body:
- if msg is None:
- msg = 'expected body:\n%r\n\nactual body:\n%r' % (
- data, json_body)
- self._handlewebError(msg)
+class RESTControllerTest(ControllerTestCase):
@classmethod
def setup_server(cls):
'tools.session_expire_at_browser_close.on': True
}
if not hasattr(cls, '_cp_config'):
- cls._cp_config = dict(cls._cp_config_default)
+ cls._cp_config = {}
+ if 'tools.authenticate.on' not in cls._cp_config:
config['tools.authenticate.on'] = False
- else:
- cls._cp_config.update(cls._cp_config_default)
- if 'tools.authenticate.on' not in cls._cp_config:
- config['tools.authenticate.on'] = False
cls._cp_config.update(config)
return cls
return decorate
def AuthRequired(enabled=True):
def decorate(cls):
if not hasattr(cls, '_cp_config'):
- cls._cp_config = dict(cls._cp_config_default)
cls._cp_config = {
'tools.authenticate.on': enabled
}
else:
- cls._cp_config.update(cls._cp_config_default)
cls._cp_config['tools.authenticate.on'] = enabled
return cls
return decorate
"""
Base class for all controllers providing API endpoints.
"""
- _cp_config_default = {
- 'request.error_page': {'default': json_error_page},
- }
# pylint: disable=too-many-instance-attributes
projects / ceph-ci.git / commitdiff