import quincy beta 17.1.0

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

diff --git a/ceph/src/pybind/mgr/dashboard/tests/test_pool.py b/ceph/src/pybind/mgr/dashboard/tests/test_pool.py
index 02e2b641ca470d5f1a253a90751553289bcba5d6..6f87e955d8fe69fefdc9050bd0a2292a3b6f1483 100644 (file)
--- a/ceph/src/pybind/mgr/dashboard/tests/test_pool.py
+++ b/ceph/src/pybind/mgr/dashboard/tests/test_pool.py
@@ -7,6 +7,7 @@ try:
 except ImportError:
     import unittest.mock as mock
 
+from .. import mgr
 from ..controllers.pool import Pool
 from ..controllers.task import Task
 from ..tests import ControllerTestCase
@@ -119,3 +120,61 @@ class PoolControllerTest(ControllerTestCase):
         self.assertEqual(_get.call_count, 6)
         self.assertEqual(task.percentages, [0, 5, 50, 73, 98])
         TaskManager.current_task = orig_method
+
+    @mock.patch('dashboard.controllers.osd.CephService.get_pool_list_with_stats')
+    @mock.patch('dashboard.controllers.osd.CephService.get_pool_list')
+    def test_pool_list(self, get_pool_list, get_pool_list_with_stats):
+        get_pool_list.return_value = [{
+            'type': 3,
+            'crush_rule': 1,
+            'application_metadata': {
+                'test_key': 'test_metadata'
+            },
+            'pool_name': 'test_name'
+        }]
+        mgr.get.side_effect = lambda key: {
+            'osd_map_crush': {
+                'rules': [{
+                    'rule_id': 1,
+                    'rule_name': 'test-rule'
+                }]
+            }
+        }[key]
+        Pool._pool_list()
+        mgr.get.assert_called_with('osd_map_crush')
+        self.assertEqual(get_pool_list.call_count, 1)
+        # with stats
+        get_pool_list_with_stats.return_value = get_pool_list.return_value
+        Pool._pool_list(attrs='type', stats='True')
+        self.assertEqual(get_pool_list_with_stats.call_count, 1)
+
+    @mock.patch('dashboard.controllers.pool.Pool._get')
+    @mock.patch('dashboard.services.ceph_service.CephService.send_command')
+    def test_set_pool_name(self, send_command, _get):
+        _get.return_value = {
+            'options': {
+                'compression_min_blob_size': '1'
+            },
+            'application_metadata': ['data1', 'data2']
+        }
+
+        def _send_cmd(*args, **kwargs):  # pylint: disable=unused-argument
+            pass
+
+        send_command.side_effect = _send_cmd
+        NotificationQueue.start_queue()
+        TaskManager.init()
+        self._task_put('/api/pool/test-pool', {
+            "flags": "ec_overwrites",
+            "application_metadata": ['data3', 'data2'],
+            "configuration": "test-conf",
+            "compression_mode": 'unset',
+            'compression_min_blob_size': '1',
+            'compression_max_blob_size': '1',
+            'compression_required_ratio': '1',
+            'pool': 'test-pool',
+            'pg_num': 64
+        })
+        NotificationQueue.stop()
+        self.assertEqual(_get.call_count, 1)
+        self.assertEqual(send_command.call_count, 10)