[Jenkins]Jenkins

Code Coverage

Cobertura Coverage Report > osm_policy_module.tests.unit.common >

test_message_bus_client.py

Trend

Classes100%
 
Lines98%
   
Conditionals100%
 

File Coverage summary

NameClassesLinesConditionals
test_message_bus_client.py
100%
1/1
98%
39/40
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
test_message_bus_client.py
98%
39/40
N/A

Source

osm_policy_module/tests/unit/common/test_message_bus_client.py

1

# -*- coding: utf-8 -*-




2






3

# Copyright 2018 Whitestack, LLC




4

# *************************************************************




5






6

# This file is part of OSM Monitoring module




7

# All Rights Reserved to Whitestack, LLC




8






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




12






13

#         http://www.apache.org/licenses/LICENSE-2.0




14






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




19

# under the License.




20






21

# For those usages not covered by the Apache License, Version 2.0 please




22

# contact: bdiaz@whitestack.com or glavado@whitestack.com




23

##




24
1
import asyncio




25
1
from unittest import TestCase, mock




26






27
1
from osm_common.msgkafka import MsgKafka




28






29
1
from osm_policy_module.common.message_bus_client import MessageBusClient




30
1
from osm_policy_module.core.config import Config




31






32






33
1
class TestMessageBusClient(TestCase):




34
1
    def setUp(self):




35
1
        self.config = Config()




36
1
        self.config.set("message", "driver", "kafka")




37






38
1
    @mock.patch.object(MsgKafka, "aioread")




39
1
    def test_aioread(self, aioread):




40
1
        async def mock_callback():




41
0
            pass




42






43
1
        future = asyncio.Future(loop=asyncio.new_event_loop())




44
1
        future.set_result("mock")




45
1
        aioread.return_value = future




46
1
        msg_bus = MessageBusClient(self.config)




47
1
        topic = "test_topic"




48
1
        asyncio.run(msg_bus.aioread([topic], mock_callback))




49
1
        aioread.assert_called_with(["test_topic"], aiocallback=mock_callback)




50






51
1
    @mock.patch.object(MsgKafka, "aiowrite")




52
1
    def test_aiowrite(self, aiowrite):




53
1
        future = asyncio.Future(loop=asyncio.new_event_loop())




54
1
        future.set_result("mock")




55
1
        aiowrite.return_value = future




56
1
        msg_bus = MessageBusClient(self.config)




57
1
        topic = "test_topic"




58
1
        key = "test_key"




59
1
        msg = {"test": "test_msg"}




60
1
        asyncio.run(msg_bus.aiowrite(topic, key, msg))




61
1
        aiowrite.assert_called_with(topic, key, msg)




62






63
1
    @mock.patch.object(MsgKafka, "aioread")




64
1
    def test_aioread_once(self, aioread):




65
1
        future = asyncio.Future(loop=asyncio.new_event_loop())




66
1
        future.set_result("mock")




67
1
        aioread.return_value = future




68
1
        msg_bus = MessageBusClient(self.config)




69
1
        topic = "test_topic"




70
1
        asyncio.run(msg_bus.aioread_once(topic))




71
1
        aioread.assert_called_with("test_topic")