diff options
| -rw-r--r-- | pyagl/services/audiomixer.py | 1 | ||||
| -rw-r--r-- | pyagl/services/bluetooth-map.py | 0 | ||||
| -rw-r--r-- | pyagl/tests/test_audiomixer.py | 76 | ||||
| -rw-r--r-- | tests/test_bluetooth.py | 104 | ||||
| -rw-r--r-- | tests/test_gps.py | 71 |
5 files changed, 76 insertions, 176 deletions
diff --git a/pyagl/services/audiomixer.py b/pyagl/services/audiomixer.py index 3f35bc8..e98a7b8 100644 --- a/pyagl/services/audiomixer.py +++ b/pyagl/services/audiomixer.py @@ -15,7 +15,6 @@ class AudioMixerService(AGLBaseService): parser.add_argument('--setvolume', help='Set volume level', type=float) parser.add_argument('--getvolume', help='Get volume level', action='store_true') - def __init__(self, ip, port=None, service='agl-service-audiomixer'): super().__init__(api='audiomixer', ip=ip, port=port, service=service) diff --git a/pyagl/services/bluetooth-map.py b/pyagl/services/bluetooth-map.py deleted file mode 100644 index e69de29..0000000 --- a/pyagl/services/bluetooth-map.py +++ /dev/null diff --git a/pyagl/tests/test_audiomixer.py b/pyagl/tests/test_audiomixer.py new file mode 100644 index 0000000..de07944 --- /dev/null +++ b/pyagl/tests/test_audiomixer.py @@ -0,0 +1,76 @@ +import asyncio +import os +import pytest + +from pyagl.services.base import AFBResponse, AFBT +from pyagl.services.audiomixer import AudioMixerService as AMS +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 AMS(ip=address, port=port) + yield ams + await ams.websocket.close() + + +async def test_list_controls(event_loop, service: AMS): + msgid = await service.list_controls() + resp = await service.afbresponse() + assert msgid == resp.msgid + assert resp.type == AFBT.RESPONSE + assert resp.status == 'success' + + +async def test_volume_verb(event_loop, service: AMS): + msgid = await service.volume() + resp = await service.afbresponse() + assert msgid == resp.msgid + 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() + assert msgid == resp.msgid + 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() + assert msgid == resp.msgid + 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() + assert msgid == resp.msgid + 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() + assert msgid == resp.msgid + 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() + assert msgid == resp.msgid + assert resp.type == AFBT.RESPONSE + assert resp.status == 'success' diff --git a/tests/test_bluetooth.py b/tests/test_bluetooth.py deleted file mode 100644 index d6b899e..0000000 --- a/tests/test_bluetooth.py +++ /dev/null @@ -1,104 +0,0 @@ -import asyncio -import os -import pytest -from bluetooth import BluetoothService as BTS -import logging -from aglbaseservice import AFBResponse, AFBT - -logger = logging.getLogger(f'pytest-{BTS.service}') -logger.setLevel(logging.DEBUG) -pytestmark = pytest.mark.asyncio - - -@pytest.fixture(scope='module') -def event_loop(): - loop = asyncio.get_event_loop() - yield loop - loop.close() - - -@pytest.fixture(scope='module') -async def service(): - address = os.environ.get('AGL_TGT_IP', 'localhost') - port = os.environ.get('AGL_TGT_PORT', None) - - bts = await BTS(ip=address, port=port) - yield bts - await bts.websocket.close() - - -@pytest.mark.xfail -@pytest.fixture(scope='module') -def btaddr(): - bthtestaddr = os.environ.get('AGL_TEST_BT_ADDR', None) - if not bthtestaddr: - pytest.xfail('No test bluetooth test address set in environment variables') - - return bthtestaddr - - -@pytest.mark.dependency -async def test_default_adapter(event_loop, service: BTS): - id = await service.default_adapter() - resp = AFBResponse(await service.response()) - assert resp.status == 'success', resp - assert 'adapter' in resp.data.keys() - assert resp.data['adapter'] == 'hci0' - - -@pytest.mark.dependency(depends=['test_default_adapter']) -async def test_managed_objects(event_loop, service: BTS): - id = await service.managed_objects() - resp = AFBResponse(await service.response()) - assert resp.status == 'success', str(resp) - - -@pytest.mark.dependency(depends=['test_default_adapter']) -async def test_has_single_adapter(event_loop, service: BTS): - id = await service.managed_objects() - resp = AFBResponse(await service.response()) - assert len(resp.data['adapters']) == 1, \ - f'Detected {len(resp.data["adapters"])} adapters. Multiple adapters may also break testing' - - -@pytest.mark.dependency(depends=['test_default_adapter']) -async def test_adapter_state(event_loop, service: BTS): - id = await service.adapter_state('hci0') - resp = AFBResponse(await service.response()) - assert resp.status == 'success', 'adapter state verb failed' - - -async def test_pairing_verb(event_loop, service: BTS, btaddr): - id = await service.pair(btaddr) - resp = await service.afbresponse() - assert id == resp.msgid - assert resp.status == 'success', f'pair verb failed - {resp.info}' - - -async def test_connect_verb(event_loop, service: BTS, btaddr): - id = await service.connect(btaddr) - resp = await service.afbresponse() - assert id == resp.msgid - assert resp.status == 'success', f'connect verb failed - {resp.info}' - - -async def test_disconnect_verb(event_loop, service: BTS, btaddr): - id = await service.disconnect(btaddr) - resp = await service.afbresponse() - assert id == resp.msgid - assert resp.status == 'success', f'disconnect verb failed - {resp.info}' - - -async def test_remove_pairing_verb(event_loop, service: BTS, btaddr): - id = await service.remove_device(btaddr) - resp = await service.afbresponse() - assert id == resp.msgid - assert resp.status == 'success' - - -@pytest.mark.xfail(reason='This is expected to fail because there has to be an ongoing pairing attempt') -async def test_confirm_pairing_verb(event_loop, service: BTS, btaddr): - id = await service.confirm_pairing(pincode='123456') - resp = await service.afbresponse() - assert id == resp.msgid - assert resp.status == 'success', f'confirm_pairing verb failed - {resp.info}' diff --git a/tests/test_gps.py b/tests/test_gps.py deleted file mode 100644 index 2e62405..0000000 --- a/tests/test_gps.py +++ /dev/null @@ -1,71 +0,0 @@ -import asyncio -import os -import pytest -from gps import GPSService as GPS -import logging -from aglbaseservice import AFBResponse, AFBT -from concurrent.futures import TimeoutError - -logger = logging.getLogger('pytest-gps') -logger.setLevel(logging.DEBUG) -pytestmark = pytest.mark.asyncio - -@pytest.fixture(scope='module') -def event_loop(): - loop = asyncio.get_event_loop() - yield loop - loop.close() - -@pytest.fixture(scope='module') -async def service(): - address = os.environ.get('AGL_TGT_IP', 'localhost') - port = os.environ.get('AGL_TGT_PORT', None) - gpss = await GPS(ip=address, port=port) - yield gpss - await gpss.websocket.close() - -# @pytest.fixture(scope='module') -# async def response(event_loop, service): -# async for _response in service.listener(): -# yield _response - -async def test_location_verb(event_loop, service: GPS): - id = await service.location() - resp = AFBResponse(await service.response()) - assert resp.msgid == id - -@pytest.mark.xfail # expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable -async def test_location_result(event_loop, service: GPS): - id = await service.location() - resp = AFBResponse(await service.response()) - assert resp.status == 'success' - - - -async def test_subscribe_location(event_loop, service: GPS): - id = await service.subscribe('location') - resp = AFBResponse(await service.response()) - assert resp.msgid == id - assert resp.status == 'success' - -async def test_unsubscribe(event_loop, service: GPS): - id = await service.unsubscribe('location') - resp = AFBResponse(await service.response()) - assert resp.msgid == id - assert resp.status == 'success' - -@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): - id = await service.subscribe('location') - resp = AFBResponse(await service.response()) - assert resp.msgid == id - assert resp.status == 'success' # successful subscription - try: - resp = await asyncio.wait_for(service.response(), 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") - - |