aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorEdi Feschiyan <edi.feschiyan@konsulko.com>2020-08-07 16:59:54 +0300
committerEdi Feschiyan <efeschiyan@pm.me>2020-08-07 16:59:54 +0300
commit4b4deff8e1e27ca99aec41aad3dbd46303792729 (patch)
treedcd5e9cee4815bbd0dad782a6e1bde5ccca850a9
parent1783f229db62c47e990909bc193cc9bea963ca2f (diff)
Registering "regular" and "hwrequired" test markers, adding them to tests
Enabling custom pytest --lava option for LAVA CI integration Enabling tox for CI bluetooth::default_adapter should support setting a default one according to documentation in README.md
-rw-r--r--conftest.py32
-rw-r--r--pyagl/services/bluetooth.py8
-rw-r--r--pyagl/tests/test_audiomixer.py14
-rw-r--r--pyagl/tests/test_bluetooth.py42
-rw-r--r--pyagl/tests/test_bluetooth_map.py14
-rw-r--r--pyagl/tests/test_bluetooth_pbap.py41
-rw-r--r--pyagl/tests/test_geoclue.py6
-rw-r--r--pyagl/tests/test_gps.py21
-rw-r--r--pyagl/tests/test_homescreen.py4
-rw-r--r--pyagl/tests/test_network.py66
-rw-r--r--pyagl/tests/test_nfc.py4
-rw-r--r--pyagl/tests/test_weather.py13
-rw-r--r--pytest.ini15
-rw-r--r--tox.ini17
14 files changed, 270 insertions, 27 deletions
diff --git a/conftest.py b/conftest.py
new file mode 100644
index 0000000..2176354
--- /dev/null
+++ b/conftest.py
@@ -0,0 +1,32 @@
+import pytest
+
+
+def pytest_addoption(parser):
+ parser.addoption('-L', '--lava', action='store_true', help='enable LAVA signals')
+
+
+def lava_result_convert(pytest_outcome):
+ """ Convert the pytestoutcome to the string expected by LAVA."""
+ if pytest_outcome == 'passed':
+ return 'pass'
+ elif pytest_outcome == 'skipped':
+ return 'pass'
+ elif pytest_outcome == 'xfailed':
+ return 'pass'
+ else:
+ return'fail'
+
+
+def pytest_report_teststatus(config, report):
+ """ Insert strings that LAVA expects to capture test results."""
+# Get pytest test name and remove the 'test_' prefix
+ if config.getoption('--lava'):
+ test_name = report.location[2][5:]
+ if report.when == 'setup':
+ print('\n')
+ print(f'<LAVA_SIGNAL_STARTTC {test_name}>')
+ elif report.when == 'call':
+ test_result = lava_result_convert(report.outcome)
+ print('\n')
+ print(f'<LAVA_SIGNAL_ENDTC {test_name}>')
+ print(f'<LAVA_SIGNAL_TESTCASE TEST_CASE_ID={test_name} RESULT={test_result}>')
diff --git a/pyagl/services/bluetooth.py b/pyagl/services/bluetooth.py
index 296d5ee..7e413b3 100644
--- a/pyagl/services/bluetooth.py
+++ b/pyagl/services/bluetooth.py
@@ -44,8 +44,12 @@ class BluetoothService(AGLBaseService):
return await self.request('adapter_state', p)
- async def default_adapter(self):
- return await self.request('default_adapter', "")
+ async def default_adapter(self, adapter=None):
+ if adapter is not None:
+ return await self.request('default_adapter', {'adapter': adapter})
+ else:
+ return await self.request('default_adapter')
+
async def connect(self, device: str = 'hci0'):
return await self.request('connect', {'device': device})
diff --git a/pyagl/tests/test_audiomixer.py b/pyagl/tests/test_audiomixer.py
index 1f49760..640c6cb 100644
--- a/pyagl/tests/test_audiomixer.py
+++ b/pyagl/tests/test_audiomixer.py
@@ -25,6 +25,8 @@ async def service():
await ams.websocket.close()
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_list_controls(event_loop, service: AMS):
msgid = await service.list_controls()
resp = await service.afbresponse()
@@ -33,6 +35,8 @@ async def test_list_controls(event_loop, service: AMS):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_volume_verb(event_loop, service: AMS):
msgid = await service.volume()
resp = await service.afbresponse()
@@ -41,6 +45,8 @@ async def test_volume_verb(event_loop, service: AMS):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_set_volume0(event_loop, service: AMS):
msgid = await service.volume(value=0)
resp = await service.afbresponse()
@@ -49,6 +55,8 @@ async def test_set_volume0(event_loop, service: AMS):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_set_maxvolume(event_loop, service: AMS):
msgid = await service.volume(value=1)
resp = await service.afbresponse()
@@ -57,6 +65,8 @@ async def test_set_maxvolume(event_loop, service: AMS):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_get_mute(event_loop, service: AMS):
msgid = await service.mute()
resp = await service.afbresponse()
@@ -65,6 +75,8 @@ async def test_get_mute(event_loop, service: AMS):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_set_mute(event_loop, service: AMS):
msgid = await service.mute(value=1)
resp = await service.afbresponse()
@@ -73,6 +85,8 @@ async def test_set_mute(event_loop, service: AMS):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.audiomixer
async def test_set_unmute(event_loop, service: AMS):
msgid = await service.mute(value=0)
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_bluetooth.py b/pyagl/tests/test_bluetooth.py
index 14da24b..5f9232f 100644
--- a/pyagl/tests/test_bluetooth.py
+++ b/pyagl/tests/test_bluetooth.py
@@ -28,6 +28,7 @@ async def service():
@pytest.mark.xfail
+@pytest.mark.bluetooth
@pytest.fixture(scope='module')
def btaddr():
bthtestaddr = os.environ.get('AGL_BT_TEST_ADDR', None)
@@ -38,71 +39,90 @@ def btaddr():
return bthtestaddr
+@pytest.mark.regular
+@pytest.mark.bluetooth
@pytest.mark.dependency
async def test_default_adapter(event_loop, service: BTS):
- msgid = await service.default_adapter()
+ msgid = await service.default_adapter('hci0')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
assert 'adapter' in resp.data.keys()
assert resp.data['adapter'] == 'hci0'
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_subscribe_device_changes(event_loop, service: BTS):
msgid = await service.subscribe('device_changes')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_unsubscribe_device_changes(event_loop, service: BTS):
msgid = await service.unsubscribe('device_changes')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+
-
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_subscribe_adapter_changes(event_loop, service: BTS):
msgid = await service.subscribe('adapter_changes')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_unsubscribe_adapter_changes(event_loop, service: BTS):
msgid = await service.unsubscribe('adapter_changes')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_subscribe_media(event_loop, service: BTS):
msgid = await service.subscribe('media')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_unsubscribe_media(event_loop, service: BTS):
msgid = await service.unsubscribe('media')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_subscribe_agent(event_loop, service: BTS):
msgid = await service.subscribe('agent')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.regular
+@pytest.mark.bluetooth
async def test_unsubscribe_agent(event_loop, service: BTS):
msgid = await service.unsubscribe('agent')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
-
+@pytest.mark.regular
+@pytest.mark.bluetooth
@pytest.mark.dependency(depends=['test_default_adapter'])
async def test_managed_objects(event_loop, service: BTS):
msgid = await service.managed_objects()
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
-
-@pytest.mark.dependency(depends=['test_default_adapter'])
+@pytest.mark.hwrequired
+@pytest.mark.bluetooth
async def test_has_single_adapter(event_loop, service: BTS):
msgid = await service.managed_objects()
resp = await service.afbresponse()
@@ -110,6 +130,8 @@ async def test_has_single_adapter(event_loop, service: BTS):
f'Detected {len(resp.data["adapters"])} adapters. Multiple adapters may also affect testing'
+@pytest.mark.regular
+@pytest.mark.bluetooth
@pytest.mark.dependency(depends=['test_default_adapter'])
async def test_adapter_state(event_loop, service: BTS):
msgid = await service.adapter_state('hci0')
@@ -117,6 +139,8 @@ async def test_adapter_state(event_loop, service: BTS):
assert resp.status == 'success', 'adapter state verb failed'
+@pytest.mark.hwrequired
+@pytest.mark.bluetooth
async def test_pairing_verb(event_loop, service: BTS, btaddr):
msgid = await service.pair(btaddr)
resp = await service.afbresponse()
@@ -124,6 +148,8 @@ async def test_pairing_verb(event_loop, service: BTS, btaddr):
assert resp.status == 'success', f'pair verb failed - {resp.info}'
+@pytest.mark.hwrequired
+@pytest.mark.bluetooth
async def test_connect_verb(event_loop, service: BTS, btaddr):
msgid = await service.connect(btaddr)
resp = await service.afbresponse()
@@ -131,6 +157,8 @@ async def test_connect_verb(event_loop, service: BTS, btaddr):
assert resp.status == 'success', f'connect verb failed - {resp.info}'
+@pytest.mark.hwrequired
+@pytest.mark.bluetooth
async def test_disconnect_verb(event_loop, service: BTS, btaddr):
msgid = await service.disconnect(btaddr)
resp = await service.afbresponse()
@@ -138,6 +166,8 @@ async def test_disconnect_verb(event_loop, service: BTS, btaddr):
assert resp.status == 'success', f'disconnect verb failed - {resp.info}'
+@pytest.mark.hwrequired
+@pytest.mark.bluetooth
async def test_remove_pairing_verb(event_loop, service: BTS, btaddr):
msgid = await service.remove_device(btaddr)
resp = await service.afbresponse()
@@ -145,6 +175,8 @@ async def test_remove_pairing_verb(event_loop, service: BTS, btaddr):
assert resp.status == 'success'
+@pytest.mark.hwrequired
+@pytest.mark.bluetooth
@pytest.mark.xfail(reason='This is expected to fail because there has to be an ongoing pairing attempt')
async def test_confirm_pairing_verb(event_loop, service: BTS, btaddr):
msgid = await service.confirm_pairing(pincode='123456')
diff --git a/pyagl/tests/test_bluetooth_map.py b/pyagl/tests/test_bluetooth_map.py
index 68b885e..871dc76 100644
--- a/pyagl/tests/test_bluetooth_map.py
+++ b/pyagl/tests/test_bluetooth_map.py
@@ -40,6 +40,8 @@ def composetext():
return text
+@pytest.mark.hwrequired
+@pytest.mark.btmap
@pytest.mark.xfail(reason='Expected to fail if no messages are found')
@pytest.fixture(scope='module')
async def messages(service: BMP):
@@ -50,6 +52,8 @@ async def messages(service: BMP):
yield resp.data['messages']
+@pytest.mark.hwrequired
+@pytest.mark.btmap
@pytest.mark.dependency
async def test_list_messages(event_loop, service: BMP):
msgid = await service.list_messages()
@@ -57,6 +61,8 @@ async def test_list_messages(event_loop, service: BMP):
assert resp.msgid == msgid
+@pytest.mark.hwrequired
+@pytest.mark.btmap
@pytest.mark.dependency(depends=['test_list_messages'])
@pytest.mark.xfail(reason='Expected to fail if there are no messages, need message id to test the verb')
async def test_message_verb(event_loop, service, messages):
@@ -66,6 +72,8 @@ async def test_message_verb(event_loop, service, messages):
resp = await service.afbresponse()
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.btmap
@pytest.mark.dependency
async def test_subscribe_notifications(event_loop, service: BMP):
msgid = await service.subscribe('notification')
@@ -74,6 +82,8 @@ async def test_subscribe_notifications(event_loop, service: BMP):
assert resp.status == 'success'
+@pytest.mark.hwrequired
+@pytest.mark.btmap
@pytest.mark.dependency
async def test_compose_message(event_loop, service: BMP, recipient, composetext):
msgid = await service.compose(recipient, composetext)
@@ -81,6 +91,8 @@ async def test_compose_message(event_loop, service: BMP, recipient, composetext)
assert resp.status == 'success'
+@pytest.mark.hwrequired
+@pytest.mark.btmap
@pytest.mark.dependency(depends=['test_compose_message', 'test_subscribe_notifications'])
async def test_message_to_self(event_loop, service: BMP, recipient, composetext):
resp = await service.afbresponse()
@@ -91,6 +103,8 @@ async def test_message_to_self(event_loop, service: BMP, recipient, composetext)
assert resp.data['message'] == composetext, 'Message mismatching'
+@pytest.mark.hwrequired
+@pytest.mark.btmap
@pytest.mark.dependency(depends=['test_subscribe_notifications'])
async def test_unsubscribe_notifications(event_loop, service: BMP):
msgid = await service.unsubscribe('notification')
diff --git a/pyagl/tests/test_bluetooth_pbap.py b/pyagl/tests/test_bluetooth_pbap.py
index c8e35b0..d13ede2 100644
--- a/pyagl/tests/test_bluetooth_pbap.py
+++ b/pyagl/tests/test_bluetooth_pbap.py
@@ -42,6 +42,8 @@ def searchvcf():
return vcf
+@pytest.mark.regular
+@pytest.mark.btpbap
@pytest.mark.dependency
async def test_status(event_loop, service: PBAP):
msgid = await service.status()
@@ -53,13 +55,15 @@ async def test_status(event_loop, service: PBAP):
pytest.xfail('BT PBAP is not currently connected to the phone')
assert resp.data['connected'] is True
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
async def test_search(event_loop, service: PBAP, phonenumber):
msgid = await service.search(phonenumber)
resp = await service.afbresponse()
assert resp.status == 'success'
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_status'])
async def test_import_contacts(event_loop, service: PBAP):
msgid = await service.import_contacts()
@@ -71,7 +75,8 @@ async def test_import_contacts(event_loop, service: PBAP):
assert len(vcards) > 0
print(f' import verb from OBEX returned {len(vcards)} .vcf objects ')
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_import_contacts'])
async def test_contacts(event_loop, service: PBAP):
msgid = await service.contacts()
@@ -83,7 +88,8 @@ async def test_contacts(event_loop, service: PBAP):
assert len(vcards) > 0
print(f' contacts verb returned {len(vcards)} cached .vcf objects ')
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_incoming_calls(event_loop, service: PBAP):
msgid = await service.history(param='ich')
@@ -93,7 +99,8 @@ async def test_history_incoming_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_outgoing_calls(event_loop, service: PBAP):
msgid = await service.history(param='och')
@@ -103,7 +110,8 @@ async def test_history_outgoing_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_missed_calls(event_loop, service: PBAP):
msgid = await service.history(param='mch')
@@ -113,7 +121,8 @@ async def test_history_missed_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_history_combined_calls(event_loop, service: PBAP):
msgid = await service.history(param='cch')
@@ -123,14 +132,16 @@ async def test_history_combined_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_entry_phonebook_verb(event_loop, service: PBAP, searchvcf):
msgid = await service.entry(handle=searchvcf)
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_entry_phonebook_verb'])
async def test_entry_phonebook(event_loop, service: PBAP, searchvcf):
msgid = await service.entry(handle=searchvcf)
@@ -139,7 +150,8 @@ async def test_entry_phonebook(event_loop, service: PBAP, searchvcf):
if resp.data['vcards'] is None:
pytest.xfail('The response did not contain "vcards" entries')
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_incoming_calls(event_loop, service: PBAP):
msgid = await service.entry(handle='1.vcf', param='ich')
@@ -149,7 +161,8 @@ async def test_incoming_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_outgoing_calls(event_loop, service: PBAP):
msgid = await service.entry(handle='1.vcf', param='och')
@@ -159,7 +172,8 @@ async def test_outgoing_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_missed_calls(event_loop, service: PBAP):
msgid = await service.entry(handle='1.vcf', param='mch')
@@ -169,7 +183,8 @@ async def test_missed_calls(event_loop, service: PBAP):
vcards = resp.data['vcards']
assert len(vcards) > 0
-
+@pytest.mark.btpbap
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_contacts'])
async def test_combined_calls(event_loop, service: PBAP):
msgid = await service.entry(handle='1.vcf', param='cch')
diff --git a/pyagl/tests/test_geoclue.py b/pyagl/tests/test_geoclue.py
index f63533e..d5a8853 100644
--- a/pyagl/tests/test_geoclue.py
+++ b/pyagl/tests/test_geoclue.py
@@ -23,6 +23,8 @@ async def service():
await svc.websocket.close()
+@pytest.mark.regular
+@pytest.mark.geoclue
async def test_location(event_loop, service: gcs):
msgid = await service.location()
resp = await service.afbresponse()
@@ -31,6 +33,8 @@ async def test_location(event_loop, service: gcs):
assert resp.status == 'success'
+@pytest.mark.regular
+@pytest.mark.geoclue
async def test_subscribe(event_loop, service: gcs):
msgid = await service.subscribe()
resp = await service.afbresponse()
@@ -40,6 +44,8 @@ async def test_subscribe(event_loop, service: gcs):
assert event.api == f'{service.api}/location'
+@pytest.mark.regular
+@pytest.mark.geoclue
async def test_unsubscribe(event_loop, service: gcs):
msgid = await service.unsubscribe()
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_gps.py b/pyagl/tests/test_gps.py
index db3a7a4..2753b89 100644
--- a/pyagl/tests/test_gps.py
+++ b/pyagl/tests/test_gps.py
@@ -28,13 +28,15 @@ async def service():
# async for _response in service.listener():
# yield _response
-
+@pytest.mark.gps
+@pytest.mark.regular
async def test_location_verb(event_loop, service: GPS):
msgid = await service.location()
resp = await service.afbresponse()
assert resp.msgid == msgid
-
+@pytest.mark.gps
+@pytest.mark.regular
@pytest.mark.xfail(reason='expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable')
async def test_location_result(event_loop, service: GPS):
msgid = await service.location()
@@ -42,18 +44,26 @@ async def test_location_result(event_loop, service: GPS):
assert resp.status == 'success'
+@pytest.mark.gps
+@pytest.mark.regular
async def test_subscribe_verb(event_loop, service: GPS):
msgid = await service.subscribe()
resp = await service.afbresponse()
assert resp.msgid == msgid
assert resp.status == 'success'
+
+@pytest.mark.gps
+@pytest.mark.regular
@pytest.mark.dependency
async def test_enable_recording(event_loop, service: GPS):
msgid = await service.record()
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+
+@pytest.mark.gps
+@pytest.mark.regular
@pytest.mark.dependency(depends=['test_enable_recording'])
async def test_disable_recording(event_loop, service: GPS):
msgid = await service.record('off')
@@ -61,6 +71,8 @@ async def test_disable_recording(event_loop, service: GPS):
assert resp.status == 'success', resp.info
+@pytest.mark.gps
+@pytest.mark.regular
@pytest.mark.dependency
async def test_subscribe_location(event_loop, service: GPS):
msgid = await service.subscribe('location')
@@ -68,7 +80,8 @@ async def test_subscribe_location(event_loop, service: GPS):
assert resp.msgid == msgid
assert resp.status == 'success'
-
+@pytest.mark.gps
+@pytest.mark.hwrequired
@pytest.mark.dependency(depends=['test_subscribe_location'])
@pytest.mark.xfail # expecting this to fail because of "No 3D GNSS fix" and GPS is unavailable
async def test_location_events(event_loop, service: GPS):
@@ -86,6 +99,8 @@ async def test_location_events(event_loop, service: GPS):
pytest.xfail("Did not receive location event")
+@pytest.mark.gps
+@pytest.mark.regular
async def test_unsubscribe(event_loop, service: GPS):
msgid = await service.unsubscribe('location')
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_homescreen.py b/pyagl/tests/test_homescreen.py
index 1ac7330..758feb3 100644
--- a/pyagl/tests/test_homescreen.py
+++ b/pyagl/tests/test_homescreen.py
@@ -29,13 +29,13 @@ async def service():
# resp = await service.afbresponse()
# assert resp.status == 'success'
-
+@pytest.mark.regular
async def test_subscribe(event_loop, service: hcs):
msgid = await service.subscribe(event='tap_shortcut')
resp = await service.afbresponse()
assert resp.msgid == msgid
-
+@pytest.mark.regular
async def test_unsubscribe(event_loop, service: hcs):
msgid = await service.unsubscribe(event='tap_shortcut')
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_network.py b/pyagl/tests/test_network.py
index e69000b..7efce11 100644
--- a/pyagl/tests/test_network.py
+++ b/pyagl/tests/test_network.py
@@ -7,11 +7,13 @@ 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')
@@ -20,17 +22,24 @@ async def service():
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
+
+@pytest.mark.network
+@pytest.mark.regular
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.network
+@pytest.mark.regular
async def test_global_offline(event_loop, service: NS):
addr, _ = service.websocket.remote_address
print(f"Remote address is {addr}")
@@ -41,11 +50,16 @@ async def test_global_offline(event_loop, service: NS):
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
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.network
+@pytest.mark.regular
@pytest.mark.dependency
async def test_technologies_verb(event_loop, service: NS):
msgid = await service.technologies()
@@ -54,6 +68,8 @@ async def test_technologies_verb(event_loop, service: NS):
assert 'values' in resp.data
+@pytest.mark.network
+@pytest.mark.regular
@pytest.mark.dependency(depends=['test_technologies_verb'])
async def test_expected_existing_technologies(event_loop, service: NS, expected_available_techs):
msgid = await service.technologies()
@@ -62,6 +78,9 @@ async def test_expected_existing_technologies(event_loop, service: NS, expected_
for t in expected_available_techs:
assert t in techs, f'"{t}" technology is expected to be available, but it is not'
+
+@pytest.mark.network
+@pytest.mark.regular
@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:
@@ -70,11 +89,13 @@ async def test_get_property(event_loop, service: NS, expected_available_techs):
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()))
+ # diverging_fields = frozenset(expected_fields).symmetric_difference(frozenset(resp.data.keys()))
diverging_fields = frozenset(expected_fields).difference(frozenset(resp.data.keys()))
assert len(diverging_fields) == 0, f'the following property fields are diverging from the expected: {diverging_fields}'
+@pytest.mark.network
+@pytest.mark.regular
@pytest.mark.xfail(reason='Expecting this to throw "permission denied" via the API, tethering from connmanctl succeeds')
async def test_enable_wifi_tethering(event_loop, service: NS, expected_available_techs):
if 'wifi' not in expected_available_techs:
@@ -85,6 +106,9 @@ async def test_enable_wifi_tethering(event_loop, service: NS, expected_available
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+
+@pytest.mark.network
+@pytest.mark.regular
@pytest.mark.dependency(depends='test_enable_wifi_tethering')
async def test_disable_wifi_tethering(event_loop, service: NS, expected_available_techs):
if 'wifi' not in expected_available_techs:
@@ -102,6 +126,8 @@ async def test_disable_wifi_tethering(event_loop, service: NS, expected_availabl
# print(resp)
+@pytest.mark.network
+@pytest.mark.regular
async def test_services_verb(event_loop, service: NS):
msgid = await service.services()
resp = await service.afbresponse()
@@ -109,78 +135,104 @@ async def test_services_verb(event_loop, service: NS):
assert 'values' in resp.data
+@pytest.mark.network
+@pytest.mark.regular
async def test_subscribe_global_state(event_loop, service: NS):
msgid = await service.subscribe('global_state')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_unsubscribe_global_state(event_loop, service: NS):
msgid = await service.unsubscribe('global_state')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_subscribe_technologies(event_loop, service: NS):
msgid = await service.subscribe('technologies')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_unsubscribe_technologies(event_loop, service: NS):
msgid = await service.unsubscribe('technologies')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_subscribe_tech_props(event_loop, service: NS):
msgid = await service.subscribe('technology_properties')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_unsubscribe_tech_props(event_loop, service: NS):
msgid = await service.unsubscribe('technology_properties')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_subscribe_services(event_loop, service: NS):
msgid = await service.subscribe('services')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_unsubscribe_services(event_loop, service: NS):
msgid = await service.unsubscribe('services')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_subscribe_service_props(event_loop, service: NS):
msgid = await service.subscribe('service_properties')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_unsubscribe_service_props(event_loop, service: NS):
msgid = await service.unsubscribe('service_properties')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_subscribe_agent(event_loop, service: NS):
msgid = await service.subscribe('agent')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_unsubscribe_agent(event_loop, service: NS):
msgid = await service.unsubscribe('agent')
resp = await service.afbresponse()
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_enable_wifi(event_loop, service: NS, expected_available_techs):
if 'wifi' not in expected_available_techs:
pytest.skip('Skipping enable_technology for "wifi" because it is not expected to be available')
@@ -189,6 +241,8 @@ async def test_enable_wifi(event_loop, service: NS, expected_available_techs):
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_disable_wifi(event_loop, service: NS, expected_available_techs):
if 'wifi' not in expected_available_techs:
pytest.skip('Skipping disable_technology for "wifi" because it is not expected to be available')
@@ -197,6 +251,8 @@ async def test_disable_wifi(event_loop, service: NS, expected_available_techs):
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_enable_bluetooth(event_loop, service: NS, expected_available_techs):
if 'bluetooth' not in expected_available_techs:
pytest.skip('Skipping enable_technology for "bluetooth" because it is not expected to be available')
@@ -205,6 +261,8 @@ async def test_enable_bluetooth(event_loop, service: NS, expected_available_tech
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_disable_bluetooth(event_loop, service: NS, expected_available_techs):
if 'bluetooth' not in expected_available_techs:
pytest.skip('Skipping disable_technology for "bluetooth" because it is not expected to be available')
@@ -213,6 +271,8 @@ async def test_disable_bluetooth(event_loop, service: NS, expected_available_tec
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_enable_ethernet(event_loop, service: NS, expected_available_techs):
if 'ethernet' not in expected_available_techs:
pytest.skip('Skipping enable_technology for "ethernet" because it is not expected to be available')
@@ -221,6 +281,8 @@ async def test_enable_ethernet(event_loop, service: NS, expected_available_techs
assert resp.status == 'success', resp.info
+@pytest.mark.network
+@pytest.mark.regular
async def test_disable_ethernet(event_loop, service: NS, expected_available_techs):
addr, _ = service.websocket.remote_address
if addr != 'localhost' or addr != '127.0.0.1':
@@ -231,4 +293,4 @@ async def test_disable_ethernet(event_loop, service: NS, expected_available_tech
msgid = await service.disable_technology('ethernet')
resp = await service.afbresponse()
- assert resp.status == 'success', resp.info \ No newline at end of file
+ assert resp.status == 'success', resp.info
diff --git a/pyagl/tests/test_nfc.py b/pyagl/tests/test_nfc.py
index d09f443..b845b8d 100644
--- a/pyagl/tests/test_nfc.py
+++ b/pyagl/tests/test_nfc.py
@@ -24,12 +24,16 @@ async def service():
await svc.websocket.close()
+@pytest.mark.nfc
+@pytest.mark.regular
async def subscribe(event_loop, service: nfcs):
msgid = service.subscribe()
resp = await service.afbresponse()
assert resp.msgid == msgid
+@pytest.mark.nfc
+@pytest.mark.regular
async def unsubscribe(event_loop, service: nfcs):
msgid = service.unsubscribe()
resp = await service.afbresponse()
diff --git a/pyagl/tests/test_weather.py b/pyagl/tests/test_weather.py
index 713491c..1a5e204 100644
--- a/pyagl/tests/test_weather.py
+++ b/pyagl/tests/test_weather.py
@@ -23,6 +23,8 @@ async def service():
await gpss.websocket.close()
+@pytest.mark.weather
+@pytest.mark.regular
async def test_apikey(event_loop, service: ws):
msgid = await service.apikey()
resp = await service.afbresponse()
@@ -30,6 +32,8 @@ async def test_apikey(event_loop, service: ws):
assert resp.data['api_key'] == 'a860fa437924aec3d0360cc749e25f0e'
+@pytest.mark.weather
+@pytest.mark.regular
async def test_current_weather(event_loop, service: ws):
msgid = await service.current_weather()
resp = await service.afbresponse()
@@ -37,17 +41,24 @@ async def test_current_weather(event_loop, service: ws):
assert 'sys' in resp.data
+@pytest.mark.weather
+@pytest.mark.regular
async def test_bad_subscription(event_loop, service: ws):
msgid = await service.subscribe('non-existant')
resp = await service.afbresponse()
assert resp.status == 'failed'
+@pytest.mark.weather
+@pytest.mark.regular
async def test_bad_unsubscription(event_loop, service: ws):
msgid = await service.unsubscribe('non-existant')
resp = await service.afbresponse()
assert resp.status == 'failed'
+
+@pytest.mark.weather
+@pytest.mark.regular
@pytest.mark.dependency
async def test_subscribe_weather(event_loop, service: ws):
event = 'weather'
@@ -58,6 +69,8 @@ async def test_subscribe_weather(event_loop, service: ws):
assert eventresp.api == f'{service.api}/{event}'
+@pytest.mark.weather
+@pytest.mark.regular
@pytest.mark.dependency(depends=['test_subscribe_weather'])
async def test_unsubscribe_weather(event_loop, service: ws):
msgid = await service.subscribe('weather')
diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..217ef9a
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,15 @@
+[pytest]
+norecursedirs = templates
+
+markers =
+ regular: regular verb tests with expected values
+ hwrequired: verb tests requiring available physical hardware
+ audiomixer: agl-service-audiomixer tests
+ bluetooth: agl-service-bluetooth tests
+ btmap: agl-service-bluetooth-map tests
+ btpbap: agl-service-bluetooth-pbap tests
+ geoclue: agl-service-geoclue tests
+ network: agl-service-network tests
+ nfc: agl-service-nfc tests
+ gps: agl-service-gps tests
+ weather: agl-service-weather tests
diff --git a/tox.ini b/tox.ini
new file mode 100644
index 0000000..c55cd13
--- /dev/null
+++ b/tox.ini
@@ -0,0 +1,17 @@
+[tox]
+envlist = py38
+
+[testenv]
+setenv =
+ AGL_TGT_IP = {env:AGL_TGT_IP:"127.0.0.1"}
+
+deps =
+ websockets
+ pytest-dependency
+ pytest
+ parse
+ pytest-asyncio
+ asyncssh
+
+commands =
+ pytest --lava -qk "not hwrequired"