import quincy beta 17.1.0

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

# -*- 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)