Merge pull request #20 from ARMmbed/only_one_apikey

Option to run testset without creating temp apikey for notification channel

README.md

@@ -23,6 +23,7 @@ $ pip install -I pelion_test_lib*.whl
 - The [API key](https://www.pelion.com/docs/device-management/current/integrate-web-app/api-keys.html) is used from Mbed's default `CLOUD_SDK_API_KEY` environment variable, but you can override it by defining a separate variable using a command-line command:
     - Linux: `export PELION_CLOUD_API_KEY=[api_key_here]`
     - Windows: `set PELION_CLOUD_API_KEY=[api_key_here]`
+- Test run will create temporary API key for the WebSocket callback channel by default. If you want to prevent that and use only the exported API key, add `--use_one_apikey` startup argument.
 - Tests use [Mbed LS](https://github.com/ARMmbed/mbed-os-tools/tree/master/packages/mbed-ls) to select the board from the serial port.
   - If you have only one board connected to the serial port, you don't need to select the device for the tests.
   - If there are multiple boards connected to the serial port, run `mbedls` to check the target board's ID, and use it in the test run's argument `--target_id=[id]`.

changelog.md

@@ -1,8 +1,10 @@
 # Pelion E2E Python Test Library Changelog
 
 
-## 0.2.5  2020-03-xx
+## 0.2.5  2020-03-20
 - Added fixture for multi device update campaign.
+- Option to run test set without creating temporary API key.
+- Breaking change: API key fixture name change and it returns only the key part.
 
 ## 0.2.4  2020-02-27
 - Completed the Python 2 support.

pelion_test_lib/fixtures/cloud_fixtures.py

@@ -50,49 +50,50 @@ def cloud():
 
     cloud_api = PelionCloud(api_gw, api_key)
 
-    payload = {'name': 'pelion_e2e_dynamic_api_key'}
-    r = cloud_api.account.create_api_key(payload, expected_status_code=201)
-    resp = r.json()
-    cloud_api.rest_api.set_default_api_key(resp['key'])
-
     yield cloud_api
 
-    log.debug('Cleaning out the Cloud API fixture')
-    headers = {'Authorization': 'Bearer {}'.format(api_key)}
-    cloud_api.account.delete_api_key(resp['id'], headers=headers, expected_status_code=204)
-
 
 @pytest.fixture(scope='module')
-def temp_api_key(cloud):
+def api_key(cloud, request):
     """
-    Create new temporary function level developer api key
-    :param cloud: Cloud API fixture
-    :return: Rest API response for new developer api key
+    Create new temporary module level developer API key
+    When running testset with 'use_one_apikey' argument this doesn't create new one but returns current API key!
+    :param cloud: Cloud fixture
+    :return: API key
     """
-    payload = {'name': 'pelion_e2e_dynamic_api_key'}
-    r = cloud.account.create_api_key(payload, expected_status_code=201)
-    resp = r.json()
+    use_current = request.config.getoption('use_one_apikey', False)
+    temp_key_id = None
 
-    log.info('Created new developer api key for test case, id: {}'.format(resp['id']))
+    payload = {'name': 'pelion_e2e_dynamic_api_key'}
+    if use_current:
+        log.info('Using current API key, not creating temporary one')
+        key = cloud.rest_api.api_key
+    else:
+        log.info('Creating new developer API key')
+        r = cloud.account.create_api_key(payload, expected_status_code=201)
+        resp = r.json()
+        key = resp['key']
+        temp_key_id = resp['id']
+        log.info('Created new developer API key for the test run, ID: {}'.format(temp_key_id))
 
-    yield resp
+    yield key
 
-    log.info('Cleaning out the generated test case developer api key, id: {}'.format(resp['id']))
-    cloud.account.delete_api_key(resp['id'], expected_status_code=204)
+    if temp_key_id:
+        log.info('Cleaning out the generated test set developer API key, ID: {}'.format(temp_key_id))
+        cloud.account.delete_api_key(temp_key_id, expected_status_code=204)
 
 
 @pytest.fixture(scope='module')
-def websocket(cloud, temp_api_key):
+def websocket(cloud, api_key):
     log.info('Register and open WebSocket notification channel')
-    headers = {'Authorization': 'Bearer {}'.format(temp_api_key['key'])}
+    headers = {'Authorization': 'Bearer {}'.format(api_key)}
     cloud.connect.register_websocket_channel(headers=headers, expected_status_code=[200, 201])
     sleep(5)
     # Get host part from api address
     host = cloud.api_gw.split('//')[1]
 
     log.info('Opening WebSocket handler')
-    ws = websocket_handler.WebSocketRunner('wss://{}/v2/notification/websocket-connect'.format(host),
-                                           temp_api_key['key'])
+    ws = websocket_handler.WebSocketRunner('wss://{}/v2/notification/websocket-connect'.format(host), api_key)
     handler = websocket_handler.WebSocketHandler(ws)
     yield handler
 

tests/conftest.py

@@ -37,6 +37,7 @@ def pytest_addoption(parser):
     parser.addoption('--delta_manifest', action='store', default=False,
                      help='set true if given update_bin is a delta image')
     parser.addoption('--local_binary', action='store', help='local linux client binary path')
+    parser.addoption('--use_one_apikey', action='store_true', default=False, help='do not create temp api key')
 
 
 def pytest_report_teststatus(report):

tests/dev-client-tests.py

@@ -32,26 +32,26 @@ def test_01_get_device_id(cloud, client):
     cloud.device_directory.get_device(client.endpoint_id(), expected_status_code=200)
 
 
-def test_02_put_resource(cloud, client, websocket, temp_api_key):
-    headers = {'Authorization': 'Bearer {}'.format(temp_api_key['key'])}
+def test_02_put_resource(cloud, client, websocket, api_key):
+    headers = {'Authorization': 'Bearer {}'.format(api_key)}
     async_id = put_async_device_request(cloud, RESOURCE_PATH, client.endpoint_id(), PATTERN, headers)
     log.info('Put value "{}" to resource "{}". Async-id: "{}"'.format(PATTERN, RESOURCE_PATH, async_id))
 
     async_response = websocket.wait_for_async_response(async_response_id=async_id, timeout=180, assert_errors=True)
     assert async_response['decoded_payload'] == b'', 'Invalid payload received from the device!'
 
 
-def test_03_get_resource(cloud, client, websocket, temp_api_key):
-    headers = {'Authorization': 'Bearer {}'.format(temp_api_key['key'])}
+def test_03_get_resource(cloud, client, websocket, api_key):
+    headers = {'Authorization': 'Bearer {}'.format(api_key)}
     async_id = get_async_device_request(cloud, RESOURCE_PATH, client.endpoint_id(), headers)
     log.info('Get value from resource "{}". Async-id: "{}"'.format(RESOURCE_PATH, async_id))
 
     async_response = websocket.wait_for_async_response(async_response_id=async_id, timeout=180, assert_errors=True)
     assert async_response['decoded_payload'] == PATTERN.encode('ascii'), 'Invalid payload received from the device!'
 
 
-def test_04_subscribe_resource(cloud, client, websocket, temp_api_key):
-    headers = {'Authorization': 'Bearer {}'.format(temp_api_key['key'])}
+def test_04_subscribe_resource(cloud, client, websocket, api_key):
+    headers = {'Authorization': 'Bearer {}'.format(api_key)}
     r = cloud.connect.set_subscription_for_resource(client.endpoint_id(), '3201/0/5853', headers,
                                                     expected_status_code=[200, 202])
     if r.status_code == 202:
@@ -67,8 +67,8 @@ def test_04_subscribe_resource(cloud, client, websocket, temp_api_key):
     websocket.wait_for_notification(client.endpoint_id(), RESOURCE_PATH, PATTERN2, timeout=180, assert_errors=True)
 
 
-def test_05_factory_reset(cloud, client, websocket, temp_api_key):
-    headers = {'Authorization': 'Bearer {}'.format(temp_api_key['key'])}
+def test_05_factory_reset(cloud, client, websocket, api_key):
+    headers = {'Authorization': 'Bearer {}'.format(api_key)}
     endpoint_id = client.endpoint_id()
 
     # store bootstrap time and execution mode