pelion_helper.py

00001 """
00002 Copyright 2019 ARM Limited
00003 Licensed under the Apache License, Version 2.0 (the "License");
00004 you may not use this file except in compliance with the License.
00005 You may obtain a copy of the License at
00006 
00007     http://www.apache.org/licenses/LICENSE-2.0
00008 
00009 Unless required by applicable law or agreed to in writing, software
00010 distributed under the License is distributed on an "AS IS" BASIS,
00011 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
00012 See the License for the specific language governing permissions and
00013 limitations under the License.
00014 """
00015 
00016 # pylint: disable=missing-docstring,too-many-instance-attributes
00017 # pylint: disable=line-too-long,method-hidden
00018 
00019 import json
00020 import os
00021 import uuid
00022 import requests
00023 from mbed_cloud import AccountManagementAPI
00024 from mbed_cloud import ConnectAPI
00025 from mbed_cloud import DeviceDirectoryAPI
00026 from mbed_cloud import UpdateAPI
00027 from icetea_lib.bench import Bench
00028 from icetea_lib.bench import TestStepFail
00029 
00030 
00031 class PelionBase (Bench):
00032     """
00033     Base class containing common implementation shared by tests
00034     """
00035 
00036     def __init__(self, **kwargs):
00037         Bench.__init__(self, **kwargs)
00038         self.test_config  = None
00039         self.device_id  = None
00040         self.manifest_id  = None
00041         self.account_api  = None
00042         self.connect_api  = None
00043         self.device_api  = None
00044         self.update_api  = None
00045         self.rest_headers  = None
00046         self.rest_address  = None
00047         self.pattern_value1  = "1000:1000:1000:1000"
00048         self.pattern_value2  = "2000:1000:2000:1000"
00049         self.pattern_value3  = "3000:1000:3000:1000"
00050         self.notified_value  = ""
00051 
00052     def setup(self):
00053         # Check if API key is in environmental vars
00054         if 'MBED_CLOUD_SDK_API_KEY' in os.environ:
00055             api_key = (os.environ[str('MBED_CLOUD_SDK_API_KEY')])
00056         else:
00057             api_key = self.config.get("api_key")
00058 
00059         if not api_key.startswith('ak_'):
00060             raise TestStepFail("No API key in MBED_CLOUD_SDK_API_KEY or in pelion.tc_cfg")
00061 
00062         self.device_id  = self.config.get("device_id")
00063         host = self.config.get("host")
00064         self.test_config  = {"api_key": api_key, "host": host}
00065         self.account_api  = AccountManagementAPI(self.test_config )
00066         self.connect_api  = ConnectAPI(self.test_config )
00067         self.device_api  = DeviceDirectoryAPI(self.test_config )
00068 
00069         # Additional parameters for handling REST requests without SDK
00070         self.rest_headers  = {'Authorization': 'Bearer ' + api_key}
00071         self.rest_address  = self.config.get("host")
00072 
00073         # Init delay due to internal usage of PULL notification in SDK. Notications might be lost between
00074         # tests.
00075         # TODO: Remove workaround after limitation in SDK has been fixed.
00076         self.delay(5)
00077 
00078     def setup_update(self):
00079         self.manifest_id  = self.config.get("manifest_id")
00080         self.update_api  = UpdateAPI(self.test_config )
00081 
00082     def _callback_fn(self, device_id, path, value):
00083         string = value.decode("utf-8")
00084         self.logger.info("Notification for %s received: %r value: %r", device_id, path, string)
00085         self.notified_value  = string
00086 
00087     def verify_registration(self, expected_state):
00088         self.logger.info("Verify device to in state %s", expected_state)
00089         device = self.device_api .get_device(self.device_id )
00090         if device is None:
00091             raise TestStepFail("device_id %s does not exist/is not listed in Device Directory" % self.device_id )
00092         else:
00093             if device.state == expected_state:
00094                 self.logger.info("Endpoint %s is in state: %s", self.device_id , expected_state)
00095             else:
00096                 raise TestStepFail("Endpoint %s is in state %s, expected %s" % (self.device_id , device.state, expected_state))
00097 
00098     def prepare_campaign_filter(self, state):
00099         # Create filter for the update campaign. Here we only update one endpoint.
00100         payload = {}
00101         payload["device_filter"] = "id=" + self.device_id 
00102         payload["root_manifest_id"] = self.manifest_id 
00103         payload["name"] = str(uuid.uuid4())
00104         payload["state"] = state
00105         return payload
00106 
00107     ## Function for starting a update campaign
00108     ## TODO: Replace with Python SDK implementation
00109     def post_campaign(self, payload):
00110         req_address = self.rest_address  + '/v3/update-campaigns'
00111         res = requests.post(req_address, json=payload, headers=self.rest_headers )
00112         if res.status_code == 409:
00113             # Endpoint can be targeted by only one active campaign. The older campaign must be deleted/stopped.
00114             raise TestStepFail("Campaign already exists for device %s" % self.device_id )
00115         elif res.status_code != 201:
00116             raise TestStepFail("Campaign creation failed with %d" % res.status_code)
00117 
00118         data = res.json()
00119         campaign_id = data["id"]
00120         return campaign_id
00121 
00122     ## Function for checking firmware update status from Cloud
00123     ## TODO: Replace with Python SDK implementation
00124     def check_campaign_state(self, campaign_id):
00125         results = []
00126         base_url = self.rest_address  + "/v3/update-campaigns"
00127         campaign_url = base_url + "/" + campaign_id
00128         metadata_url = campaign_url + "/" + "campaign-device-metadata"
00129 
00130         # Short wait until campaign has been initialized properly.
00131         self.delay(5)
00132 
00133         # Fetch basic campaign information
00134         response = requests.get(campaign_url, headers=self.rest_headers )
00135         if response.status_code != 200:
00136             raise TestStepFail("Get Campaign state returned %d" % response.status_code)
00137 
00138         resp = json.loads(response.content)
00139         results.append(resp["state"])
00140 
00141         # Fetch campaign metadata information
00142         response = requests.get(metadata_url, headers=self.rest_headers )
00143         if response.status_code != 200:
00144             raise TestStepFail("Get Campaign metadata returned %d" % response.status_code)
00145 
00146         resp = json.loads(response.content)
00147         if 'data' in resp and resp['data'][0]['deployment_state']:
00148             meta_data = resp['data']
00149             device_state = meta_data[0]['deployment_state']
00150             results.append(device_state)
00151         else:
00152             raise TestStepFail("No metadata for %d" % self.device_id )
00153 
00154         return results
00155 
00156     restTimeout = 60