1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Whitestack, LLC
4 # *************************************************************
6 # This file is part of OSM Monitoring module
7 # All Rights Reserved to Whitestack, LLC
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
21 # For those usages not covered by the Apache License, Version 2.0 please
22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
25 from unittest
import TestCase
, mock
27 from osm_common
.msgkafka
import MsgKafka
29 from osm_mon
.core
.message_bus_client
import MessageBusClient
30 from osm_mon
.core
.config
import Config
33 class TestMessageBusClient(TestCase
):
35 self
.config
= Config()
36 self
.config
.set("message", "driver", "kafka")
38 @mock.patch
.object(MsgKafka
, "aioread")
39 def test_aioread(self
, aioread
):
40 async def mock_callback():
43 future
= asyncio
.Future(loop
=asyncio
.new_event_loop())
44 future
.set_result("mock")
45 aioread
.return_value
= future
46 msg_bus
= MessageBusClient(self
.config
)
48 asyncio
.run(msg_bus
.aioread([topic
], mock_callback
))
49 aioread
.assert_called_with(["test_topic"], aiocallback
=mock_callback
)
51 @mock.patch
.object(MsgKafka
, "aiowrite")
52 def test_aiowrite(self
, aiowrite
):
53 future
= asyncio
.Future(loop
=asyncio
.new_event_loop())
54 future
.set_result("mock")
55 aiowrite
.return_value
= future
56 msg_bus
= MessageBusClient(self
.config
)
59 msg
= {"test": "test_msg"}
60 asyncio
.run(msg_bus
.aiowrite(topic
, key
, msg
))
61 aiowrite
.assert_called_with(topic
, key
, msg
)
63 @mock.patch
.object(MsgKafka
, "aioread")
64 def test_aioread_once(self
, aioread
):
65 future
= asyncio
.Future(loop
=asyncio
.new_event_loop())
66 future
.set_result("mock")
67 aioread
.return_value
= future
68 msg_bus
= MessageBusClient(self
.config
)
70 asyncio
.run(msg_bus
.aioread_once(topic
))
71 aioread
.assert_called_with("test_topic")