diff options
| -rw-r--r-- | pyagl/services/bluetooth-pbap.py | 0 | ||||
| -rw-r--r-- | pyagl/services/bluetooth_pbap.py | 35 | ||||
| -rw-r--r-- | pyagl/tests/test_bluetooth.py | 1 | ||||
| -rw-r--r-- | pyagl/tests/test_bluetooth_pbap.py | 160 | ||||
| -rw-r--r-- | pyagl/tests/test_weather.py | 35 | ||||
| -rw-r--r-- | templates/{{cookiecutter.tests_dir}}/test_{{cookiecutter.service_slug}}.py | 3 |
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() + |