test/adapters/test_message_router.py

   1 # -------------------------------------------------------------------------
   2 #   Copyright (c) 2017-2018 AT&T Intellectual Property
   3 #
   4 #   Licensed under the Apache License, Version 2.0 (the "License");
   5 #   you may not use this file except in compliance with the License.
   6 #   You may obtain a copy of the License at
   7 #
   8 #       http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #   Unless required by applicable law or agreed to in writing, software
  11 #   distributed under the License is distributed on an "AS IS" BASIS,
  12 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 #   See the License for the specific language governing permissions and
  14 #   limitations under the License.
  15 #
  16 # -------------------------------------------------------------------------
  17 #
  18 import osdf.adapters.dcae.message_router as MR
  19 import unittest
  20
  21 from osdf.operation.exceptions import MessageBusConfigurationException
  22 from unittest.mock import patch
  23
  24
  25 class TestMessageRouter(unittest.TestCase):
  26
  27     def test_valid_MR(self):
  28         mr = MR.MessageRouterClient(dmaap_url="https://MYHOST:3905")
  29
  30     def test_valid_MR_with_base_urls(self):
  31         base_urls = ["https://MYHOST1:3905/","https://MYHOST2:3905/"]
  32         mr = MR.MessageRouterClient(mr_host_base_urls=base_urls, topic="MY-TOPIC")
  33
  34     def test_invalid_valid_MR_with_base_urls(self):
  35         """Topic missing"""
  36         base_urls = ["https://MYHOST1:3905/","https://MYHOST2:3905/"]
  37         try:
  38             mr = MR.MessageRouterClient(mr_host_base_urls=base_urls)
  39         except MessageBusConfigurationException:
  40             return
  41
  42         raise Exception("Allows invalid MR configuration") # if it failed to error out
  43
  44     @patch('osdf.adapters.dcae.message_router.MessageRouterClient.http_request', return_value={})
  45     def test_mr_http_request_mocked(self, http_request):
  46         mr = MR.MessageRouterClient(dmaap_url="https://MYHOST:3905")
  47         mr.http_request = http_request
  48         assert mr.get() == {}
  49         assert mr.post("Hello") == {}
  50
  51     def test_mr_http_request_non_existent_host(self):
  52         mr = MR.MessageRouterClient(dmaap_url="https://MYHOST:3905")
  53         try:
  54             mr.get()
  55         except:
  56             return
  57
  58         raise Exception("Allows invalid host") # if it failed to error out
  59 if __name__ == "__main__":
  60     unittest.main()
  61