summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorEdi Feschiyan <edi.feschiyan@konsulko.com>2020-07-13 14:32:12 +0300
committerEdi Feschiyan <edi.feschiyan@konsulko.com>2020-07-13 14:32:54 +0300
commitd358a2bc1f9c750e8c16b0ad70b4dff35b92a9be (patch)
treee273e8dc46f005cb727391dbb85aa28e4ceabb3a
parentd58969b579998667179d58f908257e2c02e465ba (diff)
WIP: Adding taskmanager and homescreen service and tests
PEP8 cleanups, newlines replacing AFBResponse with service.afbresponse() Adding BT-PBAP CLI callables Adding bluetooth subscription tests
-rw-r--r--pyagl/services/audiomixer.py14
-rw-r--r--pyagl/services/base.py5
-rw-r--r--pyagl/services/bluetooth.py3
-rw-r--r--pyagl/services/bluetooth_map.py3
-rw-r--r--pyagl/services/bluetooth_pbap.py53
-rw-r--r--pyagl/services/geoclue.py4
-rw-r--r--pyagl/services/gps.py6
-rw-r--r--pyagl/services/homescreen.py51
-rw-r--r--pyagl/services/mediaplayer.py18
-rw-r--r--pyagl/services/nfc.py4
-rw-r--r--pyagl/services/taskmanager.py3
-rw-r--r--pyagl/services/weather.py6
-rw-r--r--pyagl/tests/test_audiomixer.py5
-rw-r--r--pyagl/tests/test_bluetooth.py62
-rw-r--r--pyagl/tests/test_bluetooth_map.py11
-rw-r--r--pyagl/tests/test_bluetooth_pbap.py18
-rw-r--r--pyagl/tests/test_geoclue.py13
-rw-r--r--pyagl/tests/test_gps.py43
-rw-r--r--pyagl/tests/test_homescreen.py42
-rw-r--r--pyagl/tests/test_network.py163
-rw-r--r--pyagl/tests/test_nfc.py3
-rw-r--r--pyagl/tests/test_weather.py30
-rw-r--r--templates/{{cookiecutter.services_dir}}/{{cookiecutter.service_slug}}.py2
23 files changed, 497 insertions, 65 deletions
diff --git a/pyagl/services/audiomixer.py b/pyagl/services/audiomixer.py
index e98a7b8..9cb1100 100644
--- a/pyagl/services/audiomixer.py
+++ b/pyagl/services/audiomixer.py
@@ -44,44 +44,44 @@ async def main():
if args.list_controls:
resp = await ams.list_controls()
print(f'Requesting list_controls with id {resp}')
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.setvolume is not None:
resp = await ams.volume(args.setvolume)
print(f'Setting volume to {args.setvolume} with id {resp}')
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.getvolume:
resp = await ams.volume()
print(f'Requesting volume with id {resp}')
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.setmute is not None:
resp = await ams.mute(args.setmute)
print(f'Setting mute to {args.setmute} with id {resp}')
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.getmute:
resp = await ams.mute()
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.subscribe:
for event in args.subscribe:
msgid = await ams.subscribe(event)
print(f'Subscribing to {event} with id {msgid}')
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.unsubscribe:
for event in args.unsubscribe:
msgid = await ams.unsubscribe(event)
print(f'Unsubscribing from {event} with id {msgid}')
- r = AFBResponse(await ams.response())
+ r = await ams.afbresponse()
print(r)
if args.listener:
diff --git a/pyagl/services/base.py b/pyagl/services/base.py
index 091a9aa..1940393 100644
--- a/pyagl/services/base.py
+++ b/pyagl/services/base.py
@@ -17,6 +17,7 @@ import re
# logging.getLogger('AGLBaseService')
# logging.basicConfig(level=logging.DEBUG)
+
# AFB message type
class AFBT(IntEnum):
REQUEST = 2,
@@ -24,6 +25,7 @@ class AFBT(IntEnum):
ERROR = 4,
EVENT = 5
+
msgq = {}
AFBLEN = 3
@@ -34,13 +36,16 @@ def newrand():
result = bs[0] * bs[1] * bs[2] * bs[3] + bs[4]
yield result
+
def addrequest(msgid, msg):
msgq[msgid] = {'request': msg, 'response': None}
+
def addresponse(msgid, msg):
if msgid in msgq.keys():
msgq[msgid]['response'] = msg
+
class AFBResponse:
type: AFBT
msgid: int
diff --git a/pyagl/services/bluetooth.py b/pyagl/services/bluetooth.py
index d111d09..296d5ee 100644
--- a/pyagl/services/bluetooth.py
+++ b/pyagl/services/bluetooth.py
@@ -75,7 +75,7 @@ async def main(loop):
if args.default_adapter:
msgid = await bts.default_adapter()
print(f'Requesting default adapter with id {msgid}')
- r = AFBResponse(await bts.response())
+ r = await bts.afbresponse()
print(r)
if args.adapter_state:
@@ -87,7 +87,6 @@ async def main(loop):
bts.logger.debug(await bts.adapter_state('hci0', {'uuids': ['0000110e-0000-1000-8000-00805f9b34fb']}))
-
if __name__ == '__main__':
loop = asyncio.get_event_loop()
diff --git a/pyagl/services/bluetooth_map.py b/pyagl/services/bluetooth_map.py
index 24797db..c928b43 100644
--- a/pyagl/services/bluetooth_map.py
+++ b/pyagl/services/bluetooth_map.py
@@ -2,6 +2,7 @@ from pyagl.services.base import AGLBaseService, AFBResponse
import asyncio
import os
+
class BTMAPService(AGLBaseService):
service = 'agl-service-bluetooth-map'
parser = AGLBaseService.getparser()
@@ -10,7 +11,6 @@ class BTMAPService(AGLBaseService):
parser.add_argument('--message', help='Message to send to the recipient')
parser.add_argument('--list_messages', help='List text messages over MAP', action='store_true')
-
def __init__(self, ip, port=None, service='agl-service-bluetooth-map'):
super().__init__(api='bluetooth-map', ip=ip, port=port, service=service)
@@ -29,6 +29,7 @@ class BTMAPService(AGLBaseService):
async def unsubscribe(self, event='notification'):
return await super().subscribe(event)
+
async def main(loop):
args = BTMAPService.parser.parse_args()
bmp = await BTMAPService(args.ipaddr)
diff --git a/pyagl/services/bluetooth_pbap.py b/pyagl/services/bluetooth_pbap.py
index 6472b60..027ac89 100644
--- a/pyagl/services/bluetooth_pbap.py
+++ b/pyagl/services/bluetooth_pbap.py
@@ -1,11 +1,19 @@
from pyagl.services.base import AGLBaseService, AFBResponse
+import json
+from pprint import pprint
import asyncio
import os
+
class BTPBAPService(AGLBaseService):
service = 'agl-service-bluetooth-pbap'
parser = AGLBaseService.getparser()
+ parser.add_argument('--import_contacts', action='store_true')
+ parser.add_argument('--status', action='store_true')
+ parser.add_argument('--contacts', action='store_true')
+ parser.add_argument('--history', help='Request call history - ICH/OCH/MCH/CCH')
+
def __init__(self, ip, port=None, service='agl-service-bluetooth-pbap'):
super().__init__(api='bluetooth-pbap', ip=ip, port=port, service=service)
@@ -33,3 +41,48 @@ class BTPBAPService(AGLBaseService):
async def history(self, param):
return await super().request('history', {'list': param})
+async def main(loop):
+ args = BTPBAPService.parser.parse_args()
+ svc = await BTPBAPService(args.ipaddr, args.port)
+
+ if args.loglevel:
+ svc.logger.setLevel(args.loglevel)
+
+ if args.import_contacts:
+ msgid = await svc.import_contacts()
+ print(f'Sent import_contacts request with messageid {msgid}')
+ resp = await svc.afbresponse()
+ if resp.status == 'success':
+ print(json.dumps(resp.data, indent=2, sort_keys=True))
+ else:
+ print(resp)
+
+ if args.status:
+ msgid = await svc.status()
+ print(f'Sent status request with messageid {msgid}')
+ resp = await svc.afbresponse()
+ print(resp.data)
+
+ if args.contacts:
+ msgid = await svc.contacts()
+ print(f'Sent contacts request with messageid {msgid}')
+ print(await svc.afbresponse())
+
+ if args.history:
+ msgid = await svc.history(param=args.history)
+ print(f'Sent history request with messageid {msgid}')
+ print(await svc.afbresponse())
+
+ if args.subscribe:
+ for event in args.subscribe:
+ msgid = await svc.subscribe(event)
+ print(f'Subscribing for event {event} with messageid {msgid}')
+ print(await svc.afbresponse())
+
+ if args.listener:
+ async for response in svc.listener():
+ print(response)
+
+if __name__ == '__main__':
+ loop = asyncio.get_event_loop()
+ loop.run_until_complete(main(loop))
diff --git a/pyagl/services/geoclue.py b/pyagl/services/geoclue.py
index c14c6db..5e5b443 100644
--- a/pyagl/services/geoclue.py
+++ b/pyagl/services/geoclue.py
@@ -28,13 +28,13 @@ async def main(loop):
if args.location:
msgid = await gcs.location()
print(f'Sent location request with messageid {msgid}')
- print(AFBResponse(await gcs.response()))
+ print(await gcs.afbresponse())
if args.subscribe:
for event in args.subscribe:
msgid = await gcs.subscribe(event)
print(f"Subscribed for {event} with messageid {msgid}")
- print(AFBResponse(await gcs.response()))
+ print(await gcs.afbresponse())
if args.listener:
async for response in gcs.listener():
print(response)
diff --git a/pyagl/services/gps.py b/pyagl/services/gps.py
index 69c253e..e91e9df 100644
--- a/pyagl/services/gps.py
+++ b/pyagl/services/gps.py
@@ -35,17 +35,17 @@ async def main(loop):
if args.record:
msgid = await gpss.record(args.record)
print(f'Sent gps record request with value {args.record} with messageid {msgid}')
- print(AFBResponse(await gpss.response()))
+ print(await gpss.afbresponse())
if args.location:
msgid = await gpss.location()
- print(AFBResponse(await gpss.response()))
+ print(await gpss.afbresponse())
if args.subscribe:
for event in args.subscribe:
msgid = await gpss.subscribe(event)
print(f'Subscribed for event {event} with messageid {msgid}')
- print(AFBResponse(await gpss.response()))
+ print(await gpss.afbresponse())
if args.listener:
async for response in gpss.listener():
diff --git a/pyagl/services/homescreen.py b/pyagl/services/homescreen.py
new file mode 100644
index 0000000..4af47d9
--- /dev/null
+++ b/pyagl/services/homescreen.py
@@ -0,0 +1,51 @@
+from pyagl.services.base import AGLBaseService, AFBResponse
+import asyncio
+import os
+
+events = ['tap_shortcut', 'hideWindow', 'replyShowWindow', 'on_screen_message', 'on_screen_reply', 'showNotification',
+ 'showNotification', 'application-list-changed']
+
+
+class HomeScreenService(AGLBaseService):
+ service = 'agl-service-homescreen'
+ parser = AGLBaseService.getparser()
+ parser.add_argument('--tap_shortcut', help='Invoke tap_shortcut with parameter')
+
+ def __init__(self, ip, port=None):
+ super().__init__(api='homescreen', ip=ip, port=port, service='agl-service-homescreen')
+
+ async def tap_shortcut(self, app: str):
+ return await self.request('tap_shortcut', {'appid': app})
+
+ async def subscribe(self, event):
+ return await super().subscribe(event='event')
+
+ async def unsubscribe(self, event):
+ return await super().unsubscribe(event='event')
+
+
+async def main(loop):
+ args = HomeScreenService.parser.parse_args()
+ svc = await HomeScreenService(args.ipaddr, args.port)
+
+ if args.loglevel:
+ svc.logger.setLevel(args.loglevel)
+
+ if args.tap_shortcut:
+ msgid = await svc.tap_shortcut(args.tap_shortcut)
+ print(f'Sent tap_shortcut request with value {args.tap_shortcut} and messageid {msgid}')
+ print(await svc.afbresponse())
+
+ if args.subscribe:
+ for event in args.subscribe:
+ msgid = await svc.subscribe(event)
+ print(f'Subscribing for event {event} with messageid {msgid}')
+ print(await svc.afbresponse())
+
+ if args.listener:
+ async for response in svc.listener():
+ print(response)
+
+if __name__ == '__main__':
+ loop = asyncio.get_event_loop()
+ loop.run_until_complete(main(loop))
diff --git a/pyagl/services/mediaplayer.py b/pyagl/services/mediaplayer.py
index 4472531..31f49dd 100644
--- a/pyagl/services/mediaplayer.py
+++ b/pyagl/services/mediaplayer.py
@@ -92,49 +92,49 @@ async def main(loop):
if args.playlist:
msgid = await MPS.playlist()
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
for l in r.data['list']: print(l)
if args.control:
msgid = await MPS.control(args.control)
print(f'Sent {args.control} request with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
if args.seek:
msgid = await MPS.control('seek', args.seek)
print(f'Sent seek request to {args.seek} msec with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
if args.fastforward:
msgid = await MPS.control('fast-forward', args.fastforward)
print(f'Sent fast-forward request for {args.fastforward} msec with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
if args.rewind:
msgid = await MPS.control('rewind', -args.rewind)
print(f'Sent rewind request for {args.rewind} msec with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
if args.picktrack:
msgid = await MPS.control('pick-track', args.picktrack)
print(f'Sent pick-track request with index {args.rewind} with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
if args.volume:
msgid = await MPS.control('volume', int(args.volume))
print(f'Sent volume request: {args.rewind} with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
if args.loop:
msgid = await MPS.control('loop', args.loop)
print(f'Sent loop-state request: {args.loop} with messageid {msgid}')
- r = AFBResponse(await MPS.response())
+ r = await MPS.afbresponse()
print(r)
# if args.avrcp:
@@ -147,7 +147,7 @@ async def main(loop):
for event in args.subscribe:
msgid = await MPS.subscribe(event)
print(f"Subscribed for event {event} with messageid {msgid}")
- r = await MPS.response()
+ r = await MPS.afbresponse()
print(r)
if args.listener:
diff --git a/pyagl/services/nfc.py b/pyagl/services/nfc.py
index 78f9b6e..8e333a1 100644
--- a/pyagl/services/nfc.py
+++ b/pyagl/services/nfc.py
@@ -23,14 +23,14 @@ async def main(loop):
for event in args.subscribe:
msgid = await nfcs.subscribe(event)
print(f"Subscribing for event {event} with messageid {msgid}")
- r = AFBResponse(await nfcs.response())
+ r = await nfcs.afbresponse()
print(r)
if args.unsubscribe:
for event in args.unsubscribe:
msgid = await nfcs.unsubscribe(event)
print(f"Unsubscribing for event {event} with messageid {msgid}")
- r = AFBResponse(await nfcs.response())
+ r = await nfcs.afbresponse()
print(r)
if args.listener:
diff --git a/pyagl/services/taskmanager.py b/pyagl/services/taskmanager.py
index c3cd3b6..96f98d2 100644
--- a/pyagl/services/taskmanager.py
+++ b/pyagl/services/taskmanager.py
@@ -6,3 +6,6 @@ import os
class TaskManagerService(AGLBaseService):
service = 'agl-service-taskmanager'
parser = AGLBaseService.getparser()
+
+ def __init__(self, ip, port=None, service='agl-service-taskmanager'):
+ super().__init__(api='taskmanager', ip=ip, port=port, service=service) \ No newline at end of file
diff --git a/pyagl/services/weather.py b/pyagl/services/weather.py
index 389be70..c5ce95c 100644
--- a/pyagl/services/weather.py
+++ b/pyagl/services/weather.py
@@ -24,19 +24,19 @@ async def main():
aws = await WeatherService(ip=args.ipaddr, port=args.port)
if args.current:
msgid = await aws.current_weather()
- resp = AFBResponse(await aws.response())
+ resp = await aws.afbresponse()
print(json.dumps(resp.data, indent=2))
if args.apikey:
msgid = await aws.apikey()
- resp = AFBResponse(await aws.response())
+ resp = await aws.afbresponse()
print(resp.data['api_key'])
if args.subscribe:
for event in args.subscribe:
msgid = await aws.subscribe(event)
print(f'Subscribed for event {event} with messageid {msgid}')
- resp = AFBResponse(await aws.response())
+ resp = await aws.afbresponse()
print(resp)
if args.listener:
diff --git a/pyagl/tests/test_audiomixer.py b/pyagl/tests/test_audiomixer.py
index de07944..1f49760 100644
--- a/pyagl/tests/test_audiomixer.py
+++ b/pyagl/tests/test_audiomixer.py
@@ -40,6 +40,7 @@ async def test_volume_verb(event_loop, service: AMS):
assert resp.type == AFBT.RESPONSE
assert resp.status == 'success'
+
async def test_set_volume0(event_loop, service: AMS):
msgid = await service.volume(value=0)
resp = await service.afbresponse()
@@ -47,6 +48,7 @@ async def test_set_volume0(event_loop, service: AMS):
assert resp.type == AFBT.RESPONSE
assert resp.status == 'success'
+
async def test_set_maxvolume(event_loop, service: AMS):
msgid = await service.volume(value=1)
resp = await service.afbresponse()
@@ -54,6 +56,7 @@ async def test_set_maxvolume(event_loop, service: AMS):
assert resp.type == AFBT.RESPONSE
assert resp.status == 'success'
+
async def test_get_mute(event_loop, service: AMS):
msgid = await service.mute()
resp = await service.afbresponse()
@@ -61,6 +64,7 @@ async def test_get_mute(event_loop, service: AMS):
assert resp.type == AFBT.RESPONSE
assert resp.status == 'success'
+
async def test_set_mute(event_loop, service: AMS):
msgid = await service.mute(value=1)
resp = await service.afbresponse()
@@ -68,6 +72,7 @@ async def test_set_mute(event_loop, service: AMS):
assert resp.type == AFBT.RESPONSE
assert resp.status == 'success'
+
async def test_set_unmute(event_loop, service: AMS):
msgid = await service.mute(value=0)
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_bluetooth.py b/pyagl/tests/test_bluetooth.py
index 44cb613..f5903ec 100644
--- a/pyagl/tests/test_bluetooth.py
+++ b/pyagl/tests/test_bluetooth.py
@@ -40,31 +40,79 @@ def btaddr():
@pytest.mark.dependency
async def test_default_adapter(event_loop, service: BTS):
msgid = await service.default_adapter()
- resp = AFBResponse(await service.response())
- assert resp.status == 'success', resp
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
assert 'adapter' in resp.data.keys()
assert resp.data['adapter'] == 'hci0'
+async def test_subscribe_device_changes(event_loop, service: BTS):
+ msgid = await service.subscribe('device_changes')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_device_changes(event_loop, service: BTS):
+ msgid = await service.unsubscribe('device_changes')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_adapter_changes(event_loop, service: BTS):
+ msgid = await service.subscribe('adapter_changes')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_adapter_changes(event_loop, service: BTS):
+ msgid = await service.unsubscribe('adapter_changes')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_media(event_loop, service: BTS):
+ msgid = await service.subscribe('media')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_media(event_loop, service: BTS):
+ msgid = await service.unsubscribe('media')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_agent(event_loop, service: BTS):
+ msgid = await service.subscribe('agent')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_agent(event_loop, service: BTS):
+ msgid = await service.unsubscribe('agent')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
@pytest.mark.dependency(depends=['test_default_adapter'])
async def test_managed_objects(event_loop, service: BTS):
msgid = await service.managed_objects()
- resp = AFBResponse(await service.response())
- assert resp.status == 'success', str(resp)
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
@pytest.mark.dependency(depends=['test_default_adapter'])
async def test_has_single_adapter(event_loop, service: BTS):
msgid = await service.managed_objects()
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert len(resp.data['adapters']) == 1, \
- f'Detected {len(resp.data["adapters"])} adapters. Multiple adapters may also break testing'
+ f'Detected {len(resp.data["adapters"])} adapters. Multiple adapters may also affect testing'
@pytest.mark.dependency(depends=['test_default_adapter'])
async def test_adapter_state(event_loop, service: BTS):
msgid = await service.adapter_state('hci0')
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert resp.status == 'success', 'adapter state verb failed'
diff --git a/pyagl/tests/test_bluetooth_map.py b/pyagl/tests/test_bluetooth_map.py
index 9dec04c..68b885e 100644
--- a/pyagl/tests/test_bluetooth_map.py
+++ b/pyagl/tests/test_bluetooth_map.py
@@ -24,6 +24,7 @@ async def service():
yield ams
await ams.websocket.close()
+
@pytest.fixture(scope='module')
def recipient():
phonenumber = os.environ.get('AGL_BTMAP_RECIPIENT', None)
@@ -31,6 +32,7 @@ def recipient():
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',
@@ -38,7 +40,6 @@ def composetext():
return text
-
@pytest.mark.xfail(reason='Expected to fail if no messages are found')
@pytest.fixture(scope='module')
async def messages(service: BMP):
@@ -48,12 +49,14 @@ async def messages(service: BMP):
assert len(resp.data) > 0, "no messages found "
yield resp.data['messages']
+
@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.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):
@@ -87,3 +90,9 @@ async def test_message_to_self(event_loop, service: BMP, recipient, composetext)
assert 'message' in resp.data
assert resp.data['message'] == composetext, 'Message mismatching'
+
+@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
diff --git a/pyagl/tests/test_bluetooth_pbap.py b/pyagl/tests/test_bluetooth_pbap.py
index 26db0ba..7084f28 100644
--- a/pyagl/tests/test_bluetooth_pbap.py
+++ b/pyagl/tests/test_bluetooth_pbap.py
@@ -24,6 +24,15 @@ async def service():
yield ams
await ams.websocket.close()
+
+@pytest.fixture(scope='module')
+def phonenumber():
+ phonenumenv = os.environ.get('AGL_PBAP_PHONENUM')
+ if not phonenumenv:
+ pytest.xfail('Please export AGL_PBAP_PHONENUM with valid phone number, existing in the phonebook to run tests')
+
+ return phonenumenv
+
@pytest.fixture(scope='module')
def searchvcf():
vcf = os.environ.get('AGL_PBAP_VCF')
@@ -45,8 +54,8 @@ async def test_status(event_loop, service: PBAP):
assert resp.data['connected'] is True
-async def test_search(event_loop, service: PBAP):
- msgid = await service.search("+359887224379")
+async def test_search(event_loop, service: PBAP, phonenumber):
+ msgid = await service.search(phonenumber)
resp = await service.afbresponse()
assert resp.status == 'success'
@@ -74,6 +83,7 @@ async def test_contacts(event_loop, service: PBAP):
assert len(vcards) > 0
print(f' contacts verb returned {len(vcards)} cached .vcf objects ')
+
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_incoming_calls(event_loop, service: PBAP):
msgid = await service.history(param='ich')
@@ -83,6 +93,7 @@ async def test_history_incoming_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
+
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_outgoing_calls(event_loop, service: PBAP):
msgid = await service.history(param='och')
@@ -92,6 +103,7 @@ async def test_history_outgoing_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
+
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_missed_calls(event_loop, service: PBAP):
msgid = await service.history(param='mch')
@@ -101,6 +113,7 @@ async def test_history_missed_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
+
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_combined_calls(event_loop, service: PBAP):
msgid = await service.history(param='cch')
@@ -110,6 +123,7 @@ async def test_history_combined_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
+
@pytest.mark.dependency(depends=['test_contacts'])
async def test_entry_phonebook(event_loop, service: PBAP, searchvcf):
msgid = await service.entry(handle=searchvcf)
diff --git a/pyagl/tests/test_geoclue.py b/pyagl/tests/test_geoclue.py
index 40b995a..64875d8 100644
--- a/pyagl/tests/test_geoclue.py
+++ b/pyagl/tests/test_geoclue.py
@@ -7,11 +7,13 @@ from pyagl.services.base import AFBResponse, AFBT
from pyagl.services.geoclue import GeoClueService as gcs
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')
@@ -28,3 +30,14 @@ async def test_location(event_loop, service: gcs):
assert resp.type == AFBT.RESPONSE
assert resp.status == 'success'
+
+async def test_subscribe(event_loop, service: gcs):
+ msgid = await service.subscribe()
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+
+
+async def test_unsubscribe(event_loop, service: gcs):
+ msgid = await service.unsubscribe()
+ resp = await service.afbresponse()
+ assert resp.status == 'success' \ No newline at end of file
diff --git a/pyagl/tests/test_gps.py b/pyagl/tests/test_gps.py
index cb28be6..3bfeb48 100644
--- a/pyagl/tests/test_gps.py
+++ b/pyagl/tests/test_gps.py
@@ -8,11 +8,13 @@ from concurrent.futures import TimeoutError
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')
@@ -26,47 +28,64 @@ async def service():
# async for _response in service.listener():
# yield _response
+
async def test_location_verb(event_loop, service: GPS):
msgid = await service.location()
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert resp.msgid == msgid
-@pytest.mark.xfail # expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable
+
+@pytest.mark.xfail(reason='expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable')
async def test_location_result(event_loop, service: GPS):
msgid = await service.location()
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert resp.status == 'success'
+
async def test_subscribe_verb(event_loop, service: GPS):
msgid = await service.subscribe()
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert resp.msgid == msgid
assert resp.status == 'success'
+
+async def test_enable_recording(event_loop, service: GPS):
+ msgid = await service.record()
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_disable_recording(event_loop, service: GPS):
+ msgid = await service.record('off')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+@pytest.mark.dependency
async def test_subscribe_location(event_loop, service: GPS):
msgid = await service.subscribe('location')
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert resp.msgid == msgid
assert resp.status == 'success'
-async def test_unsubscribe(event_loop, service: GPS):
- msgid = await service.unsubscribe('location')
- resp = AFBResponse(await service.response())
- assert resp.msgid == msgid
- assert resp.status == 'success'
+@pytest.mark.dependency(depends=['test_subscribe_location'])
@pytest.mark.xfail # expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable
async def test_location_events(event_loop, service: GPS):
msgid = await service.subscribe('location')
- resp = AFBResponse(await service.response())
+ resp = await service.afbresponse()
assert resp.msgid == msgid
assert resp.status == 'success' # successful subscription
try:
resp = await asyncio.wait_for(service.afbresponse(), 10)
- resp = AFBResponse(resp)
assert resp.type == AFBT.EVENT, f'Expected EVENT response, got {resp.type.name} instead'
# TODO one more assert for the actual received event, haven't received a location event yet
except TimeoutError:
pytest.xfail("Did not receive location event")
+async def test_unsubscribe(event_loop, service: GPS):
+ msgid = await service.unsubscribe('location')
+ resp = await service.afbresponse()
+ assert resp.msgid == msgid
+ assert resp.status == 'success' \ No newline at end of file
diff --git a/pyagl/tests/test_homescreen.py b/pyagl/tests/test_homescreen.py
new file mode 100644
index 0000000..2935bfa
--- /dev/null
+++ b/pyagl/tests/test_homescreen.py
@@ -0,0 +1,42 @@
+import asyncio
+import os
+import pytest
+import logging
+from pyagl.services.base import AFBResponse, AFBT
+from concurrent.futures import TimeoutError
+
+from pyagl.services.homescreen import HomeScreenService as hcs
+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)
+ svc = await hcs(ip=address, port=port)
+ yield svc
+ await svc.websocket.close()
+
+
+async def test_tap_shortcut_media(event_loop, service: hcs):
+ msgid = await service.tap_shortcut('mediaplayer')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+
+
+async def test_subscribe(event_loop, service: hcs):
+ msgid = await service.subscribe(event='tap_shortcut')
+ resp = await service.afbresponse()
+ assert resp.msgid == msgid
+
+
+async def test_unsubscribe(event_loop, service: hcs):
+ msgid = await service.unsubscribe(event='tap_shortcut')
+ resp = await service.afbresponse()
+ assert resp.msgid == msgid
diff --git a/pyagl/tests/test_network.py b/pyagl/tests/test_network.py
index f1bc56d..6c7c12d 100644
--- a/pyagl/tests/test_network.py
+++ b/pyagl/tests/test_network.py
@@ -31,8 +31,10 @@ async def test_state(event_loop, service: NS):
assert resp.status == 'success', resp.info
assert resp.data == 'online'
-@pytest.mark.skip(reason='We want to skip global offline mode until this is ran locally on the board')
async def test_global_offline(event_loop, service: NS):
+ addr, _ = service.websocket.remote_address
+ if addr is not 'localhost' or addr is not '127.0.0.1':
+ pytest.skip('We want to skip global offline mode until this is ran locally on the board against localhost')
msgid = await service.offline(True)
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
@@ -67,30 +69,165 @@ async def test_get_property(event_loop, service: NS, expected_available_techs):
assert resp.status == 'success', resp.info
assert isinstance(resp.data, dict)
expected_fields = frozenset(['name', 'type', 'powered', 'connected', 'tethering'])
- diverging_fields = frozenset(expected_fields).symmetric_difference(frozenset(resp.data.keys()))
+ #diverging_fields = frozenset(expected_fields).symmetric_difference(frozenset(resp.data.keys()))
+ diverging_fields = frozenset(expected_fields).difference(frozenset(resp.data.keys()))
assert len(diverging_fields) == 0, f'the following property fields are diverging from the expected: {diverging_fields}'
+@pytest.mark.xfail(reason='Expecting this to throw "permission denied" via the API, tethering from connmanctl succeeds')
async def test_enable_wifi_tethering(event_loop, service: NS, expected_available_techs):
if 'wifi' not in expected_available_techs:
- pytest.skip('Skipping wifi tethering test because its not expected to work')
+ pytest.skip('Skipping wifi tethering test because its not marked as available technology')
msgid = await service.set_property('wifi', {'tethering': True,
- 'tetheringidentifier': 'AGL',
- 'tetheringpassphrase': 'agltestwifi'})
+ 'TetheringIdentifier': 'AGL',
+ 'TetheringPassphrase': 'agltestwifi'})
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
-
+@pytest.mark.dependency(depends='test_enable_wifi_tethering')
async def test_disable_wifi_tethering(event_loop, service: NS, expected_available_techs):
if 'wifi' not in expected_available_techs:
- pytest.skip('Skipping wifi tethering test because its not expected to work')
+ pytest.skip('Skipping wifi tethering test because its not marked as available technology')
msgid = await service.set_property('wifi', {'tethering': False})
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
-async def test_set_property(event_loop, service: NS, expected_available_techs):
- for t in expected_available_techs:
- msgid = await service.set_property(t, {'tethering': True})
- resp = await service.afbresponse()
- assert resp.status == 'success', resp.info
- print(resp) \ No newline at end of file
+#
+# async def test_set_property(event_loop, service: NS, expected_available_techs):
+# for t in expected_available_techs:
+# msgid = await service.set_property(t, {'tethering': True})
+# resp = await service.afbresponse()
+# assert resp.status == 'success', resp.info
+# print(resp)
+
+
+async def test_services_verb(event_loop, service: NS):
+ msgid = await service.services()
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+ assert 'values' in resp.data
+
+
+async def test_subscribe_global_state(event_loop, service: NS):
+ msgid = await service.subscribe('global_state')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_global_state(event_loop, service: NS):
+ msgid = await service.unsubscribe('global_state')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_technologies(event_loop, service: NS):
+ msgid = await service.subscribe('technologies')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_technologies(event_loop, service: NS):
+ msgid = await service.unsubscribe('technologies')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_tech_props(event_loop, service: NS):
+ msgid = await service.subscribe('technology_properties')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_tech_props(event_loop, service: NS):
+ msgid = await service.unsubscribe('technology_properties')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_services(event_loop, service: NS):
+ msgid = await service.subscribe('services')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_services(event_loop, service: NS):
+ msgid = await service.unsubscribe('services')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_service_props(event_loop, service: NS):
+ msgid = await service.subscribe('service_properties')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_service_props(event_loop, service: NS):
+ msgid = await service.unsubscribe('service_properties')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_subscribe_agent(event_loop, service: NS):
+ msgid = await service.subscribe('agent')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_unsubscribe_agent(event_loop, service: NS):
+ msgid = await service.unsubscribe('agent')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_enable_wifi(event_loop, service: NS, expected_available_techs):
+ if 'wifi' not in expected_available_techs:
+ pytest.skip('Skipping enable_technology for "wifi" because it is not expected to be available')
+ msgid = await service.enable_technology('wifi')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_disable_wifi(event_loop, service: NS, expected_available_techs):
+ if 'wifi' not in expected_available_techs:
+ pytest.skip('Skipping disable_technology for "wifi" because it is not expected to be available')
+ msgid = await service.disable_technology('wifi')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_enable_bluetooth(event_loop, service: NS, expected_available_techs):
+ if 'bluetooth' not in expected_available_techs:
+ pytest.skip('Skipping enable_technology for "bluetooth" because it is not expected to be available')
+ msgid = await service.enable_technology('bluetooth')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_disable_bluetooth(event_loop, service: NS, expected_available_techs):
+ if 'bluetooth' not in expected_available_techs:
+ pytest.skip('Skipping disable_technology for "bluetooth" because it is not expected to be available')
+ msgid = await service.disable_technology('bluetooth')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_enable_ethernet(event_loop, service: NS, expected_available_techs):
+ if 'ethernet' not in expected_available_techs:
+ pytest.skip('Skipping enable_technology for "ethernet" because it is not expected to be available')
+ msgid = await service.enable_technology('ethernet')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info
+
+
+async def test_disable_ethernet(event_loop, service: NS, expected_available_techs):
+ addr, _ = service.websocket.remote_address
+ if addr is not 'localhost' or addr is not '127.0.0.1':
+ pytest.skip('Skipping this test until it is ran locally on the board, '
+ 'presuming it is the only available interface to connect to remotely for testing')
+ if 'ethernet' not in expected_available_techs:
+ pytest.skip('Skipping disable_technology for "ethernet" because it is not expected to be available')
+
+ msgid = await service.disable_technology('ethernet')
+ resp = await service.afbresponse()
+ assert resp.status == 'success', resp.info \ No newline at end of file
diff --git a/pyagl/tests/test_nfc.py b/pyagl/tests/test_nfc.py
index 3b4376f..d09f443 100644
--- a/pyagl/tests/test_nfc.py
+++ b/pyagl/tests/test_nfc.py
@@ -8,11 +8,13 @@ from concurrent.futures import TimeoutError
from pyagl.services.nfc import NFCService as nfcs
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')
@@ -21,6 +23,7 @@ async def service():
yield svc
await svc.websocket.close()
+
async def subscribe(event_loop, service: nfcs):
msgid = service.subscribe()
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_weather.py b/pyagl/tests/test_weather.py
index 696b1d8..7b35c6d 100644
--- a/pyagl/tests/test_weather.py
+++ b/pyagl/tests/test_weather.py
@@ -8,11 +8,13 @@ from pyagl.services.weather import WeatherService as ws
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')
@@ -21,15 +23,43 @@ async def service():
yield gpss
await gpss.websocket.close()
+
async def test_apikey(event_loop, service: ws):
msgid = await service.apikey()
resp = await service.afbresponse()
assert resp.msgid == msgid
assert resp.data['api_key'] == 'a860fa437924aec3d0360cc749e25f0e'
+
async def test_current_weather(event_loop, service: ws):
msgid = await service.current_weather()
resp = await service.afbresponse()
assert resp.msgid == msgid
assert 'sys' in resp.data
+
+async def test_bad_subscription(event_loop, service: ws):
+ msgid = await service.subscribe('non-existant')
+ resp = await service.afbresponse()
+ assert resp.status == 'failed'
+
+
+async def test_bad_unsubscription(event_loop, service: ws):
+ msgid = await service.unsubscribe('non-existant')
+ resp = await service.afbresponse()
+ assert resp.status == 'failed'
+
+
+async def test_subscribe_weather(event_loop, service: ws):
+ event = 'weather'
+ msgid = await service.subscribe(event)
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ eventresp = await service.afbresponse() # immediately emits event after subscription
+ assert eventresp.api == f'{service.api}/{event}'
+
+
+async def test_unsubscribe_weather(event_loop, service: ws):
+ msgid = await service.subscribe('weather')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
diff --git a/templates/{{cookiecutter.services_dir}}/{{cookiecutter.service_slug}}.py b/templates/{{cookiecutter.services_dir}}/{{cookiecutter.service_slug}}.py
index dab3a68..0d79e53 100644
--- a/templates/{{cookiecutter.services_dir}}/{{cookiecutter.service_slug}}.py
+++ b/templates/{{cookiecutter.services_dir}}/{{cookiecutter.service_slug}}.py
@@ -13,5 +13,5 @@ class {{cookiecutter.classname}}(AGLBaseService):
async def main(loop):
args = {{cookiecutter.classname}}.parser.parse_args()
- svc = {{cookiecutter.classname}}(args.ipaddr)
+ svc = await {{cookiecutter.classname}}(args.ipaddr)