"Fossies" - the Fresh Open Source Software Archive

Member "masakari-9.0.0/masakari/tests/unit/engine/test_rpcapi.py" (13 May 2020, 3769 Bytes) of package /linux/misc/openstack/masakari-9.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_rpcapi.py": 8.0.0_vs_9.0.0.

    1 #    Copyright (c) 2016 NTT DATA
    2 #
    3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #    not use this file except in compliance with the License. You may obtain
    5 #    a copy of the License at
    6 #
    7 #         http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 #    Unless required by applicable law or agreed to in writing, software
   10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #    License for the specific language governing permissions and limitations
   13 #    under the License.
   14 
   15 """
   16 Unit Tests for masakari.engine.rpcapi
   17 """
   18 
   19 import copy
   20 from unittest import mock
   21 
   22 
   23 from masakari import context
   24 from masakari.engine import rpcapi as engine_rpcapi
   25 from masakari import objects
   26 from masakari import test
   27 from masakari.tests.unit.engine import fake_engine
   28 
   29 
   30 class EngineRpcAPITestCase(test.TestCase):
   31     def setUp(self):
   32         super(EngineRpcAPITestCase, self).setUp()
   33         self.context = context.RequestContext()
   34         self.fake_notification_obj = fake_engine.fake_notification_obj(
   35             self.context)
   36 
   37     def _test_engine_api(self, method, rpc_method, server=None, fanout=False,
   38                          **kwargs):
   39         rpcapi = engine_rpcapi.EngineAPI()
   40         expected_retval = 'foo' if rpc_method == 'call' else None
   41 
   42         target = {
   43             "server": server,
   44             "fanout": fanout,
   45             "version": kwargs.pop('version', rpcapi.RPC_API_VERSION)
   46         }
   47 
   48         expected_msg = copy.deepcopy(kwargs)
   49 
   50         self.fake_args = None
   51         self.fake_kwargs = None
   52 
   53         def _fake_prepare_method(*args, **kwds):
   54             for kwd in kwds:
   55                 self.assertEqual(target[kwd], kwds[kwd])
   56             return rpcapi.client
   57 
   58         def _fake_rpc_method(*args, **kwargs):
   59             self.fake_args = args
   60             self.fake_kwargs = kwargs
   61             if expected_retval:
   62                 return expected_retval
   63 
   64         with mock.patch.object(rpcapi.client, "prepare") as mock_prepared:
   65             mock_prepared.side_effect = _fake_prepare_method
   66 
   67             with mock.patch.object(rpcapi.client, rpc_method) as mock_method:
   68                 mock_method.side_effect = _fake_rpc_method
   69                 retval = getattr(rpcapi, method)(self.context, **kwargs)
   70                 self.assertEqual(expected_retval, retval)
   71                 expected_args = [self.context, method, expected_msg]
   72                 for arg, expected_arg in zip(self.fake_args, expected_args):
   73                     self.assertEqual(expected_arg, arg)
   74 
   75                 for kwarg, value in self.fake_kwargs.items():
   76                     if isinstance(value, objects.Notification):
   77                         expected_back = expected_msg[kwarg].obj_to_primitive()
   78                         backup = value.obj_to_primitive()
   79                         self.assertEqual(expected_back, backup)
   80                     else:
   81                         self.assertEqual(expected_msg[kwarg], value)
   82 
   83     @mock.patch("masakari.rpc.get_client")
   84     def test_process_notification(self, mock_get_client):
   85         self._test_engine_api('process_notification',
   86                               rpc_method='cast',
   87                               notification=self.fake_notification_obj,
   88                               version='1.0')
   89 
   90     @mock.patch("masakari.rpc.get_client")
   91     def test_get_notification_recovery_workflow_details(self,
   92                                                         mock_get_client):
   93         self._test_engine_api('get_notification_recovery_workflow_details',
   94                               rpc_method='call',
   95                               notification=self.fake_notification_obj,
   96                               version='1.1')