diff options
| -rw-r--r-- | pyagl/services/network.py | 51 | ||||
| -rw-r--r-- | pyagl/services/taskmanager.py | 8 | ||||
| -rw-r--r-- | pyagl/tests/test_geoclue.py | 30 | ||||
| -rw-r--r-- | pyagl/tests/test_gps.py | 3 | ||||
| -rw-r--r-- | pyagl/tests/test_network.py | 96 | ||||
| -rw-r--r-- | pyagl/tests/test_nfc.py | 33 | ||||
| -rw-r--r-- | setup.py | 2 |
7 files changed, 220 insertions, 3 deletions
diff --git a/pyagl/services/network.py b/pyagl/services/network.py index e69de29..8aa6f82 100644 --- a/pyagl/services/network.py +++ b/pyagl/services/network.py @@ -0,0 +1,51 @@ +from pyagl.services.base import AGLBaseService, AFBResponse +import asyncio +import os + + +class NetworkService(AGLBaseService): + service = 'agl-service-network' + parser = AGLBaseService.getparser() + + def __init__(self, ip, port=None, service='agl-service-network'): + super().__init__(api='network-manager', ip=ip, port=port, service=service) + + async def state(self): + return await self.request('state') + + async def offline(self, value=True): + return await self.request('offline', {'value': value}) + + async def technologies(self): + return await self.request('technologies') + + async def get_property(self, technology): + return await self.request('get_property', {'technology': technology}) + + async def set_property(self, technology: str, properties: dict): + return await self.request('set_property', {'technology': technology, 'properties': properties}) + + async def services(self): + return await self.request('services') + + async def enable_technology(self, technology): + return await self.request('enable_technology', {'technology': technology}) + + async def disable_technology(self, technology): + return await self.request('disable_technology', {'technology': technology}) + + async def scan_services(self, technology): + return await self.request('scan_services', {'technology': technology}) + + async def remove_service(self, service): + return await self.request('remove_service', {'service': service}) + + async def connect_service(self, service): + return await self.request('connect_service', {'service': service}) + + async def disconnect_service(self, service): + return await self.request('disconnect_service', {'service': service}) + + async def agent_response(self): + pass + diff --git a/pyagl/services/taskmanager.py b/pyagl/services/taskmanager.py index e69de29..c3cd3b6 100644 --- a/pyagl/services/taskmanager.py +++ b/pyagl/services/taskmanager.py @@ -0,0 +1,8 @@ +from pyagl.services.base import AGLBaseService, AFBResponse +import asyncio +import os + + +class TaskManagerService(AGLBaseService): + service = 'agl-service-taskmanager' + parser = AGLBaseService.getparser() diff --git a/pyagl/tests/test_geoclue.py b/pyagl/tests/test_geoclue.py new file mode 100644 index 0000000..40b995a --- /dev/null +++ b/pyagl/tests/test_geoclue.py @@ -0,0 +1,30 @@ +import asyncio +import os +import pytest +import logging +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') + port = os.environ.get('AGL_TGT_PORT', None) + svc = await gcs(ip=address, port=port) + yield svc + await svc.websocket.close() + + +async def test_location(event_loop, service: gcs): + msgid = await service.location() + resp = await service.afbresponse() + assert resp.msgid == msgid + assert resp.type == AFBT.RESPONSE + assert resp.status == 'success' + diff --git a/pyagl/tests/test_gps.py b/pyagl/tests/test_gps.py index 7e56d1e..cb28be6 100644 --- a/pyagl/tests/test_gps.py +++ b/pyagl/tests/test_gps.py @@ -12,7 +12,6 @@ pytestmark = pytest.mark.asyncio def event_loop(): loop = asyncio.get_event_loop() yield loop - loop.close() @pytest.fixture(scope='module') async def service(): @@ -39,7 +38,7 @@ async def test_location_result(event_loop, service: GPS): assert resp.status == 'success' async def test_subscribe_verb(event_loop, service: GPS): - msgid = await service.subscribe("") + msgid = await service.subscribe() resp = AFBResponse(await service.response()) assert resp.msgid == msgid assert resp.status == 'success' diff --git a/pyagl/tests/test_network.py b/pyagl/tests/test_network.py new file mode 100644 index 0000000..f1bc56d --- /dev/null +++ b/pyagl/tests/test_network.py @@ -0,0 +1,96 @@ +import asyncio +import os +import pytest +import logging +from pyagl.services.base import AFBResponse, AFBT +from pyagl.services.network import NetworkService as NS + +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) + ns = await NS(ip=address, port=port) + yield ns + await ns.websocket.close() + +@pytest.fixture(scope='module') +def expected_available_techs(): + techs = os.environ.get('AGL_AVAILABLE_TECHS', 'wifi,ethernet,bluetooth').split(',') + return techs + +async def test_state(event_loop, service: NS): + msgid = await service.state() + resp = await service.afbresponse() + 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): + msgid = await service.offline(True) + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +async def test_global_online(event_loop, service: NS): + msgid = await service.offline(False) + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + +@pytest.mark.dependency +async def test_technologies_verb(event_loop, service: NS): + msgid = await service.technologies() + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + assert 'values' in resp.data + + +@pytest.mark.dependency(depends=['test_technologies_verb']) +async def test_expected_existing_technologies(event_loop, service: NS, expected_available_techs): + msgid = await service.technologies() + resp = await service.afbresponse() + techs = [t['technology'] for t in resp.data['values']] + for t in expected_available_techs: + assert t in techs, f'"{t}" technology is expected to be available, but it is not' + +@pytest.mark.dependency(depends=['test_expected_existing_technologies']) +async def test_get_property(event_loop, service: NS, expected_available_techs): + for t in expected_available_techs: + msgid = await service.get_property(t) + resp = await service.afbresponse() + 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())) + assert len(diverging_fields) == 0, f'the following property fields are diverging from the expected: {diverging_fields}' + + +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') + msgid = await service.set_property('wifi', {'tethering': True, + 'tetheringidentifier': 'AGL', + 'tetheringpassphrase': 'agltestwifi'}) + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +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') + 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 diff --git a/pyagl/tests/test_nfc.py b/pyagl/tests/test_nfc.py new file mode 100644 index 0000000..3b4376f --- /dev/null +++ b/pyagl/tests/test_nfc.py @@ -0,0 +1,33 @@ +import asyncio +import os +import pytest +import logging +from pyagl.services.base import AFBResponse, AFBT +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') + port = os.environ.get('AGL_TGT_PORT', None) + svc = await nfcs(ip=address, port=port) + yield svc + await svc.websocket.close() + +async def subscribe(event_loop, service: nfcs): + msgid = service.subscribe() + resp = await service.afbresponse() + assert resp.msgid == msgid + + +async def unsubscribe(event_loop, service: nfcs): + msgid = service.unsubscribe() + resp = await service.afbresponse() + assert resp.msgid == msgid @@ -14,7 +14,7 @@ setuptools.setup( url="https://github.com/refresher/pyagl", packages=setuptools.find_packages(), license="Apache 2.0", - install_requires=['websockets', 'parse', 'asyncssh', 'pytest', 'pytest-asyncio'], + install_requires=['websockets', 'parse', 'asyncssh', 'pytest', 'pytest-asyncio', 'pytest-depends'], classifiers=[ 'Development Status :: 4 - Beta', 'Environment :: Console', |