summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--pyagl/services/network.py51
-rw-r--r--pyagl/services/taskmanager.py8
-rw-r--r--pyagl/tests/test_geoclue.py30
-rw-r--r--pyagl/tests/test_gps.py3
-rw-r--r--pyagl/tests/test_network.py96
-rw-r--r--pyagl/tests/test_nfc.py33
-rw-r--r--setup.py2
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
diff --git a/setup.py b/setup.py
index afb45ab..1ddf39d 100644
--- a/setup.py
+++ b/setup.py
@@ -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',