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_policy_module
.common
.message_bus_client
import MessageBusClient
30 from osm_policy_module
.core
.config
import Config
33 class TestMessageBusClient(TestCase
):
35 self
.config
= Config()
36 self
.config
.set("message", "driver", "kafka")
37 self
.loop
= asyncio
.new_event_loop()
39 @mock.patch
.object(MsgKafka
, "aioread")
40 def test_aioread(self
, aioread
):
41 async def mock_callback():
44 future
= asyncio
.Future(loop
=self
.loop
)
45 future
.set_result("mock")
46 aioread
.return_value
= future
47 msg_bus
= MessageBusClient(self
.config
, loop
=self
.loop
)
49 self
.loop
.run_until_complete(msg_bus
.aioread([topic
], mock_callback
))
50 aioread
.assert_called_with(["test_topic"], self
.loop
, aiocallback
=mock_callback
)
52 @mock.patch
.object(MsgKafka
, "aiowrite")
53 def test_aiowrite(self
, aiowrite
):
54 future
= asyncio
.Future(loop
=self
.loop
)
55 future
.set_result("mock")
56 aiowrite
.return_value
= future
57 msg_bus
= MessageBusClient(self
.config
, loop
=self
.loop
)
60 msg
= {"test": "test_msg"}
61 self
.loop
.run_until_complete(msg_bus
.aiowrite(topic
, key
, msg
))
62 aiowrite
.assert_called_with(topic
, key
, msg
, self
.loop
)
64 @mock.patch
.object(MsgKafka
, "aioread")
65 def test_aioread_once(self
, aioread
):
66 future
= asyncio
.Future(loop
=self
.loop
)
67 future
.set_result("mock")
68 aioread
.return_value
= future
69 msg_bus
= MessageBusClient(self
.config
, loop
=self
.loop
)
71 self
.loop
.run_until_complete(msg_bus
.aioread_once(topic
))
72 aioread
.assert_called_with("test_topic", self
.loop
)