4 from osm_common
.msgbase
import MsgBase
, MsgException
7 def exception_message(message
):
8 return "messaging exception " + message
16 def test_constructor():
18 assert msgbase
is not None
19 assert isinstance(msgbase
, MsgBase
)
22 def test_connect(msg_base
):
23 msg_base
.connect(None)
26 def test_disconnect(msg_base
):
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
37 def test_read(msg_base
):
38 with pytest
.raises(MsgException
) as excinfo
:
40 assert str(excinfo
.value
).startswith(exception_message("Method 'read' not implemented"))
41 assert excinfo
.value
.http_code
== http
.HTTPStatus
.INTERNAL_SERVER_ERROR
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
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