osm_common/tests/test_msgbase.py

   1 import http
   2 import pytest
   3
   4 from osm_common.msgbase import MsgBase, MsgException
   5
   6
   7 def exception_message(message):
   8     return "messaging exception " + message
   9
  10
  11 @pytest.fixture
  12 def msg_base():
  13     return MsgBase()
  14
  15
  16 def test_constructor():
  17     msgbase = MsgBase()
  18     assert msgbase is not None
  19     assert isinstance(msgbase, MsgBase)
  20
  21
  22 def test_connect(msg_base):
  23     msg_base.connect(None)
  24
  25
  26 def test_disconnect(msg_base):
  27     msg_base.disconnect()
  28
  29
  30 def test_write(msg_base):
  31     with pytest.raises(MsgException) as excinfo:
  32         msg_base.write("test", "test", "test")
  33     assert str(excinfo.value).startswith(exception_message("Method 'write' not implemented"))
  34     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  35
  36
  37 def test_read(msg_base):
  38     with pytest.raises(MsgException) as excinfo:
  39         msg_base.read("test")
  40     assert str(excinfo.value).startswith(exception_message("Method 'read' not implemented"))
  41     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  42
  43
  44 def test_aiowrite(msg_base, event_loop):
  45     with pytest.raises(MsgException) as excinfo:
  46         event_loop.run_until_complete(msg_base.aiowrite("test", "test", "test", event_loop))
  47     assert str(excinfo.value).startswith(exception_message("Method 'aiowrite' not implemented"))
  48     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  49
  50
  51 def test_aioread(msg_base, event_loop):
  52     with pytest.raises(MsgException) as excinfo:
  53         event_loop.run_until_complete(msg_base.aioread("test", event_loop))
  54     assert str(excinfo.value).startswith(exception_message("Method 'aioread' not implemented"))
  55     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR