diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_bluetooth.py | 104 | ||||
| -rw-r--r-- | tests/test_gps.py | 71 | ||||
| -rw-r--r-- | tests/test_weather.py | 0 |
3 files changed, 175 insertions, 0 deletions
diff --git a/tests/test_bluetooth.py b/tests/test_bluetooth.py new file mode 100644 index 0000000..d6b899e --- /dev/null +++ b/tests/test_bluetooth.py @@ -0,0 +1,104 @@ +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 new file mode 100644 index 0000000..2e62405 --- /dev/null +++ b/tests/test_gps.py @@ -0,0 +1,71 @@ +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") + + diff --git a/tests/test_weather.py b/tests/test_weather.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/test_weather.py |