Add wifi testcases to pyagl.tests.test_network.pylamprey_12.1.0lamprey/12.1.012.1.0

Adding testcases for wifi
2 second sleep was added in test_enable_wifi to let wifi interfaces like
wlcore to have time to initialize, otherwise causes errors during scan
Having AGL_TEST_WIFI_SSID is optional and if unset tests should be skipped

Bug-AGL：SPEC-4069

Signed-off-by: duerpei <duep.fnst@fujitsu.com>
Signed-off-by: Edi Feschiyan <edi.feschiyan@konsulko.com>
Change-Id: I4d2a5993de30058a4022d6e2f8f408a636e74e74

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
+