diff options
| -rw-r--r-- | pyagl/services/network.py | 7 | ||||
| -rw-r--r-- | pyagl/tests/test_network.py | 137 |
2 files changed, 75 insertions, 69 deletions
diff --git a/pyagl/services/network.py b/pyagl/services/network.py index 09867c2..8cdb88f 100644 --- a/pyagl/services/network.py +++ b/pyagl/services/network.py @@ -25,6 +25,7 @@ class NetworkService(AGLBaseService): parser.add_argument('--state', action='store_true') parser.add_argument('--services', action='store_true') + parser.add_argument('--scan') def __init__(self, ip, port=None, service='agl-service-network'): super().__init__(api='network-manager', ip=ip, port=port, service=service, timeout = 20) @@ -84,6 +85,12 @@ async def main(loop): resp = await nets.afbresponse() print(resp.data) + if args.scan: + msgid = await nets.scan_services(args.scan) + print(f'Sent scan request with messageid {msgid}') + resp = await nets.afbresponse() + print(resp) + if args.listener: for response in nets.listener(): print(response) diff --git a/pyagl/tests/test_network.py b/pyagl/tests/test_network.py index 931fddd..62c2a2f 100644 --- a/pyagl/tests/test_network.py +++ b/pyagl/tests/test_network.py @@ -46,35 +46,23 @@ def expected_available_interfaces(): @pytest.fixture(scope='module') def scannable_interfaces(): - scannable = os.environ.get('AGL_SCANNABLE_INTERFACES', 'wifi,bluetooth').split(',') + # even though bluetooth supports scan, agl-service-network is effectively an interface + # to connman and connman is currently unable to scan bluetooth, so leaving only wifi + scannable = os.environ.get('AGL_SCANNABLE_INTERFACES', ['wifi']) + return scannable - -@pytest.fixture() -async def get_wifi_name(event_loop, service: NS): - #get a name_string of a wifi service +@pytest.fixture(scope='module') +def get_wifi_name(): + # to get the name of a network use 'connmanctl', 'scan wifi' and use + # the connection string provided by 'services' which looks for example + # should look like wifi_dca6323a2b94_4368696e614e6574_managed_none wifi_name = os.environ.get('AGL_TEST_WIFI_SSID') - if not wifi_name: - pytest.xfail('Please export AGL_TEST_WIFI_SSID with valid wifi name to run the test.' - 'The value of it might look like: wifi_dca6323a2b94_4368696e614e6574_managed_xxx') - #Check if the wifi is in the current list - msgid = await service.services() - resp = await service.afbresponse() - data = resp.data - t_values = data['values'] - for ser in t_values: - service = ser['service'] - if service == wifi_name: - return wifi_name - - pytest.skip('the service is not in the list') - + return wifi_name 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' - async def test_global_offline(event_loop, service: NS): addr, _ = service.websocket.remote_address @@ -85,19 +73,15 @@ async def test_global_offline(event_loop, service: NS): resp = await service.afbresponse() assert resp.status == 'success', resp.info -async def test_disable_wifi(event_loop, service: NS, expected_available_interfaces): - if 'wifi' not in expected_available_interfaces: - pytest.skip('Skipping disable_technology for "wifi" because it is not expected to be available') - msgid = await service.disable_technology('wifi') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - - +@pytest.mark.dependency async def test_enable_wifi(event_loop, service: NS, expected_available_interfaces): if 'wifi' not in expected_available_interfaces: pytest.skip('Skipping enable_technology for "wifi" because it is not expected to be available') msgid = await service.enable_technology('wifi') resp = await service.afbresponse() + # adding a sleep so if an interface needs to load a firmware blob - e.g. wlcore - also needs time + # to initialize the interface, otherwise scan might throw errors + await asyncio.sleep(2) assert resp.status == 'success', resp.info @@ -109,14 +93,6 @@ async def test_enable_bluetooth(event_loop, service: NS, expected_available_inte assert resp.status == 'success', resp.info -async def test_disable_bluetooth(event_loop, service: NS, expected_available_interfaces): - if 'bluetooth' not in expected_available_interfaces: - pytest.skip('Skipping disable_technology for "bluetooth" because it is not expected to be available') - msgid = await service.disable_technology('bluetooth') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - - async def test_enable_ethernet(event_loop, service: NS, expected_available_interfaces): if 'ethernet' not in expected_available_interfaces: pytest.skip('Skipping enable_technology for "ethernet" because it is not expected to be available') @@ -125,24 +101,12 @@ async def test_enable_ethernet(event_loop, service: NS, expected_available_inter assert resp.status == 'success', resp.info -async def test_disable_ethernet(event_loop, service: NS, expected_available_interfaces): - addr, _ = service.websocket.remote_address - if addr != 'localhost' or addr != '127.0.0.1': - pytest.skip('Skipping this test until it is ran locally on the board, ' - 'presuming it is the only available interface to connect to remotely for testing') - if 'ethernet' not in expected_available_interfaces: - pytest.skip('Skipping disable_technology for "ethernet" because it is not expected to be available') - - msgid = await service.disable_technology('ethernet') - resp = await service.afbresponse() - assert resp.status == 'success', resp.info - @pytest.mark.dependency(depends=['test_enable_wifi']) async def test_scan_services(event_loop, service: NS, expected_available_interfaces, scannable_interfaces): for t in scannable_interfaces: if t in expected_available_interfaces: - msgid = await service.scan_services(technology=t) + msgid = await service.scan_services(t) resp = await service.afbresponse() assert resp.status == 'success', f'scan_services failed for technology {t} - {resp.info}' @@ -204,7 +168,6 @@ async def test_disable_wifi_tethering(event_loop, service: NS, expected_availabl 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}) @@ -279,7 +242,6 @@ async def test_unsubscribe_service_props(event_loop, service: NS): resp = await service.afbresponse() assert resp.status == 'success', resp.info - async def test_subscribe_agent(event_loop, service: NS): msgid = await service.subscribe('agent') resp = await service.afbresponse() @@ -292,30 +254,67 @@ async def test_unsubscribe_agent(event_loop, service: NS): assert resp.status == 'success', resp.info -async def test_scan_wifi(event_loop, service: NS): - msgid = await service.scan_services('wifi') - resp = await service.afbresponse() - if 'No carrier' in resp.info: - pytest.xfail('scan wifi services not success,you can try it again') - assert resp.status == 'success' - - +# this test connects to an unencrypted network. if an encrypted network is used +# additional implementation would be needed to enable the authentication agent +# provided by connmanctl and provide secrets to it so it is able to connect +# to the encrypted network +@pytest.mark.dependency(depends=['test_enable_wifi', 'test_scan_services']) +@pytest.mark.dependency async def test_connect_service_wifi(event_loop, service: NS, get_wifi_name): - wifi_name = get_wifi_name - msgid = await service.connect_service(wifi_name) + if not get_wifi_name: + pytest.skip('AGL_TEST_WIFI_SSID not set in os.env - skipping') + msgid = await service.services() resp = await service.afbresponse() - assert resp.status == 'success' or resp.info == 'Connect error: GDBus.Error:net.connman.Error.AlreadyConnected: Already connected' + services = [s['service'] for s in resp.data['values']] + assert get_wifi_name in services, f'Service string {get_wifi_name} was not found in the current services list' + msgid = await service.connect_service(get_wifi_name) + resp = await service.afbresponse() + if 'net.connman.Error.AlreadyConnected' in resp.info: + pytest.skip('Already connected to a network') + assert resp.status == 'success', resp.info +@pytest.mark.dependency(depends=['test_connect_service_wifi']) async def test_disconnect_service_wifi(event_loop, service: NS, get_wifi_name): - wifi_name = get_wifi_name - msgid = await service.disconnect_service(wifi_name) + msgid = await service.disconnect_service(get_wifi_name) resp = await service.afbresponse() - assert resp.status == 'success' or resp.info == 'Disconnect error Not connected' + if 'Not connected' in resp.info: + pytest.skip('Not currently connected') + assert resp.status == 'success' +@pytest.mark.dependency(depends=['test_disconnect_service_wifi']) async def test_remove_service_wifi(event_loop, service: NS, get_wifi_name): - wifi_name = get_wifi_name - msgid = await service.remove_service(wifi_name) + msgid = await service.remove_service(get_wifi_name) resp = await service.afbresponse() assert resp.status == 'success' + + +async def test_disable_wifi(event_loop, service: NS, expected_available_interfaces): + if 'wifi' not in expected_available_interfaces: + pytest.skip('Skipping disable_technology for "wifi" because it is not expected to be available') + msgid = await service.disable_technology('wifi') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +async def test_disable_bluetooth(event_loop, service: NS, expected_available_interfaces): + if 'bluetooth' not in expected_available_interfaces: + pytest.skip('Skipping disable_technology for "bluetooth" because it is not expected to be available') + msgid = await service.disable_technology('bluetooth') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + + +async def test_disable_ethernet(event_loop, service: NS, expected_available_interfaces): + addr, _ = service.websocket.remote_address + if addr != 'localhost' or addr != '127.0.0.1': + pytest.skip('Skipping this test until it is ran locally on the board, ' + 'presuming it is the only available interface to connect to remotely for testing') + if 'ethernet' not in expected_available_interfaces: + pytest.skip('Skipping disable_technology for "ethernet" because it is not expected to be available') + + msgid = await service.disable_technology('ethernet') + resp = await service.afbresponse() + assert resp.status == 'success', resp.info + |