diff options
author | 2020-06-12 22:44:25 +0300 | |
---|---|---|
committer | 2020-06-12 22:44:25 +0300 | |
commit | 8a8b87e65c0b3d579f8ea420e23a9cd07528dfe1 (patch) | |
tree | d5ced56f53104781bf156b38dbad7aba48578ca1 /test_gps.py | |
parent | e2bb2d3ff3f909b9417040de4dd8ea876777a6ee (diff) |
Adding cookiecutter, preparing for setuptools, new services
Diffstat (limited to 'test_gps.py')
-rw-r--r-- | test_gps.py | 57 |
1 files changed, 0 insertions, 57 deletions
diff --git a/test_gps.py b/test_gps.py deleted file mode 100644 index e1ad15e..0000000 --- a/test_gps.py +++ /dev/null @@ -1,57 +0,0 @@ -import asyncio -import os -import pytest -from gps import GPSService as GPS -import logging -from aglbaseservice import AFBResponse, AFBT - -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') - gpss = await GPS(ip=address) - yield gpss - await gpss.websocket.close() - -# @pytest.fixture(scope='module') -# async def response(event_loop, service): -# async for _response in service.listener(): -# yield _response - -@pytest.mark.xfail # expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable -async def test_location(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' - -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 - - 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' - |