| #!/usr/bin/env python3 |
| |
| # |
| # Copyright (c) 2022 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. |
| # |
| |
| # Commissioning test. |
| import os |
| import sys |
| from optparse import OptionParser |
| from base import TestFail, TestTimeout, BaseTestHelper, FailIfNot, logger |
| from cluster_objects import NODE_ID, ClusterObjectTests |
| from network_commissioning import NetworkCommissioningTests |
| import asyncio |
| |
| # The thread network dataset tlv for testing, splited into T-L-V. |
| |
| TEST_THREAD_NETWORK_DATASET_TLV = "0e080000000000010000" + \ |
| "000300000c" + \ |
| "35060004001fffe0" + \ |
| "0208fedcba9876543210" + \ |
| "0708fd00000000001234" + \ |
| "0510ffeeddccbbaa99887766554433221100" + \ |
| "030e54657374696e674e6574776f726b" + \ |
| "0102d252" + \ |
| "041081cb3b2efa781cc778397497ff520fa50c0302a0ff" |
| # Network id, for the thread network, current a const value, will be changed to XPANID of the thread network. |
| TEST_THREAD_NETWORK_ID = "fedcba9876543210" |
| TEST_DISCRIMINATOR = 3840 |
| |
| ENDPOINT_ID = 0 |
| LIGHTING_ENDPOINT_ID = 1 |
| GROUP_ID = 0 |
| |
| |
| def main(): |
| optParser = OptionParser() |
| optParser.add_option( |
| "-t", |
| "--timeout", |
| action="store", |
| dest="testTimeout", |
| default=75, |
| type='int', |
| help="The program will return with timeout after specified seconds.", |
| metavar="<timeout-second>", |
| ) |
| optParser.add_option( |
| "-a", |
| "--address", |
| action="store", |
| dest="deviceAddress1", |
| default='', |
| type='str', |
| help="Address of the first device", |
| ) |
| optParser.add_option( |
| '--paa-trust-store-path', |
| dest="paaPath", |
| default='', |
| type='str', |
| help="Path that contains valid and trusted PAA Root Certificates." |
| ) |
| optParser.add_option( |
| '--fail-on-report', |
| action="store_true", |
| dest="report", |
| default=False, |
| help='Use this flag to simulate a failure handling the report. Without this flag, failure is simulated on the stage' |
| ) |
| |
| (options, remainingArgs) = optParser.parse_args(sys.argv[1:]) |
| |
| timeoutTicker = TestTimeout(options.testTimeout) |
| timeoutTicker.start() |
| |
| test = BaseTestHelper(nodeid=112233, testCommissioner=True, |
| paaTrustStorePath=options.paaPath) |
| |
| FailIfNot(test.SetNetworkCommissioningParameters(dataset=TEST_THREAD_NETWORK_DATASET_TLV), |
| "Failed to set network commissioning parameters") |
| |
| logger.info("Testing PASE connection to device") |
| |
| # TODO: Start at stage 2 once handling for arming failsafe on pase is done. |
| if options.report: |
| for testFailureStage in range(3, 18): |
| FailIfNot(test.TestPaseOnly(ip=options.deviceAddress1, |
| setuppin=20202021, |
| nodeid=1), |
| "Failed to establish PASE connection with device") |
| FailIfNot(test.TestCommissionFailureOnReport(1, testFailureStage), |
| "Commissioning failure tests failed for simulated report failure on stage {}".format(testFailureStage)) |
| |
| else: |
| for testFailureStage in range(3, 18): |
| FailIfNot(test.TestPaseOnly(ip=options.deviceAddress1, |
| setuppin=20202021, |
| nodeid=1), |
| "Failed to establish PASE connection with device") |
| FailIfNot(test.TestCommissionFailure(1, testFailureStage), |
| "Commissioning failure tests failed for simulated stage failure on stage {}".format(testFailureStage)) |
| |
| # Ensure we can still commission for real |
| FailIfNot(test.TestPaseOnly(ip=options.deviceAddress1, |
| setuppin=20202021, |
| nodeid=1), |
| "Failed to establish PASE connection with device") |
| FailIfNot(test.TestCommissionFailure(1, 0), "Failed to commission device") |
| |
| logger.info("Testing on off cluster") |
| FailIfNot(test.TestOnOffCluster(nodeid=1, |
| endpoint=LIGHTING_ENDPOINT_ID, |
| group=GROUP_ID), "Failed to test on off cluster") |
| |
| timeoutTicker.stop() |
| |
| logger.info("Test finished") |
| |
| # TODO: Python device controller cannot be shutdown clean sometimes and will block on AsyncDNSResolverSockets shutdown. |
| # Call os._exit(0) to force close it. |
| os._exit(0) |
| |
| |
| if __name__ == "__main__": |
| try: |
| main() |
| except Exception as ex: |
| logger.exception(ex) |
| TestFail("Exception occurred when running tests.") |