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 |
1 |
import asyncio |
20 |
1 |
import http |
21 |
|
|
22 |
1 |
from osm_common.msgbase import MsgBase, MsgException |
23 |
1 |
import pytest |
24 |
|
|
25 |
|
|
26 |
1 |
def exception_message(message): |
27 |
0 |
return "messaging exception " + message |
28 |
|
|
29 |
|
|
30 |
1 |
@pytest.fixture |
31 |
1 |
def msg_base(): |
32 |
0 |
return MsgBase() |
33 |
|
|
34 |
|
|
35 |
1 |
def test_constructor(): |
36 |
1 |
msgbase = MsgBase() |
37 |
1 |
assert msgbase is not None |
38 |
1 |
assert isinstance(msgbase, MsgBase) |
39 |
|
|
40 |
|
|
41 |
1 |
def test_connect(msg_base): |
42 |
0 |
msg_base.connect(None) |
43 |
|
|
44 |
|
|
45 |
1 |
def test_disconnect(msg_base): |
46 |
0 |
msg_base.disconnect() |
47 |
|
|
48 |
|
|
49 |
1 |
def test_write(msg_base): |
50 |
0 |
with pytest.raises(MsgException) as excinfo: |
51 |
0 |
msg_base.write("test", "test", "test") |
52 |
0 |
assert str(excinfo.value).startswith( |
53 |
|
exception_message("Method 'write' not implemented") |
54 |
|
) |
55 |
0 |
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR |
56 |
|
|
57 |
|
|
58 |
1 |
def test_read(msg_base): |
59 |
0 |
with pytest.raises(MsgException) as excinfo: |
60 |
0 |
msg_base.read("test") |
61 |
0 |
assert str(excinfo.value).startswith( |
62 |
|
exception_message("Method 'read' not implemented") |
63 |
|
) |
64 |
0 |
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR |
65 |
|
|
66 |
|
|
67 |
1 |
def test_aiowrite(msg_base): |
68 |
0 |
with pytest.raises(MsgException) as excinfo: |
69 |
0 |
asyncio.run(msg_base.aiowrite("test", "test", "test")) |
70 |
0 |
assert str(excinfo.value).startswith( |
71 |
|
exception_message("Method 'aiowrite' not implemented") |
72 |
|
) |
73 |
0 |
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR |
74 |
|
|
75 |
|
|
76 |
1 |
def test_aioread(msg_base): |
77 |
0 |
with pytest.raises(MsgException) as excinfo: |
78 |
0 |
asyncio.run(msg_base.aioread("test")) |
79 |
0 |
assert str(excinfo.value).startswith( |
80 |
|
exception_message("Method 'aioread' not implemented") |
81 |
|
) |
82 |
0 |
assert excinfo.value.http_code == http.HTTPStatus.INTERNAL_SERVER_ERROR |