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 import pytest
  22
  23 from osm_common.msgbase import MsgBase, MsgException
  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(exception_message("Method 'write' not implemented"))
  53     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  54
  55
  56 def test_read(msg_base):
  57     with pytest.raises(MsgException) as excinfo:
  58         msg_base.read("test")
  59     assert str(excinfo.value).startswith(exception_message("Method 'read' not implemented"))
  60     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  61
  62
  63 def test_aiowrite(msg_base, event_loop):
  64     with pytest.raises(MsgException) as excinfo:
  65         event_loop.run_until_complete(msg_base.aiowrite("test", "test", "test", event_loop))
  66     assert str(excinfo.value).startswith(exception_message("Method 'aiowrite' not implemented"))
  67     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR
  68
  69
  70 def test_aioread(msg_base, event_loop):
  71     with pytest.raises(MsgException) as excinfo:
  72         event_loop.run_until_complete(msg_base.aioread("test", event_loop))
  73     assert str(excinfo.value).startswith(exception_message("Method 'aioread' not implemented"))
  74     assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR