From e6918187568dbd01842d8d1d2c808ce16a894239 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 21 Apr 2024 13:54:28 +0200 Subject: Adding upstream version 18.2.2. Signed-off-by: Daniel Baumann --- src/pybind/mgr/dashboard/tests/test_tools.py | 210 +++++++++++++++++++++++++++ 1 file changed, 210 insertions(+) create mode 100644 src/pybind/mgr/dashboard/tests/test_tools.py (limited to 'src/pybind/mgr/dashboard/tests/test_tools.py') diff --git a/src/pybind/mgr/dashboard/tests/test_tools.py b/src/pybind/mgr/dashboard/tests/test_tools.py new file mode 100644 index 000000000..29f6123f3 --- /dev/null +++ b/src/pybind/mgr/dashboard/tests/test_tools.py @@ -0,0 +1,210 @@ +# -*- coding: utf-8 -*- + +import unittest + +import cherrypy +from cherrypy.lib.sessions import RamSession + +try: + from mock import patch +except ImportError: + from unittest.mock import patch + +from ..controllers import APIRouter, BaseController, Proxy, RESTController, Router +from ..controllers._version import APIVersion +from ..services.exception import handle_rados_error +from ..tests import ControllerTestCase +from ..tools import dict_contains_path, dict_get, json_str_to_object, \ + merge_list_of_dicts_by_key, partial_dict + + +# pylint: disable=W0613 +@Router('/foo', secure=False) +class FooResource(RESTController): + elems = [] + + def list(self): + return FooResource.elems + + def create(self, a): + FooResource.elems.append({'a': a}) + return {'a': a} + + def get(self, key): + return {'detail': (key, [])} + + def delete(self, key): + del FooResource.elems[int(key)] + + def bulk_delete(self): + FooResource.elems = [] + + def set(self, key, newdata): + FooResource.elems[int(key)] = {'newdata': newdata} + return dict(key=key, newdata=newdata) + + +@Router('/foo/:key/:method', secure=False) +class FooResourceDetail(RESTController): + def list(self, key, method): + return {'detail': (key, [method])} + + +@APIRouter('/rgw/proxy', secure=False) +class GenerateControllerRoutesController(BaseController): + @Proxy() + def __call__(self, path, **params): + pass + + +@APIRouter('/fooargs', secure=False) +class FooArgs(RESTController): + def set(self, code, name=None, opt1=None, opt2=None): + return {'code': code, 'name': name, 'opt1': opt1, 'opt2': opt2} + + @handle_rados_error('foo') + def create(self, my_arg_name): + return my_arg_name + + def list(self): + raise cherrypy.NotFound() + + +class Root(object): + foo_resource = FooResource() + fooargs = FooArgs() + + +class RESTControllerTest(ControllerTestCase): + + @classmethod + def setup_server(cls): + cls.setup_controllers( + [FooResource, FooResourceDetail, FooArgs, GenerateControllerRoutesController]) + + def test_empty(self): + self._delete("/foo") + self.assertStatus(204) + self._get("/foo") + self.assertStatus('200 OK') + self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type()) + self.assertBody('[]') + + def test_fill(self): + sess_mock = RamSession() + with patch('cherrypy.session', sess_mock, create=True): + data = {'a': 'b'} + for _ in range(5): + self._post("/foo", data) + self.assertJsonBody(data) + self.assertStatus(201) + self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type()) + + self._get("/foo") + self.assertStatus('200 OK') + self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type()) + self.assertJsonBody([data] * 5) + + self._put('/foo/0', {'newdata': 'newdata'}) + self.assertStatus('200 OK') + self.assertHeader('Content-Type', APIVersion.DEFAULT.to_mime_type()) + self.assertJsonBody({'newdata': 'newdata', 'key': '0'}) + + def test_not_implemented(self): + self._put("/foo") + self.assertStatus(404) + body = self.json_body() + self.assertIsInstance(body, dict) + assert body['detail'] == "The path '/foo' was not found." + assert '404' in body['status'] + + def test_args_from_json(self): + self._put("/api/fooargs/hello", {'name': 'world'}) + self.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': None, 'opt2': None}) + + self._put("/api/fooargs/hello", {'name': 'world', 'opt1': 'opt1'}) + self.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': 'opt1', 'opt2': None}) + + self._put("/api/fooargs/hello", {'name': 'world', 'opt2': 'opt2'}) + self.assertJsonBody({'code': 'hello', 'name': 'world', 'opt1': None, 'opt2': 'opt2'}) + + def test_detail_route(self): + self._get('/foo/default') + self.assertJsonBody({'detail': ['default', []]}) + + self._get('/foo/default/default') + self.assertJsonBody({'detail': ['default', ['default']]}) + + self._get('/foo/1/detail') + self.assertJsonBody({'detail': ['1', ['detail']]}) + + self._post('/foo/1/detail', 'post-data') + self.assertStatus(404) + + def test_generate_controller_routes(self): + # We just need to add this controller in setup_server(): + # noinspection PyStatementEffect + # pylint: disable=pointless-statement + GenerateControllerRoutesController + + +class RequestLoggingToolTest(ControllerTestCase): + + _request_logging = True + + @classmethod + def setup_server(cls): + cls.setup_controllers([FooResource]) + + def test_is_logged(self): + with patch('logging.Logger.debug') as mock_logger_debug: + self._put('/foo/0', {'newdata': 'xyz'}) + self.assertStatus(200) + call_args_list = mock_logger_debug.call_args_list + _, host, _, method, user, path = call_args_list[0][0] + self.assertEqual(host, '127.0.0.1') + self.assertEqual(method, 'PUT') + self.assertIsNone(user) + self.assertEqual(path, '/foo/0') + + +class TestFunctions(unittest.TestCase): + + def test_dict_contains_path(self): + x = {'a': {'b': {'c': 'foo'}}} + self.assertTrue(dict_contains_path(x, ['a', 'b', 'c'])) + self.assertTrue(dict_contains_path(x, ['a', 'b', 'c'])) + self.assertTrue(dict_contains_path(x, ['a'])) + self.assertFalse(dict_contains_path(x, ['a', 'c'])) + self.assertTrue(dict_contains_path(x, [])) + + def test_json_str_to_object(self): + expected_result = {'a': 1, 'b': 'bbb'} + self.assertEqual(expected_result, json_str_to_object('{"a": 1, "b": "bbb"}')) + self.assertEqual(expected_result, json_str_to_object(b'{"a": 1, "b": "bbb"}')) + self.assertEqual('', json_str_to_object('')) + self.assertRaises(TypeError, json_str_to_object, None) + + def test_partial_dict(self): + expected_result = {'a': 1, 'c': 3} + self.assertEqual(expected_result, partial_dict({'a': 1, 'b': 2, 'c': 3}, ['a', 'c'])) + self.assertEqual({}, partial_dict({'a': 1, 'b': 2, 'c': 3}, [])) + self.assertEqual({}, partial_dict({}, [])) + self.assertRaises(KeyError, partial_dict, {'a': 1, 'b': 2, 'c': 3}, ['d']) + self.assertRaises(TypeError, partial_dict, None, ['a']) + self.assertRaises(TypeError, partial_dict, {'a': 1, 'b': 2, 'c': 3}, None) + + def test_dict_get(self): + self.assertFalse(dict_get({'foo': {'bar': False}}, 'foo.bar')) + self.assertIsNone(dict_get({'foo': {'bar': False}}, 'foo.bar.baz')) + self.assertEqual(dict_get({'foo': {'bar': False}, 'baz': 'xyz'}, 'baz'), 'xyz') + + def test_merge_list_of_dicts_by_key(self): + expected_result = [{'a': 1, 'b': 2, 'c': 3}, {'a': 4, 'b': 5, 'c': 6}] + self.assertEqual(expected_result, merge_list_of_dicts_by_key( + [{'a': 1, 'b': 2}, {'a': 4, 'b': 5}], [{'a': 1, 'c': 3}, {'a': 4, 'c': 6}], 'a')) + + expected_result = [{'a': 1, 'b': 2}, {'a': 4, 'b': 5, 'c': 6}] + self.assertEqual(expected_result, merge_list_of_dicts_by_key( + [{'a': 1, 'b': 2}, {'a': 4, 'b': 5}], [{}, {'a': 4, 'c': 6}], 'a')) + self.assertRaises(TypeError, merge_list_of_dicts_by_key, None) -- cgit v1.2.3