summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--pyagl/services/network.py7
-rw-r--r--pyagl/tests/test_network.py137
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
+