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
73
74
75
76
77
78
79
80
81
82
83
84
85
|
# 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.
"""
test_message_process
--------------------
Check that messages are processed correctly
"""
from unittest import mock
from oslo_metrics import message_router
from oslotest import base
import prometheus_client
class TestProcessMessage(base.BaseTestCase):
def setUp(self):
super().setUp()
def test_process_counter(self):
received_json = b"""{
"module": "oslo_messaging",
"name": "rpc_server_invocation_start_total",
"action": {
"action": "inc",
"value": null
},
"labels": {
"exchange": "foo",
"topic": "bar",
"server": "foobar",
"endpoint": "endpoint",
"namespace": "ns",
"version": "v2",
"method": "get",
"process": "done"
}
}"""
with mock.patch.object(
prometheus_client.Counter, 'inc',
) as mock_inc:
router = message_router.MessageRouter()
router.process(received_json)
mock_inc.assert_called_once_with()
def test_process_histogram(self):
received_json = b"""{
"module": "oslo_messaging",
"name": "rpc_client_processing_seconds",
"action": {
"action": "observe",
"value": 1.26
},
"labels": {
"call_type": "call",
"exchange": "foo",
"topic": "bar",
"method": "get",
"server": "foobar",
"namespace": "ns",
"version": "v2",
"process": "done",
"fanout": "foo",
"timeout": 10
}
}"""
with mock.patch.object(
prometheus_client.Histogram, 'observe',
) as mock_inc:
router = message_router.MessageRouter()
router.process(received_json)
mock_inc.assert_called_once_with(1.26)
|
