-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[OIS] Add OTA requestor integration test and CI
Add OTA requestor integration test implementation. Add OTA requestor test options in run script. Extending OIS workflow with build the OTA requestor example, build the OTA provider (Linux), test OTA requestor example. Signed-off-by: ATmobica <artur.tynecki@arm.com>
- Loading branch information
ATmobica
committed
Jul 3, 2023
1 parent
89ea136
commit d48bcf2
Showing
4 changed files
with
232 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
190 changes: 190 additions & 0 deletions
190
src/test_driver/openiotsdk/integration-tests/ota-requestor-app/test_app.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,190 @@ | ||
| # | ||
| # Copyright (c) 2023 Project CHIP Authors | ||
| # All rights reserved. | ||
| # | ||
| # Licensed under the Apache License, Version 2.0 (the "License"); | ||
| # you may not use this file except in compliance with the License. | ||
| # You may obtain a copy of the License at | ||
| # | ||
| # http://www.apache.org/licenses/LICENSE-2.0 | ||
| # | ||
| # Unless required by applicable law or agreed to in writing, software | ||
| # distributed under the License is distributed on an "AS IS" BASIS, | ||
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| # See the License for the specific language governing permissions and | ||
| # limitations under the License. | ||
| # | ||
|
|
||
| import logging | ||
| import os | ||
| import re | ||
|
|
||
| import chip.interaction_model | ||
| import pytest | ||
| from chip.clusters.Objects import OtaSoftwareUpdateRequestor | ||
| from chip.clusters.Types import NullValue | ||
| from common.utils import (connect_device, disconnect_device, discover_device, get_setup_payload, send_zcl_command, | ||
| write_zcl_attribute) | ||
|
|
||
| log = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| @pytest.fixture(scope="session") | ||
| def binaryPath(request, rootDir): | ||
| if request.config.getoption('binaryPath'): | ||
| return request.config.getoption('binaryPath') | ||
| else: | ||
| return os.path.join(rootDir, 'examples/ota-requestor-app/openiotsdk/build/chip-openiotsdk-ota-requestor-app-example.elf') | ||
|
|
||
|
|
||
| @pytest.fixture(scope="session") | ||
| def updateBinaryPath(request, rootDir): | ||
| if request.config.getoption('updateBinaryPath'): | ||
| return request.config.getoption('updateBinaryPath') | ||
| else: | ||
| return os.path.join(rootDir, 'examples/ota-requestor-app/openiotsdk/build/chip-openiotsdk-ota-requestor-app-example.ota') | ||
|
|
||
|
|
||
| @pytest.fixture(scope="session") | ||
| def controllerConfig(request): | ||
| config = { | ||
| 'vendorId': 0xFFF1, | ||
| 'fabricId': 1, | ||
| 'persistentStoragePath': '/tmp/openiotsdk-test-storage.json' | ||
| } | ||
| return config | ||
|
|
||
|
|
||
| @pytest.fixture(scope="session") | ||
| def otaProviderConfig(request, updateBinaryPath): | ||
| config = { | ||
| 'discriminator': '3841', | ||
| 'port': '5580', | ||
| 'filePath': f'{updateBinaryPath}', | ||
| 'persistentStoragePath': '/tmp/openiotsdk-test-ota-provider.json' | ||
| } | ||
| return config | ||
|
|
||
|
|
||
| @pytest.mark.smoketest | ||
| def test_smoke_test(device): | ||
| ret = device.wait_for_output("Open IoT SDK ota-requestor-app example application start") | ||
| assert ret is not None and len(ret) > 0 | ||
| ret = device.wait_for_output("Open IoT SDK ota-requestor-app example application run", timeout=30) | ||
| assert ret is not None and len(ret) > 0 | ||
|
|
||
|
|
||
| @pytest.mark.commissioningtest | ||
| def test_commissioning(device, controller): | ||
| assert controller is not None | ||
| devCtrl = controller | ||
|
|
||
| ret = device.wait_for_output("Open IoT SDK ota-requestor-app example application start") | ||
| assert ret is not None and len(ret) > 0 | ||
|
|
||
| setupPayload = get_setup_payload(device) | ||
| assert setupPayload is not None | ||
|
|
||
| commissionable_device = discover_device(devCtrl, setupPayload) | ||
| assert commissionable_device is not None | ||
|
|
||
| assert commissionable_device.vendorId == int(setupPayload.attributes['VendorID']) | ||
| assert commissionable_device.productId == int(setupPayload.attributes['ProductID']) | ||
| assert commissionable_device.addresses[0] is not None | ||
|
|
||
| nodeId = connect_device(devCtrl, setupPayload, commissionable_device) | ||
| assert nodeId is not None | ||
| log.info("Device {} connected".format(commissionable_device.addresses[0])) | ||
|
|
||
| ret = device.wait_for_output("Commissioning completed successfully", timeout=30) | ||
| assert ret is not None and len(ret) > 0 | ||
|
|
||
| assert disconnect_device(devCtrl, nodeId) | ||
|
|
||
|
|
||
| OTA_REQUESTOR_CTRL_TEST_ENDPOINT_ID = 0 | ||
|
|
||
|
|
||
| @pytest.mark.ctrltest | ||
| def test_update_ctrl(device, controller, ota_provider, softwareVersion): | ||
| assert controller is not None | ||
| devCtrl = controller | ||
| version_number, version_str = softwareVersion | ||
|
|
||
| log.info("Setup OTA provider...") | ||
|
|
||
| # Get OTA provider setup payload | ||
| setupPayloadProvider = get_setup_payload(ota_provider) | ||
| assert setupPayloadProvider is not None | ||
|
|
||
| # Discover and commission the OTA provider | ||
| commissionable_provider_device = discover_device(devCtrl, setupPayloadProvider) | ||
| assert commissionable_provider_device is not None | ||
|
|
||
| providerNodeId = connect_device(devCtrl, setupPayloadProvider, commissionable_provider_device) | ||
| assert providerNodeId is not None | ||
|
|
||
| ret = ota_provider.wait_for_output("Commissioning completed successfully", timeout=30) | ||
| assert ret is not None and len(ret) > 0 | ||
|
|
||
| log.info("OTA provider ready") | ||
| log.info("Setup OTA requestor...") | ||
|
|
||
| # Get OTA requestor setup payload | ||
| setupPayload = get_setup_payload(device) | ||
| assert setupPayload is not None | ||
|
|
||
| # Discover and commission the OTA requestor | ||
| commissionable_requestor_device = discover_device(devCtrl, setupPayload) | ||
| assert commissionable_requestor_device is not None | ||
|
|
||
| requestorNodeId = connect_device(devCtrl, setupPayload, commissionable_requestor_device) | ||
| assert requestorNodeId is not None | ||
|
|
||
| ret = device.wait_for_output("Commissioning completed successfully", timeout=30) | ||
| assert ret is not None and len(ret) > 0 | ||
|
|
||
| log.info("OTA requestor ready") | ||
| log.info("Install ACL entries") | ||
|
|
||
| # Install necessary ACL entries in OTA provider to enable access by OTA requestor | ||
| err, res = write_zcl_attribute(devCtrl, "AccessControl", "Acl", providerNodeId, OTA_REQUESTOR_CTRL_TEST_ENDPOINT_ID, | ||
| [{"fabricIndex": 1, "privilege": 5, "authMode": 2, "subjects": [requestorNodeId], "targets": NullValue}, | ||
| {"fabricIndex": 1, "privilege": 3, "authMode": 2, "subjects": NullValue, "targets": [{"cluster": 41, "endpoint": NullValue, "deviceType": NullValue}]}]) | ||
| assert err == 0 | ||
| assert res[0].Status == chip.interaction_model.Status.Success | ||
|
|
||
| ota_provider.set_verbose(False) | ||
|
|
||
| log.info("Announce the OTA provider and start the firmware update process") | ||
|
|
||
| # Announce the OTA provider and start the firmware update process | ||
| err, res = send_zcl_command(devCtrl, "OtaSoftwareUpdateRequestor", "AnnounceOTAProvider", requestorNodeId, OTA_REQUESTOR_CTRL_TEST_ENDPOINT_ID, | ||
| dict(providerNodeID=providerNodeId, vendorID=int(setupPayloadProvider.attributes['VendorID']), | ||
| announcementReason=OtaSoftwareUpdateRequestor.Enums.OTAAnnouncementReason.kUrgentUpdateAvailable, | ||
| metadataForNode=None, endpoint=0)) | ||
|
|
||
| ret = device.wait_for_output("New version of the software is available", timeout=30) | ||
| assert ret is not None and len(ret) > 1 | ||
|
|
||
| version = ret[-1].split()[-1] | ||
| assert version_number == version | ||
|
|
||
| device.set_verbose(False) | ||
|
|
||
| log.info("New software image downloading and installing...") | ||
|
|
||
| ret = device.wait_for_output("Open IoT SDK ota-requestor-app example application start", timeout=1200) | ||
| assert ret is not None and len(ret) > 0 | ||
|
|
||
| device.set_verbose(True) | ||
|
|
||
| ret = device.wait_for_output("Current software version", timeout=30) | ||
| assert ret is not None and len(ret) > 1 | ||
|
|
||
| version_app = ret[-1].split()[-2:] | ||
| assert version_number == re.sub(r"[\[\]]", "", version_app[0]) | ||
| assert version_str == version_app[1] | ||
|
|
||
| assert disconnect_device(devCtrl, requestorNodeId) | ||
| assert disconnect_device(devCtrl, providerNodeId) |