ceph/src/pybind/mgr/dashboard/tests/test_rest_tasks.py

   1 # -*- coding: utf-8 -*-
   2 # pylint: disable=blacklisted-name
   3
   4 import time
   5
   6 from . import ControllerTestCase
   7 from ..controllers import Controller, RESTController, Task
   8 from ..controllers.task import Task as TaskController
   9 from ..tools import NotificationQueue, TaskManager
  10
  11
  12 @Controller('/test/task', secure=False)
  13 class TaskTest(RESTController):
  14     sleep_time = 0.0
  15
  16     @Task('task/create', {'param': '{param}'}, wait_for=1.0)
  17     def create(self, param):
  18         time.sleep(TaskTest.sleep_time)
  19         return {'my_param': param}
  20
  21     @Task('task/set', {'param': '{2}'}, wait_for=1.0)
  22     def set(self, key, param=None):
  23         time.sleep(TaskTest.sleep_time)
  24         return {'key': key, 'my_param': param}
  25
  26     @Task('task/delete', ['{key}'], wait_for=1.0)
  27     def delete(self, key):
  28         # pylint: disable=unused-argument
  29         time.sleep(TaskTest.sleep_time)
  30
  31     @Task('task/foo', ['{param}'])
  32     @RESTController.Collection('POST')
  33     def foo(self, param):
  34         return {'my_param': param}
  35
  36     @Task('task/bar', ['{key}', '{param}'])
  37     @RESTController.Resource('PUT')
  38     def bar(self, key, param=None):
  39         return {'my_param': param, 'key': key}
  40
  41     @Task('task/query', ['{param}'])
  42     @RESTController.Collection('POST', query_params=['param'])
  43     def query(self, param=None):
  44         return {'my_param': param}
  45
  46
  47 class TaskControllerTest(ControllerTestCase):
  48     @classmethod
  49     def setup_server(cls):
  50         # pylint: disable=protected-access
  51         NotificationQueue.start_queue()
  52         TaskManager.init()
  53         TaskTest._cp_config['tools.authenticate.on'] = False
  54         TaskController._cp_config['tools.authenticate.on'] = False
  55         cls.setup_controllers([TaskTest, TaskController])
  56
  57     @classmethod
  58     def tearDownClass(cls):
  59         NotificationQueue.stop()
  60
  61     def setUp(self):
  62         TaskTest.sleep_time = 0.0
  63
  64     def test_create_task(self):
  65         self._task_post('/test/task', {'param': 'hello'})
  66         self.assertJsonBody({'my_param': 'hello'})
  67
  68     def test_long_set_task(self):
  69         TaskTest.sleep_time = 2.0
  70         self._task_put('/test/task/2', {'param': 'hello'})
  71         self.assertJsonBody({'key': '2', 'my_param': 'hello'})
  72
  73     def test_delete_task(self):
  74         self._task_delete('/test/task/hello')
  75
  76     def test_foo_task(self):
  77         self._task_post('/test/task/foo', {'param': 'hello'})
  78         self.assertJsonBody({'my_param': 'hello'})
  79
  80     def test_bar_task(self):
  81         self._task_put('/test/task/3/bar', {'param': 'hello'})
  82         self.assertJsonBody({'my_param': 'hello', 'key': '3'})
  83
  84     def test_query_param(self):
  85         self._task_post('/test/task/query')
  86         self.assertJsonBody({'my_param': None})