1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
# Copyright 2016 Mirantis Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from osprofiler.drivers import base
from osprofiler.tests import test
class MessagingTestCase(test.TestCase):
@mock.patch("oslo_utils.importutils.try_import")
def test_init_no_oslo_messaging(self, try_import_mock):
try_import_mock.return_value = None
self.assertRaises(
ValueError, base.get_driver,
"messaging://", project="project", service="service",
host="host", context={})
@mock.patch("oslo_utils.importutils.try_import")
def test_init_and_notify(self, try_import_mock):
context = "context"
transport = "transport"
project = "project"
service = "service"
host = "host"
# emulate dynamic load of oslo.messaging library
oslo_messaging_mock = mock.Mock()
try_import_mock.return_value = oslo_messaging_mock
# mock oslo.messaging APIs
notifier_mock = mock.Mock()
oslo_messaging_mock.Notifier.return_value = notifier_mock
oslo_messaging_mock.get_notification_transport.return_value = transport
notify_func = base.get_driver(
"messaging://", project=project, service=service,
context=context, host=host).notify
oslo_messaging_mock.Notifier.assert_called_once_with(
transport, publisher_id=host, driver="messaging",
topics=["profiler"], retry=0)
info = {
"a": 10,
"project": project,
"service": service,
"host": host
}
notify_func(info)
notifier_mock.info.assert_called_once_with(
context, "profiler.service", info)
notifier_mock.reset_mock()
notify_func(info, context="my_context")
notifier_mock.info.assert_called_once_with(
"my_context", "profiler.service", info)
|
