summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorEdi Feschiyan <edi.feschiyan@konsulko.com>2020-06-24 20:01:26 +0300
committerEdi Feschiyan <edi.feschiyan@konsulko.com>2020-06-24 20:01:26 +0300
commit434bcef7a61db661a5c0579746788ffb86904089 (patch)
tree8ca0e232c1fe34ec7d01c0430d00e7b938512d28
parent4f89ab88c04333d97dd2ee4a588303c07e55b7c5 (diff)
Adding bluetooth pbap service and tests
-rw-r--r--pyagl/services/bluetooth-pbap.py0
-rw-r--r--pyagl/services/bluetooth_pbap.py35
-rw-r--r--pyagl/tests/test_bluetooth.py1
-rw-r--r--pyagl/tests/test_bluetooth_pbap.py160
-rw-r--r--pyagl/tests/test_weather.py35
-rw-r--r--templates/{{cookiecutter.tests_dir}}/test_{{cookiecutter.service_slug}}.py3
6 files changed, 231 insertions, 3 deletions
diff --git a/pyagl/services/bluetooth-pbap.py b/pyagl/services/bluetooth-pbap.py
deleted file mode 100644
index e69de29..0000000
--- a/pyagl/services/bluetooth-pbap.py
+++ /dev/null
diff --git a/pyagl/services/bluetooth_pbap.py b/pyagl/services/bluetooth_pbap.py
new file mode 100644
index 0000000..6472b60
--- /dev/null
+++ b/pyagl/services/bluetooth_pbap.py
@@ -0,0 +1,35 @@
+from pyagl.services.base import AGLBaseService, AFBResponse
+import asyncio
+import os
+
+class BTPBAPService(AGLBaseService):
+ service = 'agl-service-bluetooth-pbap'
+ parser = AGLBaseService.getparser()
+
+ def __init__(self, ip, port=None, service='agl-service-bluetooth-pbap'):
+ super().__init__(api='bluetooth-pbap', ip=ip, port=port, service=service)
+
+ async def subscribe(self, event='status'):
+ return await super().subscribe(event)
+
+ async def unsubscribe(self, event='status'):
+ return await super().subscribe(event)
+
+ async def import_contacts(self):
+ return await super().request('import')
+
+ async def status(self):
+ return await super().request('status')
+
+ async def contacts(self):
+ return await super().request('contacts')
+
+ async def entry(self, handle, param='pb'):
+ return await super().request('entry', {'list': param, 'handle': handle})
+
+ async def search(self, number):
+ return await super().request('search', {'number': number})
+
+ async def history(self, param):
+ return await super().request('history', {'list': param})
+
diff --git a/pyagl/tests/test_bluetooth.py b/pyagl/tests/test_bluetooth.py
index 0670f5e..44cb613 100644
--- a/pyagl/tests/test_bluetooth.py
+++ b/pyagl/tests/test_bluetooth.py
@@ -15,7 +15,6 @@ pytestmark = pytest.mark.asyncio
def event_loop():
loop = asyncio.get_event_loop()
yield loop
- loop.close()
@pytest.fixture(scope='module')
diff --git a/pyagl/tests/test_bluetooth_pbap.py b/pyagl/tests/test_bluetooth_pbap.py
new file mode 100644
index 0000000..26db0ba
--- /dev/null
+++ b/pyagl/tests/test_bluetooth_pbap.py
@@ -0,0 +1,160 @@
+import asyncio
+import os
+import pytest
+import logging
+
+from pyagl.services.base import AFBResponse, AFBT
+from pyagl.services.bluetooth_pbap import BTPBAPService as PBAP
+
+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)
+
+ ams = await PBAP(ip=address, port=port)
+ yield ams
+ await ams.websocket.close()
+
+@pytest.fixture(scope='module')
+def searchvcf():
+ vcf = os.environ.get('AGL_PBAP_VCF')
+ if not vcf:
+ pytest.xfail('Please export AGL_PBAP_VCF with an existing .vcf string e.g. 63602.vcf')
+
+ return vcf
+
+
+@pytest.mark.dependency
+async def test_status(event_loop, service: PBAP):
+ msgid = await service.status()
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert isinstance(resp.data, dict)
+ assert 'connected' in resp.data
+ if not resp.data['connected']:
+ pytest.xfail('BT PBAP is not currently connected to the phone')
+ assert resp.data['connected'] is True
+
+
+async def test_search(event_loop, service: PBAP):
+ msgid = await service.search("+359887224379")
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+
+
+@pytest.mark.dependency(depends=['test_status'])
+async def test_import_contacts(event_loop, service: PBAP):
+ msgid = await service.import_contacts()
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ assert isinstance(resp.data['vcards'], list)
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+ print(f' import verb from OBEX returned {len(vcards)} .vcf objects ')
+
+
+@pytest.mark.dependency(depends=['test_import_contacts'])
+async def test_contacts(event_loop, service: PBAP):
+ msgid = await service.contacts()
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ assert isinstance(resp.data['vcards'], list)
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+ print(f' contacts verb returned {len(vcards)} cached .vcf objects ')
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_history_incoming_calls(event_loop, service: PBAP):
+ msgid = await service.history(param='ich')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_history_outgoing_calls(event_loop, service: PBAP):
+ msgid = await service.history(param='och')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_history_missed_calls(event_loop, service: PBAP):
+ msgid = await service.history(param='mch')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_history_combined_calls(event_loop, service: PBAP):
+ msgid = await service.history(param='cch')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_entry_phonebook(event_loop, service: PBAP, searchvcf):
+ msgid = await service.entry(handle=searchvcf)
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ assert len(resp.data['vcards']) > 0
+
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_incoming_calls(event_loop, service: PBAP):
+ msgid = await service.entry(handle='1.vcf', param='ich')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_outgoing_calls(event_loop, service: PBAP):
+ msgid = await service.entry(handle='1.vcf', param='och')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_missed_calls(event_loop, service: PBAP):
+ msgid = await service.entry(handle='1.vcf', param='mch')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
+
+@pytest.mark.dependency(depends=['test_contacts'])
+async def test_combined_calls(event_loop, service: PBAP):
+ msgid = await service.entry(handle='1.vcf', param='cch')
+ resp = await service.afbresponse()
+ assert resp.status == 'success'
+ assert 'vcards' in resp.data
+ vcards = resp.data['vcards']
+ assert len(vcards) > 0
+
diff --git a/pyagl/tests/test_weather.py b/pyagl/tests/test_weather.py
index e69de29..696b1d8 100644
--- a/pyagl/tests/test_weather.py
+++ b/pyagl/tests/test_weather.py
@@ -0,0 +1,35 @@
+import asyncio
+import os
+import pytest
+import urllib3
+import logging
+from pyagl.services.base import AFBResponse, AFBT
+from pyagl.services.weather import WeatherService as ws
+
+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)
+ gpss = await ws(ip=address, port=port)
+ yield gpss
+ await gpss.websocket.close()
+
+async def test_apikey(event_loop, service: ws):
+ msgid = await service.apikey()
+ resp = await service.afbresponse()
+ assert resp.msgid == msgid
+ assert resp.data['api_key'] == 'a860fa437924aec3d0360cc749e25f0e'
+
+async def test_current_weather(event_loop, service: ws):
+ msgid = await service.current_weather()
+ resp = await service.afbresponse()
+ assert resp.msgid == msgid
+ assert 'sys' in resp.data
+
diff --git a/templates/{{cookiecutter.tests_dir}}/test_{{cookiecutter.service_slug}}.py b/templates/{{cookiecutter.tests_dir}}/test_{{cookiecutter.service_slug}}.py
index e193205..0a93ec0 100644
--- a/templates/{{cookiecutter.tests_dir}}/test_{{cookiecutter.service_slug}}.py
+++ b/templates/{{cookiecutter.tests_dir}}/test_{{cookiecutter.service_slug}}.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():
@@ -20,4 +19,4 @@ async def service():
port = os.environ.get('AGL_TGT_PORT', None)
svc = await {{cookiecutter.classname}}(ip=address, port=port)
yield svc
- await svc.websocket.close()
+