diff options
-rw-r--r-- | pyagl/services/signal_composer.py | 2 | ||||
-rw-r--r-- | pyagl/tests/test_signal_composer.py | 52 |
2 files changed, 51 insertions, 3 deletions
diff --git a/pyagl/services/signal_composer.py b/pyagl/services/signal_composer.py index 2b0c4a6..d6b90cc 100644 --- a/pyagl/services/signal_composer.py +++ b/pyagl/services/signal_composer.py @@ -2,8 +2,6 @@ from pyagl.services.base import AGLBaseService, AFBResponse import asyncio import os -verbs = ['subscribe', 'unsubscribe', 'get', 'addObjects', 'list'] - class SignalComposerService(AGLBaseService): service = 'service-signal-composer' diff --git a/pyagl/tests/test_signal_composer.py b/pyagl/tests/test_signal_composer.py index 9e294a0..6dd872c 100644 --- a/pyagl/tests/test_signal_composer.py +++ b/pyagl/tests/test_signal_composer.py @@ -5,7 +5,8 @@ import logging from pyagl.services.base import AFBResponse, AFBT from pyagl.services.signal_composer import SignalComposerService as scs -pytestmark = pytest.mark.asyncio + +pytestmark = [pytest.mark.asyncio, pytest.mark.can] @pytest.fixture(scope='module') @@ -22,3 +23,52 @@ async def service(): yield ns await ns.websocket.close() + +@pytest.mark.regular +async def test_list(event_loop, service: scs): + msgid = await service.list() + resp = await service.afbresponse() + assert resp.status == 'success' + print(resp) + +@pytest.mark.regular +async def test_getNoFilter(event_loop, service: scs): + msgid = await service.get({'signal': 'fuel_level'}) + resp = await service.afbresponse() + assert resp.status == 'success' + + +# while rewriting the old lua tests this one tries to 'get' 'odometer' signal but its not currently present +# @pytest.mark.regular +# async def test_getFilterAvg(event_loop, service: scs): +# msgid = await service.get({'signal': 'odometer', 'options': {'average': 10}}) +# resp = await service.afbresponse() +# print(resp) +# assert resp.status == 'success' + + +@pytest.mark.regular +async def test_getFilterMin(event_loop, service: scs): + msgid = await service.get({'signal': 'latitude', 'options': {'minimum': 10}}) + resp = await service.afbresponse() + assert resp.status == 'success' + + +@pytest.mark.regular +async def test_getFilterMax(event_loop, service: scs): + msgid = await service.get({'signal': 'vehicle_speed', 'options': {'maximum': 10}}) + resp = await service.afbresponse() + assert resp.status == 'success' + + +@pytest.mark.regular +async def test_subscribe(event_loop, service: scs): + msgid = await service.subscribe({'signal': 'longitude'}) + resp = await service.afbresponse() + assert resp.status == 'success' + +@pytest.mark.regular +async def test_unsubscribe(event_loop, service: scs): + msgid = await service.unsubscribe({'service': 'longitude'}) + resp = await service.afbresponse() + assert resp.status == 'success'
\ No newline at end of file |