osm_common/tests/test_msgbase.py

   1 # Copyright 2018 Whitestack, LLC
   2 # Copyright 2018 Telefonica S.A.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # 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, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com
  18 ##
  19
  20 import http
  21
  22 from osm_common.msgbase import MsgBase, MsgException
  23 import pytest
  24
  25
  26 def exception_message(message):
  27     return "messaging exception " + message
  28
  29
  30 @pytest.fixture
  31 def msg_base():
  32     return MsgBase()
  33
  34
  35 def test_constructor():
  36     msgbase = MsgBase()
  37     assert msgbase is not None
  38     assert isinstance(msgbase, MsgBase)
  39
  40
  41 def test_connect(msg_base):
  42     msg_base.connect(None)
  43
  44
  45 def test_disconnect(msg_base):
  46     msg_base.disconnect()
  47
  48
  49 def test_write(msg_base):
  50     with pytest.raises(MsgException) as excinfo:
  51         msg_base.write("test", "test", "test")
  52     assert str(excinfo.value).startswith(
  53         exception_message("Method 'write' not implemented")
  54     )
  55     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  56
  57
  58 def test_read(msg_base):
  59     with pytest.raises(MsgException) as excinfo:
  60         msg_base.read("test")
  61     assert str(excinfo.value).startswith(
  62         exception_message("Method 'read' not implemented")
  63     )
  64     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  65
  66
  67 def test_aiowrite(msg_base, event_loop):
  68     with pytest.raises(MsgException) as excinfo:
  69         event_loop.run_until_complete(
  70             msg_base.aiowrite("test", "test", "test", event_loop)
  71         )
  72     assert str(excinfo.value).startswith(
  73         exception_message("Method 'aiowrite' not implemented")
  74     )
  75     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  76
  77
  78 def test_aioread(msg_base, event_loop):
  79     with pytest.raises(MsgException) as excinfo:
  80         event_loop.run_until_complete(msg_base.aioread("test", event_loop))
  81     assert str(excinfo.value).startswith(
  82         exception_message("Method 'aioread' not implemented")
  83     )
  84     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR