src/controller/python/test/test_scripts/mobile-device-test.py

#!/usr/bin/env python3

#
#    Copyright (c) 2021 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.
from logging import disable
import os
import sys
import click
import coloredlogs
import chip.logging
import logging
from base import TestFail, TestTimeout, BaseTestHelper, FailIfNot, logger, TestIsEnabled, SetTestSet
import base
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
TEST_SETUPPIN = 20202021

ENDPOINT_ID = 0
LIGHTING_ENDPOINT_ID = 1
GROUP_ID = 0

TEST_CONTROLLER_NODE_ID = 112233
TEST_DEVICE_NODE_ID = 1

ALL_TESTS = ['network_commissioning', 'datamodel']


def ethernet_commissioning(test: BaseTestHelper, discriminator: int, setup_pin: int, address_override: str, device_nodeid: int):
    logger.info("Testing discovery")
    device = test.TestDiscovery(discriminator=discriminator)
    FailIfNot(device, "Failed to discover any devices.")

    address = device.addresses[0]

    # FailIfNot(test.SetNetworkCommissioningParameters(dataset=TEST_THREAD_NETWORK_DATASET_TLV),
    #           "Failed to finish network commissioning")

    if address_override:
        address = address_override

    logger.info("Testing commissioning")
    FailIfNot(test.TestCommissioning(ip=address,
                                     setuppin=setup_pin,
                                     nodeid=device_nodeid),
              "Failed to finish key exchange")

    logger.info("Testing multi-controller setup on the same fabric")
    FailIfNot(asyncio.run(test.TestMultiControllerFabric(nodeid=device_nodeid)), "Failed the multi-controller test")

    logger.info("Testing CATs used on controllers")
    FailIfNot(asyncio.run(test.TestControllerCATValues(nodeid=device_nodeid)), "Failed the controller CAT test")

    ok = asyncio.run(test.TestMultiFabric(ip=address,
                                          setuppin=20202021,
                                          nodeid=1))
    FailIfNot(ok, "Failed to commission multi-fabric")

    FailIfNot(asyncio.run(test.TestAddUpdateRemoveFabric(nodeid=device_nodeid)),
              "Failed AddUpdateRemoveFabric test")

    logger.info("Testing CASE Eviction")
    FailIfNot(asyncio.run(test.TestCaseEviction(device_nodeid)), "Failed TestCaseEviction")

    logger.info("Testing closing sessions")
    FailIfNot(test.TestCloseSession(nodeid=device_nodeid), "Failed to close sessions")


@base.test_case
def TestDatamodel(test: BaseTestHelper, device_nodeid: int):
    logger.info("Testing datamodel functions")

    logger.info("Testing on off cluster")
    FailIfNot(test.TestOnOffCluster(nodeid=device_nodeid,
                                    endpoint=LIGHTING_ENDPOINT_ID,
                                    group=GROUP_ID), "Failed to test on off cluster")

    logger.info("Testing level control cluster")
    FailIfNot(test.TestLevelControlCluster(nodeid=device_nodeid,
                                           endpoint=LIGHTING_ENDPOINT_ID,
                                           group=GROUP_ID),
              "Failed to test level control cluster")

    logger.info("Testing sending commands to non exist endpoint")
    FailIfNot(not test.TestOnOffCluster(nodeid=device_nodeid,
                                        endpoint=233,
                                        group=GROUP_ID), "Failed to test on off cluster on non-exist endpoint")

    # Test experimental Python cluster objects API
    logger.info("Testing cluster objects API")
    FailIfNot(asyncio.run(ClusterObjectTests.RunTest(test.devCtrl)),
              "Failed when testing Python Cluster Object APIs")

    logger.info("Testing attribute reading")
    FailIfNot(test.TestReadBasicAttributes(nodeid=device_nodeid,
                                           endpoint=ENDPOINT_ID,
                                           group=GROUP_ID),
              "Failed to test Read Basic Attributes")

    logger.info("Testing attribute writing")
    FailIfNot(test.TestWriteBasicAttributes(nodeid=device_nodeid,
                                            endpoint=ENDPOINT_ID,
                                            group=GROUP_ID),
              "Failed to test Write Basic Attributes")

    logger.info("Testing attribute reading basic again")
    FailIfNot(test.TestReadBasicAttributes(nodeid=1,
                                           endpoint=ENDPOINT_ID,
                                           group=GROUP_ID),
              "Failed to test Read Basic Attributes")

    logger.info("Testing subscription")
    FailIfNot(test.TestSubscription(nodeid=device_nodeid, endpoint=LIGHTING_ENDPOINT_ID),
              "Failed to subscribe attributes.")

    logger.info("Testing another subscription that kills previous subscriptions")
    FailIfNot(test.TestSubscription(nodeid=device_nodeid, endpoint=LIGHTING_ENDPOINT_ID),
              "Failed to subscribe attributes.")

    logger.info("Testing re-subscription")
    FailIfNot(asyncio.run(test.TestResubscription(nodeid=device_nodeid)),
              "Failed to validated re-subscription")

    logger.info("Testing on off cluster over resolved connection")
    FailIfNot(test.TestOnOffCluster(nodeid=device_nodeid,
                                    endpoint=LIGHTING_ENDPOINT_ID,
                                    group=GROUP_ID), "Failed to test on off cluster")

    logger.info("Testing writing/reading fabric sensitive data")
    asyncio.run(test.TestFabricSensitive(nodeid=device_nodeid))


def do_tests(controller_nodeid, device_nodeid, address, timeout, discriminator, setup_pin, paa_trust_store_path):
    timeoutTicker = TestTimeout(timeout)
    timeoutTicker.start()

    test = BaseTestHelper(nodeid=controller_nodeid,
                          paaTrustStorePath=paa_trust_store_path)

    chip.logging.RedirectToPythonLogging()

    ethernet_commissioning(test, discriminator, setup_pin, address,
                           device_nodeid)

    logger.info("Testing resolve")
    FailIfNot(test.TestResolve(nodeid=device_nodeid),
              "Failed to resolve nodeid")

    # Still test network commissioning
    FailIfNot(asyncio.run(NetworkCommissioningTests(devCtrl=test.devCtrl, nodeid=device_nodeid).run()),
              "Failed to finish network commissioning")

    TestDatamodel(test, device_nodeid)

    logger.info("Testing non-controller APIs")
    FailIfNot(test.TestNonControllerAPIs(), "Non controller API test failed")

    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)


@click.command()
@click.option("--controller-nodeid", default=TEST_CONTROLLER_NODE_ID, type=int, help="NodeId of the controller.")
@click.option("--device-nodeid", default=TEST_DEVICE_NODE_ID, type=int, help="NodeId of the device.")
@click.option("--address", "-a", default='', type=str, help="Skip commissionee discovery, commission the device with the IP directly.")
@click.option("--timeout", "-t", default=240, type=int, help="The program will return with timeout after specified seconds.")
@click.option("--discriminator", default=TEST_DISCRIMINATOR, type=int, help="Discriminator of the device.")
@click.option("--setup-pin", default=TEST_SETUPPIN, type=int, help="Setup pincode of the device.")
@click.option('--enable-test', default=['all'], type=str, multiple=True, help='The tests to be executed. By default, all tests will be executed, use this option to run a specific set of tests. Use --print-test-list for a list of appliable tests.')
@click.option('--disable-test', default=[], type=str, multiple=True, help='The tests to be excluded from the set of enabled tests. Use --print-test-list for a list of appliable tests.')
@click.option('--log-level', default='WARN', type=click.Choice(['ERROR', 'WARN', 'INFO', 'DEBUG']), help="The log level of the test.")
@click.option('--log-format', default=None, type=str, help="Override logging format")
@click.option('--print-test-list', is_flag=True, help="Print a list of test cases and test sets that can be toggled via --enable-test and --disable-test, then exit")
@click.option('--paa-trust-store-path', default='', type=str, help="Path that contains valid and trusted PAA Root Certificates.")
def run(controller_nodeid, device_nodeid, address, timeout, discriminator, setup_pin, enable_test, disable_test, log_level, log_format, print_test_list, paa_trust_store_path):
    coloredlogs.install(level=log_level, fmt=log_format, logger=logger)

    if print_test_list:
        print("Test sets:")
        for name in base.configurable_tests():
            print(f"\t{name}")
        print("Test cases:")
        for name in base.configurable_test_cases():
            print(f"\t{name}")
        return

    logger.info("Test Parameters:")
    logger.info(f"\tController NodeId: {controller_nodeid}")
    logger.info(f"\tDevice NodeId:     {device_nodeid}")
    logger.info(f"\tTest Timeout:      {timeout}s")
    logger.info(f"\tDiscriminator:     {discriminator}")
    logger.info(f"\tEnabled Tests:     {enable_test}")
    logger.info(f"\tDisabled Tests:    {disable_test}")
    SetTestSet(enable_test, disable_test)
    do_tests(controller_nodeid, device_nodeid, address, timeout,
             discriminator, setup_pin, paa_trust_store_path)


if __name__ == "__main__":
    try:
        run()
    except Exception as ex:
        logger.exception(ex)
        TestFail("Exception occurred when running tests.")