CodeZoo co., ltd / Mbed OS mbed-os-example-pelion-CATM1

Mbed OS example of Pelion device management LGUPlus Client

TESTS/pelion_helper.py@2:34933ca5af82, 2020-02-21 (annotated)

Committer:
pimco01
Date:
Fri Feb 21 19:27:19 2020 +0000
Revision:
2:34933ca5af82
Parent:
0:9f917a7bf2da 
main.cpp, app.json modify

Who changed what in which revision?

UserRevisionLine numberNew contents of line
MACRUM 0:9f917a7bf2da 1 """
MACRUM 0:9f917a7bf2da 2 Copyright 2019 ARM Limited
MACRUM 0:9f917a7bf2da 3 Licensed under the Apache License, Version 2.0 (the "License");
MACRUM 0:9f917a7bf2da 4 you may not use this file except in compliance with the License.
MACRUM 0:9f917a7bf2da 5 You may obtain a copy of the License at
MACRUM 0:9f917a7bf2da 6
MACRUM 0:9f917a7bf2da 7 http://www.apache.org/licenses/LICENSE-2.0
MACRUM 0:9f917a7bf2da 8
MACRUM 0:9f917a7bf2da 9 Unless required by applicable law or agreed to in writing, software
MACRUM 0:9f917a7bf2da 10 distributed under the License is distributed on an "AS IS" BASIS,
MACRUM 0:9f917a7bf2da 11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
MACRUM 0:9f917a7bf2da 12 See the License for the specific language governing permissions and
MACRUM 0:9f917a7bf2da 13 limitations under the License.
MACRUM 0:9f917a7bf2da 14 """
MACRUM 0:9f917a7bf2da 15
MACRUM 0:9f917a7bf2da 16 # pylint: disable=missing-docstring,too-many-instance-attributes
MACRUM 0:9f917a7bf2da 17 # pylint: disable=line-too-long,method-hidden
MACRUM 0:9f917a7bf2da 18
MACRUM 0:9f917a7bf2da 19 import json
MACRUM 0:9f917a7bf2da 20 import os
MACRUM 0:9f917a7bf2da 21 import uuid
MACRUM 0:9f917a7bf2da 22 import requests
MACRUM 0:9f917a7bf2da 23 from mbed_cloud import AccountManagementAPI
MACRUM 0:9f917a7bf2da 24 from mbed_cloud import ConnectAPI
MACRUM 0:9f917a7bf2da 25 from mbed_cloud import DeviceDirectoryAPI
MACRUM 0:9f917a7bf2da 26 from mbed_cloud import UpdateAPI
MACRUM 0:9f917a7bf2da 27 from icetea_lib.bench import Bench
MACRUM 0:9f917a7bf2da 28 from icetea_lib.bench import TestStepFail
MACRUM 0:9f917a7bf2da 29
MACRUM 0:9f917a7bf2da 30
MACRUM 0:9f917a7bf2da 31 class PelionBase(Bench):
MACRUM 0:9f917a7bf2da 32 """
MACRUM 0:9f917a7bf2da 33 Base class containing common implementation shared by tests
MACRUM 0:9f917a7bf2da 34 """
MACRUM 0:9f917a7bf2da 35
MACRUM 0:9f917a7bf2da 36 def __init__(self, **kwargs):
MACRUM 0:9f917a7bf2da 37 Bench.__init__(self, **kwargs)
MACRUM 0:9f917a7bf2da 38 self.test_config = None
MACRUM 0:9f917a7bf2da 39 self.device_id = None
MACRUM 0:9f917a7bf2da 40 self.manifest_id = None
MACRUM 0:9f917a7bf2da 41 self.account_api = None
MACRUM 0:9f917a7bf2da 42 self.connect_api = None
MACRUM 0:9f917a7bf2da 43 self.device_api = None
MACRUM 0:9f917a7bf2da 44 self.update_api = None
MACRUM 0:9f917a7bf2da 45 self.rest_headers = None
MACRUM 0:9f917a7bf2da 46 self.rest_address = None
MACRUM 0:9f917a7bf2da 47 self.pattern_value1 = "1000:1000:1000:1000"
MACRUM 0:9f917a7bf2da 48 self.pattern_value2 = "2000:1000:2000:1000"
MACRUM 0:9f917a7bf2da 49 self.pattern_value3 = "3000:1000:3000:1000"
MACRUM 0:9f917a7bf2da 50 self.notified_value = ""
MACRUM 0:9f917a7bf2da 51
MACRUM 0:9f917a7bf2da 52 def setup(self):
MACRUM 0:9f917a7bf2da 53 # Check if API key is in environmental vars
MACRUM 0:9f917a7bf2da 54 if 'MBED_CLOUD_SDK_API_KEY' in os.environ:
MACRUM 0:9f917a7bf2da 55 api_key = (os.environ[str('MBED_CLOUD_SDK_API_KEY')])
MACRUM 0:9f917a7bf2da 56 else:
MACRUM 0:9f917a7bf2da 57 api_key = self.config.get("api_key")
MACRUM 0:9f917a7bf2da 58
MACRUM 0:9f917a7bf2da 59 if not api_key.startswith('ak_'):
MACRUM 0:9f917a7bf2da 60 raise TestStepFail("No API key in MBED_CLOUD_SDK_API_KEY or in pelion.tc_cfg")
MACRUM 0:9f917a7bf2da 61
MACRUM 0:9f917a7bf2da 62 self.device_id = self.config.get("device_id")
MACRUM 0:9f917a7bf2da 63 host = self.config.get("host")
MACRUM 0:9f917a7bf2da 64 self.test_config = {"api_key": api_key, "host": host}
MACRUM 0:9f917a7bf2da 65 self.account_api = AccountManagementAPI(self.test_config)
MACRUM 0:9f917a7bf2da 66 self.connect_api = ConnectAPI(self.test_config)
MACRUM 0:9f917a7bf2da 67 self.device_api = DeviceDirectoryAPI(self.test_config)
MACRUM 0:9f917a7bf2da 68
MACRUM 0:9f917a7bf2da 69 # Additional parameters for handling REST requests without SDK
MACRUM 0:9f917a7bf2da 70 self.rest_headers = {'Authorization': 'Bearer ' + api_key}
MACRUM 0:9f917a7bf2da 71 self.rest_address = self.config.get("host")
MACRUM 0:9f917a7bf2da 72
MACRUM 0:9f917a7bf2da 73 # Init delay due to internal usage of PULL notification in SDK. Notications might be lost between
MACRUM 0:9f917a7bf2da 74 # tests.
MACRUM 0:9f917a7bf2da 75 # TODO: Remove workaround after limitation in SDK has been fixed.
MACRUM 0:9f917a7bf2da 76 self.delay(5)
MACRUM 0:9f917a7bf2da 77
MACRUM 0:9f917a7bf2da 78 def setup_update(self):
MACRUM 0:9f917a7bf2da 79 self.manifest_id = self.config.get("manifest_id")
MACRUM 0:9f917a7bf2da 80 self.update_api = UpdateAPI(self.test_config)
MACRUM 0:9f917a7bf2da 81
MACRUM 0:9f917a7bf2da 82 def _callback_fn(self, device_id, path, value):
MACRUM 0:9f917a7bf2da 83 string = value.decode("utf-8")
MACRUM 0:9f917a7bf2da 84 self.logger.info("Notification for %s received: %r value: %r", device_id, path, string)
MACRUM 0:9f917a7bf2da 85 self.notified_value = string
MACRUM 0:9f917a7bf2da 86
MACRUM 0:9f917a7bf2da 87 def verify_registration(self, expected_state):
MACRUM 0:9f917a7bf2da 88 self.logger.info("Verify device to in state %s", expected_state)
MACRUM 0:9f917a7bf2da 89 device = self.device_api.get_device(self.device_id)
MACRUM 0:9f917a7bf2da 90 if device is None:
MACRUM 0:9f917a7bf2da 91 raise TestStepFail("device_id %s does not exist/is not listed in Device Directory" % self.device_id)
MACRUM 0:9f917a7bf2da 92 else:
MACRUM 0:9f917a7bf2da 93 if device.state == expected_state:
MACRUM 0:9f917a7bf2da 94 self.logger.info("Endpoint %s is in state: %s", self.device_id, expected_state)
MACRUM 0:9f917a7bf2da 95 else:
MACRUM 0:9f917a7bf2da 96 raise TestStepFail("Endpoint %s is in state %s, expected %s" % (self.device_id, device.state, expected_state))
MACRUM 0:9f917a7bf2da 97
MACRUM 0:9f917a7bf2da 98 def prepare_campaign_filter(self, state):
MACRUM 0:9f917a7bf2da 99 # Create filter for the update campaign. Here we only update one endpoint.
MACRUM 0:9f917a7bf2da 100 payload = {}
MACRUM 0:9f917a7bf2da 101 payload["device_filter"] = "id=" + self.device_id
MACRUM 0:9f917a7bf2da 102 payload["root_manifest_id"] = self.manifest_id
MACRUM 0:9f917a7bf2da 103 payload["name"] = str(uuid.uuid4())
MACRUM 0:9f917a7bf2da 104 payload["state"] = state
MACRUM 0:9f917a7bf2da 105 return payload
MACRUM 0:9f917a7bf2da 106
MACRUM 0:9f917a7bf2da 107 ## Function for starting a update campaign
MACRUM 0:9f917a7bf2da 108 ## TODO: Replace with Python SDK implementation
MACRUM 0:9f917a7bf2da 109 def post_campaign(self, payload):
MACRUM 0:9f917a7bf2da 110 req_address = self.rest_address + '/v3/update-campaigns'
MACRUM 0:9f917a7bf2da 111 res = requests.post(req_address, json=payload, headers=self.rest_headers)
MACRUM 0:9f917a7bf2da 112 if res.status_code == 409:
MACRUM 0:9f917a7bf2da 113 # Endpoint can be targeted by only one active campaign. The older campaign must be deleted/stopped.
MACRUM 0:9f917a7bf2da 114 raise TestStepFail("Campaign already exists for device %s" % self.device_id)
MACRUM 0:9f917a7bf2da 115 elif res.status_code != 201:
MACRUM 0:9f917a7bf2da 116 raise TestStepFail("Campaign creation failed with %d" % res.status_code)
MACRUM 0:9f917a7bf2da 117
MACRUM 0:9f917a7bf2da 118 data = res.json()
MACRUM 0:9f917a7bf2da 119 campaign_id = data["id"]
MACRUM 0:9f917a7bf2da 120 return campaign_id
MACRUM 0:9f917a7bf2da 121
MACRUM 0:9f917a7bf2da 122 ## Function for checking firmware update status from Cloud
MACRUM 0:9f917a7bf2da 123 ## TODO: Replace with Python SDK implementation
MACRUM 0:9f917a7bf2da 124 def check_campaign_state(self, campaign_id):
MACRUM 0:9f917a7bf2da 125 results = []
MACRUM 0:9f917a7bf2da 126 base_url = self.rest_address + "/v3/update-campaigns"
MACRUM 0:9f917a7bf2da 127 campaign_url = base_url + "/" + campaign_id
MACRUM 0:9f917a7bf2da 128 metadata_url = campaign_url + "/" + "campaign-device-metadata"
MACRUM 0:9f917a7bf2da 129
MACRUM 0:9f917a7bf2da 130 # Short wait until campaign has been initialized properly.
MACRUM 0:9f917a7bf2da 131 self.delay(5)
MACRUM 0:9f917a7bf2da 132
MACRUM 0:9f917a7bf2da 133 # Fetch basic campaign information
MACRUM 0:9f917a7bf2da 134 response = requests.get(campaign_url, headers=self.rest_headers)
MACRUM 0:9f917a7bf2da 135 if response.status_code != 200:
MACRUM 0:9f917a7bf2da 136 raise TestStepFail("Get Campaign state returned %d" % response.status_code)
MACRUM 0:9f917a7bf2da 137
MACRUM 0:9f917a7bf2da 138 resp = json.loads(response.content)
MACRUM 0:9f917a7bf2da 139 results.append(resp["state"])
MACRUM 0:9f917a7bf2da 140
MACRUM 0:9f917a7bf2da 141 # Fetch campaign metadata information
MACRUM 0:9f917a7bf2da 142 response = requests.get(metadata_url, headers=self.rest_headers)
MACRUM 0:9f917a7bf2da 143 if response.status_code != 200:
MACRUM 0:9f917a7bf2da 144 raise TestStepFail("Get Campaign metadata returned %d" % response.status_code)
MACRUM 0:9f917a7bf2da 145
MACRUM 0:9f917a7bf2da 146 resp = json.loads(response.content)
MACRUM 0:9f917a7bf2da 147 if 'data' in resp and resp['data'][0]['deployment_state']:
MACRUM 0:9f917a7bf2da 148 meta_data = resp['data']
MACRUM 0:9f917a7bf2da 149 device_state = meta_data[0]['deployment_state']
MACRUM 0:9f917a7bf2da 150 results.append(device_state)
MACRUM 0:9f917a7bf2da 151 else:
MACRUM 0:9f917a7bf2da 152 raise TestStepFail("No metadata for %d" % self.device_id)
MACRUM 0:9f917a7bf2da 153
MACRUM 0:9f917a7bf2da 154 return results
MACRUM 0:9f917a7bf2da 155
MACRUM 0:9f917a7bf2da 156 restTimeout = 60