diff options
| -rw-r--r-- | pyagl/services/audiomixer.py | 14 | ||||
| -rw-r--r-- | pyagl/services/base.py | 5 | ||||
| -rw-r--r-- | pyagl/services/bluetooth.py | 3 | ||||
| -rw-r--r-- | pyagl/services/bluetooth_map.py | 3 | ||||
| -rw-r--r-- | pyagl/services/bluetooth_pbap.py | 53 | ||||
| -rw-r--r-- | pyagl/services/geoclue.py | 4 | ||||
| -rw-r--r-- | pyagl/services/gps.py | 6 | ||||
| -rw-r--r-- | pyagl/services/homescreen.py | 51 | ||||
| -rw-r--r-- | pyagl/services/mediaplayer.py | 18 | ||||
| -rw-r--r-- | pyagl/services/nfc.py | 4 | ||||
| -rw-r--r-- | pyagl/services/taskmanager.py | 3 | ||||
| -rw-r--r-- | pyagl/services/weather.py | 6 | ||||
| -rw-r--r-- | pyagl/tests/test_audiomixer.py | 5 | ||||
| -rw-r--r-- | pyagl/tests/test_bluetooth.py | 62 | ||||
| -rw-r--r-- | pyagl/tests/test_bluetooth_map.py | 11 | ||||
| -rw-r--r-- | pyagl/tests/test_bluetooth_pbap.py | 18 | ||||
| -rw-r--r-- | pyagl/tests/test_geoclue.py | 13 | ||||
| -rw-r--r-- | pyagl/tests/test_gps.py | 43 | ||||
| -rw-r--r-- | pyagl/tests/test_homescreen.py | 42 | ||||
| -rw-r--r-- | pyagl/tests/test_network.py | 163 | ||||
| -rw-r--r-- | pyagl/tests/test_nfc.py | 3 | ||||
| -rw-r--r-- | pyagl/tests/test_weather.py | 30 | ||||
| -rw-r--r-- | templates/{{cookiecutter.services_dir}}/{{cookiecutter.service_slug}}.py | 2 |
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) |