1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
|
import asyncio
import os
import pytest
from random import randint
from pyagl.services.base import AFBResponse, AFBT
from pyagl.services.bluetooth_map import BTMAPService as BMP
import logging
pytestmark = pytest.mark.asyncio
@pytest.fixture(scope='module')
def event_loop():
loop = asyncio.get_event_loop()
yield loop
@pytest.fixture(scope='module')
async def service():
address = os.environ.get('AGL_TGT_IP', 'localhost')
port = os.environ.get('AGL_TGT_PORT', None)
ams = await BMP(ip=address, port=port)
yield ams
await ams.websocket.close()
@pytest.fixture(scope='module')
def recipient():
phonenumber = os.environ.get('AGL_BTMAP_RECIPIENT', None)
if phonenumber is None:
pytest.xfail('No phone number was set, please "export AGL_BTMAP_RECIPIENT=" with a phonenumber')
return phonenumber
@pytest.fixture(scope='module')
def composetext():
text = os.environ.get('AGL_BTMAP_TEXT',
f'#{randint(0,9999999)} - Hello from AGL. This message is generated by test_bluetooth_map.py')
return text
@pytest.mark.hwrequired
@pytest.mark.btmap
@pytest.mark.xfail(reason='Expected to fail if no messages are found')
@pytest.fixture(scope='module')
async def messages(service: BMP):
msgid = await service.list_messages()
resp = await service.afbresponse()
assert 'messages' in resp.data
assert len(resp.data) > 0, "no messages found "
yield resp.data['messages']
@pytest.mark.hwrequired
@pytest.mark.btmap
@pytest.mark.dependency
async def test_list_messages(event_loop, service: BMP):
msgid = await service.list_messages()
resp = await service.afbresponse()
assert resp.msgid == msgid
@pytest.mark.hwrequired
@pytest.mark.btmap
@pytest.mark.dependency(depends=['test_list_messages'])
@pytest.mark.xfail(reason='Expected to fail if there are no messages, need message id to test the verb')
async def test_message_verb(event_loop, service, messages):
assert len(messages), "No existing messages to test the verb with"
messageids = list(messages.keys())
msgid = await service.message(messageids.pop())
resp = await service.afbresponse()
assert resp.status == 'success'
@pytest.mark.regular
@pytest.mark.btmap
@pytest.mark.dependency
async def test_subscribe_notifications(event_loop, service: BMP):
msgid = await service.subscribe('notification')
resp = await service.afbresponse()
print(resp)
assert resp.status == 'success'
@pytest.mark.hwrequired
@pytest.mark.btmap
@pytest.mark.dependency
async def test_compose_message(event_loop, service: BMP, recipient, composetext):
msgid = await service.compose(recipient, composetext)
resp = await service.afbresponse()
assert resp.status == 'success'
@pytest.mark.hwrequired
@pytest.mark.btmap
@pytest.mark.dependency(depends=['test_compose_message', 'test_subscribe_notifications'])
async def test_message_to_self(event_loop, service: BMP, recipient, composetext):
resp = await service.afbresponse()
assert resp.api == f'{service.api}/notification'
assert resp.type == AFBT.EVENT
assert isinstance(resp.data, dict)
assert 'message' in resp.data
assert resp.data['message'] == composetext, 'Message mismatching'
@pytest.mark.hwrequired
@pytest.mark.btmap
@pytest.mark.dependency(depends=['test_subscribe_notifications'])
async def test_unsubscribe_notifications(event_loop, service: BMP):
msgid = await service.unsubscribe('notification')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
|
