import 15.2.0 Octopus source

[ceph.git] / ceph / src / pybind / mgr / dashboard / tests / __init__.py

diff --git a/ceph/src/pybind/mgr/dashboard/tests/__init__.py b/ceph/src/pybind/mgr/dashboard/tests/__init__.py
index 689e0097859ac8bbc55dc1daef2acec15863208e..421378a155d80633dbb44e8cb3fc2322fa904203 100644 (file)
--- a/ceph/src/pybind/mgr/dashboard/tests/__init__.py
+++ b/ceph/src/pybind/mgr/dashboard/tests/__init__.py
@@ -3,16 +3,19 @@
 from __future__ import absolute_import
 
 import json
+import logging
 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
+from mgr_module import CLICommand
 
-from .. import logger, mgr
+from .. import mgr
 from ..controllers import json_error_page, generate_controller_routes
 from ..services.auth import AuthManagerTool
 from ..services.exception import dashboard_exception_handler
@@ -25,6 +28,9 @@ PLUGIN_MANAGER.hook.init()
 PLUGIN_MANAGER.hook.register_commands()
 
 
+logger = logging.getLogger('tests')
+
+
 class CmdException(Exception):
     def __init__(self, retcode, message):
         super(CmdException, self).__init__(message)
@@ -84,6 +90,17 @@ class CLICommandTestMixin(KVStoreMockMixin):
         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 = {}
 
@@ -158,7 +175,7 @@ class ControllerTestCase(helper.CPWebCase):
             logger.info("task finished immediately")
             return
 
-        res = self.jsonBody()
+        res = self.json_body()
         self.assertIsInstance(res, dict)
         self.assertIn('name', res)
         self.assertIn('metadata', res)
@@ -184,7 +201,7 @@ class ControllerTestCase(helper.CPWebCase):
                                 self.task_metadata)
                     time.sleep(1)
                     self.tc._get('/api/task?name={}'.format(self.task_name))
-                    res = self.tc.jsonBody()
+                    res = self.tc.json_body()
                     for task in res['finished_tasks']:
                         if task['metadata'] == self.task_metadata:
                             # task finished
@@ -211,13 +228,12 @@ class ControllerTestCase(helper.CPWebCase):
             elif method == 'DELETE':
                 self.status = '204 No Content'
             return
+
+        if 'status' in thread.res_task['exception']:
+            self.status = thread.res_task['exception']['status']
         else:
-            if 'status' in thread.res_task['exception']:
-                self.status = thread.res_task['exception']['status']
-            else:
-                self.status = 500
-            self.body = json.dumps(thread.res_task['exception'])
-            return
+            self.status = 500
+        self.body = json.dumps(thread.res_task['exception'])
 
     def _task_post(self, url, data=None, timeout=60):
         self._task_request('POST', url, data, timeout)
@@ -228,21 +244,21 @@ class ControllerTestCase(helper.CPWebCase):
     def _task_put(self, url, data=None, timeout=60):
         self._task_request('PUT', url, data, timeout)
 
-    def jsonBody(self):
+    def json_body(self):
         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):
+    def assertJsonBody(self, data, msg=None):  # noqa: N802
         """Fail if value != self.body."""
-        json_body = self.jsonBody()
+        json_body = self.json_body()
         if data != json_body:
             if msg is None:
                 msg = 'expected body:\n%r\n\nactual body:\n%r' % (
                     data, json_body)
             self._handlewebError(msg)
 
-    def assertInJsonBody(self, data, msg=None):
-        json_body = self.jsonBody()
+    def assertInJsonBody(self, data, msg=None):  # noqa: N802
+        json_body = self.json_body()
         if data not in json_body:
             if msg is None:
                 msg = 'expected %r to be in %r' % (data, json_body)