test/adapters/test_message_router.py

   1 import osdf.adapters.dcae.message_router as MR
   2 import unittest
   3
   4 from osdf.operation.exceptions import MessageBusConfigurationException
   5 from unittest.mock import patch
   6
   7
   8 class TestMessageRouter(unittest.TestCase):
   9
  10     def test_valid_MR(self):
  11         mr = MR.MessageRouterClient(dmaap_url="https://MYHOST:3905")
  12
  13     def test_valid_MR_with_base_urls(self):
  14         base_urls = ["https://MYHOST1:3905/","https://MYHOST2:3905/"]
  15         mr = MR.MessageRouterClient(mr_host_base_urls=base_urls, topic="MY-TOPIC")
  16
  17     def test_invalid_valid_MR_with_base_urls(self):
  18         """Topic missing"""
  19         base_urls = ["https://MYHOST1:3905/","https://MYHOST2:3905/"]
  20         try:
  21             mr = MR.MessageRouterClient(mr_host_base_urls=base_urls)
  22         except MessageBusConfigurationException:
  23             return
  24
  25         raise Exception("Allows invalid MR configuration") # if it failed to error out
  26
  27     @patch('osdf.adapters.dcae.message_router.MessageRouterClient.http_request', return_value={})
  28     def test_mr_http_request_mocked(self, http_request):
  29         mr = MR.MessageRouterClient(dmaap_url="https://MYHOST:3905")
  30         mr.http_request = http_request
  31         assert mr.get() == {}
  32         assert mr.post("Hello") == {}
  33
  34     def test_mr_http_request_non_existent_host(self):
  35         mr = MR.MessageRouterClient(dmaap_url="https://MYHOST:3905")
  36         try:
  37             mr.get()
  38         except:
  39             return
  40
  41         raise Exception("Allows invalid host") # if it failed to error out
  42 if __name__ == "__main__":
  43     unittest.main()
  44