blob: e558d5389ba2da6596790b5257cdada22724784e [file] [log] [blame]
#
# 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.
#
import argparse
from binascii import unhexlify, hexlify
import logging
from chip import ChipDeviceCtrl
import chip.clusters as Clusters
from chip.ChipStack import *
from chip.storage import PersistentStorage
import chip.logging
import chip.native
import chip.FabricAdmin
import chip.CertificateAuthority
from chip.utils import CommissioningBuildingBlocks
import builtins
from typing import Optional, List, Tuple
from dataclasses import dataclass, field
from dataclasses import asdict as dataclass_asdict
import re
import os
import sys
import pathlib
import json
import uuid
import asyncio
from mobly import base_test
from mobly.config_parser import TestRunConfig, ENV_MOBLY_LOGPATH
from mobly import logger
from mobly import signals
from mobly import utils
from mobly.test_runner import TestRunner
# TODO: Add utility to commission a device if needed
# TODO: Add utilities to keep track of controllers/fabrics
logger = logging.getLogger("matter.python_testing")
logger.setLevel(logging.INFO)
DiscoveryFilterType = ChipDeviceCtrl.DiscoveryFilterType
_DEFAULT_ADMIN_VENDOR_ID = 0xFFF1
_DEFAULT_STORAGE_PATH = "admin_storage.json"
_DEFAULT_LOG_PATH = "/tmp/matter_testing/logs"
_DEFAULT_CONTROLLER_NODE_ID = 112233
_DEFAULT_DUT_NODE_ID = 111
_DEFAULT_TRUST_ROOT_INDEX = 1
# Mobly cannot deal with user config passing of ctypes objects,
# so we use this dict of uuid -> object to recover items stashed
# by reference.
_GLOBAL_DATA = {}
def stash_globally(o: object) -> str:
id = str(uuid.uuid1())
_GLOBAL_DATA[id] = o
return id
def unstash_globally(id: str) -> object:
return _GLOBAL_DATA.get(id)
def default_paa_rootstore_from_root(root_path: pathlib.Path) -> Optional[pathlib.Path]:
"""Attempt to find a PAA trust store following SDK convention at `root_path`
This attempts to find {root_path}/credentials/development/paa-root-certs.
Returns the fully resolved path on success or None if not found.
"""
start_path = root_path.resolve()
cred_path = start_path.joinpath("credentials")
dev_path = cred_path.joinpath("development")
paa_path = dev_path.joinpath("paa-root-certs")
return paa_path.resolve() if all([path.exists() for path in [cred_path, dev_path, paa_path]]) else None
def get_default_paa_trust_store(root_path: pathlib.Path) -> pathlib.Path:
"""Attempt to find a PAA trust store starting at `root_path`.
This tries to find by various heuristics, and goes up one level at a time
until found. After a given number of levels, it will stop.
This returns `root_path` if not PAA store is not found.
"""
# TODO: Add heuristics about TH default PAA location
cur_dir = pathlib.Path.cwd()
max_levels = 10
for level in range(max_levels):
paa_trust_store_path = default_paa_rootstore_from_root(cur_dir)
if paa_trust_store_path is not None:
return paa_trust_store_path
# Go back one level
cur_dir = cur_dir.joinpath("..")
else:
# On not having found a PAA dir, just return current dir to avoid blow-ups
return pathlib.Path.cwd()
@dataclass
class MatterTestConfig:
storage_path: pathlib.Path = None
logs_path: pathlib.Path = None
paa_trust_store_path: pathlib.Path = None
ble_interface_id: int = None
commission_only: bool = False
admin_vendor_id: int = _DEFAULT_ADMIN_VENDOR_ID
case_admin_subject: int = None
global_test_params: dict = field(default_factory=dict)
# List of explicit tests to run by name. If empty, all tests will run
tests: List[str] = field(default_factory=list)
commissioning_method: str = None
discriminator: int = None
setup_passcode: int = None
commissionee_ip_address_just_for_testing: str = None
maximize_cert_chains: bool = False
qr_code_content: str = None
manual_code: str = None
wifi_ssid: str = None
wifi_passphrase: str = None
thread_operational_dataset: str = None
# Node ID for basic DUT
dut_node_id: int = _DEFAULT_DUT_NODE_ID
# Node ID to use for controller/commissioner
controller_node_id: int = _DEFAULT_CONTROLLER_NODE_ID
# CAT Tags for default controller/commissioner
controller_cat_tags: List[int] = field(default_factory=list)
# Fabric ID which to use
fabric_id: int = None
# "Alpha" by default
root_of_trust_index: int = _DEFAULT_TRUST_ROOT_INDEX
# If this is set, we will reuse root of trust keys at that location
chip_tool_credentials_path: pathlib.Path = None
class MatterStackState:
def __init__(self, config: MatterTestConfig):
self._logger = logger
self._config = config
if not hasattr(builtins, "chipStack"):
chip.native.Init(bluetoothAdapter=config.ble_interface_id)
if config.storage_path is None:
raise ValueError("Must have configured a MatterTestConfig.storage_path")
self._init_stack(already_initialized=False, persistentStoragePath=config.storage_path)
self._we_initialized_the_stack = True
else:
self._init_stack(already_initialized=True)
self._we_initialized_the_stack = False
def _init_stack(self, already_initialized: bool, **kwargs):
if already_initialized:
self._chip_stack = builtins.chipStack
self._logger.warn(
"Re-using existing ChipStack object found in current interpreter: storage path %s will be ignored!" % (self._config.storage_path))
# TODO: Warn that storage will not follow what we set in config
else:
self._chip_stack = ChipStack(**kwargs)
builtins.chipStack = self._chip_stack
self._storage = self._chip_stack.GetStorageManager()
self._certificate_authority_manager = chip.CertificateAuthority.CertificateAuthorityManager(chipStack=self._chip_stack)
self._certificate_authority_manager.LoadAuthoritiesFromStorage()
if (len(self._certificate_authority_manager.activeCaList) == 0):
self._logger.warn(
"Didn't find any CertificateAuthorities in storage -- creating a new CertificateAuthority + FabricAdmin...")
ca = self._certificate_authority_manager.NewCertificateAuthority(caIndex=self._config.root_of_trust_index)
ca.maximizeCertChains = self._config.maximize_cert_chains
ca.NewFabricAdmin(vendorId=0xFFF1, fabricId=self._config.fabric_id)
elif (len(self._certificate_authority_manager.activeCaList[0].adminList) == 0):
self._logger.warn("Didn't find any FabricAdmins in storage -- creating a new one...")
self._certificate_authority_manager.activeCaList[0].NewFabricAdmin(vendorId=0xFFF1, fabricId=self._config.fabric_id)
# TODO: support getting access to chip-tool credentials issuer's data
def Shutdown(self):
if self._we_initialized_the_stack:
# Unfortunately, all the below are singleton and possibly
# managed elsewhere so we have to be careful not to touch unless
# we initialized ourselves.
self._certificate_authority_manager.Shutdown()
global_chip_stack = builtins.chipStack
global_chip_stack.Shutdown()
@property
def certificate_authorities(self):
return self._certificate_authority_manager.activeCaList
@property
def certificate_authority_manager(self):
return self._certificate_authority_manager
@property
def storage(self) -> PersistentStorage:
return self._storage
@property
def stack(self) -> ChipStack:
return builtins.chipStack
def bytes_from_hex(hex: str) -> bytes:
"""Converts any `hex` string representation including `01:ab:cd` to bytes
Handles any whitespace including newlines, which are all stripped.
"""
return unhexlify("".join(hex.replace(":", "").replace(" ", "").split()))
def hex_from_bytes(b: bytes) -> str:
"""Converts a bytes object `b` into a hex string (reverse of bytes_from_hex)"""
return hexlify(b).decode("utf-8")
class MatterBaseTest(base_test.BaseTestClass):
def __init__(self, *args):
super().__init__(*args)
@property
def matter_test_config(self) -> MatterTestConfig:
return unstash_globally(self.user_params.get("matter_test_config"))
@property
def default_controller(self) -> ChipDeviceCtrl:
return unstash_globally(self.user_params.get("default_controller"))
@property
def matter_stack(self) -> MatterStackState:
return unstash_globally(self.user_params.get("matter_stack"))
@property
def certificate_authority_manager(self) -> chip.CertificateAuthority.CertificateAuthorityManager:
return unstash_globally(self.user_params.get("certificate_authority_manager"))
@property
def dut_node_id(self) -> int:
return self.matter_test_config.dut_node_id
async def read_single_attribute(self, dev_ctrl: ChipDeviceCtrl, node_id: int, endpoint: int, attribute: object) -> object:
result = await dev_ctrl.ReadAttribute(node_id, [(endpoint, attribute)])
data = result[endpoint]
return list(data.values())[0][attribute]
def generate_mobly_test_config(matter_test_config: MatterTestConfig):
test_run_config = TestRunConfig()
# We use a default name. We don't use Mobly YAML configs, so that we can be
# freestanding without relying
test_run_config.testbed_name = "MatterTest"
log_path = matter_test_config.logs_path
log_path = _DEFAULT_LOG_PATH if log_path is None else log_path
if ENV_MOBLY_LOGPATH in os.environ:
log_path = os.environ[ENV_MOBLY_LOGPATH]
test_run_config.log_path = log_path
# TODO: For later, configure controllers
test_run_config.controller_configs = {}
test_run_config.user_params = matter_test_config.global_test_params
return test_run_config
def _find_test_class():
"""Finds the test class in a test script.
Walk through module members and find the subclass of MatterBaseTest. Only
one subclass is allowed in a test script.
Returns:
The test class in the test module.
Raises:
SystemExit: Raised if the number of test classes is not exactly one.
"""
subclasses = utils.find_subclasses_in_module([MatterBaseTest], sys.modules['__main__'])
subclasses = [c for c in subclasses if c.__name__ != "MatterBaseTest"]
if len(subclasses) != 1:
print(
'Exactly one subclass of `MatterBaseTest` should be in the main file. Found %s.' %
str([subclass.__name__ for subclass in subclasses]))
sys.exit(1)
return subclasses[0]
def int_decimal_or_hex(s: str) -> int:
val = int(s, 0)
if val < 0:
raise ValueError("Negative values not supported")
return val
def byte_string_from_hex(s: str) -> bytes:
return unhexlify(s.replace(":", "").replace(" ", "").replace("0x", ""))
def int_from_manual_code(s: str) -> int:
regex = r"^([0-9]{11}|[0-9]{21})$"
match = re.match(regex, s)
if not match:
raise ValueError("Invalid manual code format, does not match %s" % regex)
return int(s, 10)
def int_named_arg(s: str) -> Tuple[str, int]:
regex = r"^(?P<name>[a-zA-Z_0-9.]+):((?P<hex_value>0x[0-9a-fA-F_]+)|(?P<decimal_value>-?\d+))$"
match = re.match(regex, s)
if not match:
raise ValueError("Invalid int argument format, does not match %s" % regex)
name = match.group("name")
if match.group("hex_value"):
value = int(match.group("hex_value"), 0)
else:
value = int(match.group("decimal_value"), 10)
return (name, value)
def str_named_arg(s: str) -> Tuple[str, str]:
regex = r"^(?P<name>[a-zA-Z_0-9.]+):(?P<value>.*)$"
match = re.match(regex, s)
if not match:
raise ValueError("Invalid string argument format, does not match %s" % regex)
return (match.group("name"), match.group("value"))
def float_named_arg(s: str) -> Tuple[str, float]:
regex = r"^(?P<name>[a-zA-Z_0-9.]+):(?P<value>.*)$"
match = re.match(regex, s)
if not match:
raise ValueError("Invalid float argument format, does not match %s" % regex)
name = match.group("name")
value = float(match.group("value"))
return (name, value)
def json_named_arg(s: str) -> Tuple[str, object]:
regex = r"^(?P<name>[a-zA-Z_0-9.]+):(?P<value>.*)$"
match = re.match(regex, s)
if not match:
raise ValueError("Invalid JSON argument format, does not match %s" % regex)
name = match.group("name")
value = json.loads(match.group("value"))
return (name, value)
def bool_named_arg(s: str) -> Tuple[str, bool]:
regex = r"^(?P<name>[a-zA-Z_0-9.]+):((?P<truth_value>true|false)|(?P<decimal_value>[01]))$"
match = re.match(regex, s.lower())
if not match:
raise ValueError("Invalid bool argument format, does not match %s" % regex)
name = match.group("name")
if match.group("truth_value"):
value = True if match.group("truth_value") == "true" else False
else:
value = int(match.group("decimal_value")) != 0
return (name, value)
def bytes_as_hex_named_arg(s: str) -> Tuple[str, bytes]:
regex = r"^(?P<name>[a-zA-Z_0-9.]+):(?P<value>[0-9a-fA-F:]+)$"
match = re.match(regex, s)
if not match:
raise ValueError("Invalid bytes as hex argument format, does not match %s" % regex)
name = match.group("name")
value_str = match.group("value")
value_str = value_str.replace(":", "")
if len(value_str) % 2 != 0:
raise ValueError("Byte string argument value needs to be event number of hex chars")
value = unhexlify(value_str)
return (name, value)
def root_index(s: str) -> int:
CHIP_TOOL_COMPATIBILITY = {
"alpha": 1,
"beta": 2,
"gamma": 3
}
for name, id in CHIP_TOOL_COMPATIBILITY.items():
if s.lower() == name:
return id
else:
root_index = int(s)
if root_index == 0:
raise ValueError("Only support root index >= 1")
return root_index
def populate_commissioning_args(args: argparse.Namespace, config: MatterTestConfig) -> bool:
if args.commissioning_method is None:
return True
config.commissioning_method = args.commissioning_method
config.commission_only = args.commission_only
if args.dut_node_id is None:
print("error: When --commissioning-method present, --dut-node-id is mandatory!")
return False
config.dut_node_id = args.dut_node_id
if args.discriminator is None and (args.qr_code is None and args.manual_code is None):
print("error: Missing --discriminator when no --qr-code/--manual-code present!")
return False
config.discriminator = args.discriminator
if args.passcode is None and (args.qr_code is None and args.manual_code is None):
print("error: Missing --passcode when no --qr-code/--manual-code present!")
return False
config.setup_passcode = args.passcode
if args.qr_code is not None and args.manual_code is not None:
print("error: Cannot have both --qr-code and --manual-code present!")
return False
config.qr_code_content = args.qr_code
config.manual_code = args.manual_code
config.root_of_trust_index = args.root_index
# Follow root of trust index if ID not provided to have same behavior as legacy
# chip-tool that fabricID == commissioner_name == root of trust index
config.fabric_id = args.fabric_id if args.fabric_id is not None else config.root_of_trust_index
if args.chip_tool_credentials_path is not None and not args.chip_tool_credentials_path.exists():
print("error: chip-tool credentials path %s doesn't exist!" % args.chip_tool_credentials_path)
return False
config.chip_tool_credentials_path = args.chip_tool_credentials_path
if config.commissioning_method == "ble-wifi":
if args.wifi_ssid is None:
print("error: missing --wifi-ssid <SSID> for --commissioning-method ble-wifi!")
return False
if args.wifi_passphrase is None:
print("error: missing --wifi-passphrase <passphrasse> for --commissioning-method ble-wifi!")
return False
config.wifi_ssid = args.wifi_ssid
config.wifi_passphrase = args.wifi_passphrase
elif config.commissioning_method == "ble-thread":
if args.thread_dataset_hex is None:
print("error: missing --thread-dataset-hex <DATASET_HEX> for --commissioning-method ble-thread!")
return False
config.thread_operational_dataset = args.thread_dataset_hex
elif config.commissioning_method == "on-network-ip":
if args.ip_addr is None:
print("error: missing --ip-addr <IP_ADDRESS> for --commissioning-method on-network-ip")
return False
config.commissionee_ip_address_just_for_testing = args.ip_addr
if args.case_admin_subject is None:
# Use controller node ID as CASE admin subject during commissioning if nothing provided
config.case_admin_subject = config.controller_node_id
else:
# If a CASE admin subject is provided, then use that
config.case_admin_subject = args.case_admin_subject
return True
def convert_args_to_matter_config(args: argparse.Namespace) -> MatterTestConfig:
config = MatterTestConfig()
# Populate commissioning config if present, exiting on error
if not populate_commissioning_args(args, config):
sys.exit(1)
config.storage_path = pathlib.Path(_DEFAULT_STORAGE_PATH) if args.storage_path is None else args.storage_path
config.logs_path = pathlib.Path(_DEFAULT_LOG_PATH) if args.logs_path is None else args.logs_path
config.paa_trust_store_path = args.paa_trust_store_path
config.ble_interface_id = args.ble_interface_id
config.controller_node_id = args.controller_node_id
# Accumulate all command-line-passed named args
all_global_args = []
argsets = [item for item in (args.int_arg, args.float_arg, args.string_arg, args.json_arg,
args.hex_arg, args.bool_arg) if item is not None]
for argset in argsets:
all_global_args.extend(argset)
config.global_test_params = {}
for name, value in all_global_args:
config.global_test_params[name] = value
# Embed the rest of the config in the global test params dict which will be passed to Mobly tests
config.global_test_params["meta_config"] = {k: v for k, v in dataclass_asdict(config).items() if k != "global_test_params"}
return config
def parse_matter_test_args(argv: List[str]) -> MatterTestConfig:
parser = argparse.ArgumentParser(description='Matter standalone Python test')
basic_group = parser.add_argument_group(title="Basic arguments", description="Overall test execution arguments")
basic_group.add_argument('--tests',
'--test_case',
action="store",
nargs='+',
type=str,
metavar='test_a test_b...',
help='A list of tests in the test class to execute.')
basic_group.add_argument('--storage-path', action="store", type=pathlib.Path,
metavar="PATH", help="Location for persisted storage of instance")
basic_group.add_argument('--logs-path', action="store", type=pathlib.Path, metavar="PATH", help="Location for test logs")
paa_path_default = get_default_paa_trust_store(pathlib.Path.cwd())
basic_group.add_argument('--paa-trust-store-path', action="store", type=pathlib.Path, metavar="PATH", default=paa_path_default,
help="PAA trust store path (default: %s)" % str(paa_path_default))
basic_group.add_argument('--ble-interface-id', action="store", type=int,
metavar="INTERFACE_ID", help="ID of BLE adapter (from hciconfig)")
basic_group.add_argument('-N', '--controller-node-id', type=int_decimal_or_hex,
metavar='NODE_ID',
default=_DEFAULT_CONTROLLER_NODE_ID,
help='NodeID to use for initial/default controller (default: %d)' % _DEFAULT_CONTROLLER_NODE_ID)
basic_group.add_argument('-n', '--dut-node-id', type=int_decimal_or_hex,
metavar='NODE_ID', default=_DEFAULT_DUT_NODE_ID,
help='Node ID for primary DUT communication, and NodeID to assign if commissioning (default: %d)' % _DEFAULT_DUT_NODE_ID)
commission_group = parser.add_argument_group(title="Commissioning", description="Arguments to commission a node")
commission_group.add_argument('-m', '--commissioning-method', type=str,
metavar='METHOD_NAME',
choices=["on-network", "ble-wifi", "ble-thread", "on-network-ip"],
help='Name of commissioning method to use')
commission_group.add_argument('-d', '--discriminator', type=int_decimal_or_hex,
metavar='LONG_DISCRIMINATOR',
help='Discriminator to use for commissioning')
commission_group.add_argument('-p', '--passcode', type=int_decimal_or_hex,
metavar='PASSCODE',
help='PAKE passcode to use')
commission_group.add_argument('-i', '--ip-addr', type=str,
metavar='RAW_IP_ADDRESS',
help='IP address to use (only for method "on-network-ip". ONLY FOR LOCAL TESTING!')
commission_group.add_argument('--wifi-ssid', type=str,
metavar='SSID',
help='Wi-Fi SSID for ble-wifi commissioning')
commission_group.add_argument('--wifi-passphrase', type=str,
metavar='PASSPHRASE',
help='Wi-Fi passphrase for ble-wifi commissioning')
commission_group.add_argument('--thread-dataset-hex', type=byte_string_from_hex,
metavar='OPERATIONAL_DATASET_HEX',
help='Thread operational dataset as a hex string for ble-thread commissioning')
commission_group.add_argument('--admin-vendor-id', action="store", type=int_decimal_or_hex, default=_DEFAULT_ADMIN_VENDOR_ID,
metavar="VENDOR_ID", help="VendorID to use during commissioning (default 0x%04X)" % _DEFAULT_ADMIN_VENDOR_ID)
commission_group.add_argument('--case-admin-subject', action="store", type=int_decimal_or_hex,
metavar="CASE_ADMIN_SUBJECT", help="Set the CASE admin subject to an explicit value (default to commissioner Node ID)")
commission_group.add_argument('--commission-only', action="store_true", default=False,
help="If true, test exits after commissioning without running subsequent tests")
code_group = parser.add_mutually_exclusive_group(required=False)
code_group.add_argument('-q', '--qr-code', type=str,
metavar="QR_CODE", help="QR setup code content (overrides passcode and discriminator)")
code_group.add_argument('--manual-code', type=int_from_manual_code,
metavar="MANUAL_CODE", help="Manual setup code content (overrides passcode and discriminator)")
fabric_group = parser.add_argument_group(
title="Fabric selection", description="Fabric selection for single-fabric basic usage, and commissioning")
fabric_group.add_argument('-f', '--fabric-id', type=int_decimal_or_hex,
metavar='FABRIC_ID',
help='Fabric ID on which to operate under the root of trust')
fabric_group.add_argument('-r', '--root-index', type=root_index,
metavar='ROOT_INDEX_OR_NAME', default=_DEFAULT_TRUST_ROOT_INDEX,
help='Root of trust under which to operate/commission for single-fabric basic usage. alpha/beta/gamma are aliases for 1/2/3. Default (%d)' % _DEFAULT_TRUST_ROOT_INDEX)
fabric_group.add_argument('-c', '--chip-tool-credentials-path', type=pathlib.Path,
metavar='PATH',
help='Path to chip-tool credentials file root')
args_group = parser.add_argument_group(title="Config arguments", description="Test configuration global arguments set")
args_group.add_argument('--int-arg', nargs='*', type=int_named_arg, metavar="NAME:VALUE",
help="Add a named test argument for an integer as hex or decimal (e.g. -2 or 0xFFFF_1234)")
args_group.add_argument('--bool-arg', nargs='*', type=bool_named_arg, metavar="NAME:VALUE",
help="Add a named test argument for an boolean value (e.g. true/false or 0/1)")
args_group.add_argument('--float-arg', nargs='*', type=float_named_arg, metavar="NAME:VALUE",
help="Add a named test argument for a floating point value (e.g. -2.1 or 6.022e23)")
args_group.add_argument('--string-arg', nargs='*', type=str_named_arg, metavar="NAME:VALUE",
help="Add a named test argument for a string value")
args_group.add_argument('--json-arg', nargs='*', type=json_named_arg, metavar="NAME:VALUE",
help="Add a named test argument for JSON stored as a list or dict")
args_group.add_argument('--hex-arg', nargs='*', type=bytes_as_hex_named_arg, metavar="NAME:VALUE",
help="Add a named test argument for an octet string in hex (e.g. 0011cafe or 00:11:CA:FE)")
if not argv:
argv = sys.argv[1:]
return convert_args_to_matter_config(parser.parse_known_args(argv)[0])
def async_test_body(body):
"""Decorator required to be applied whenever a `test_*` method is `async def`.
Since Mobly doesn't support asyncio directly, and the test methods are called
synchronously, we need a mechanism to allow an `async def` to be converted to
a asyncio-run synchronous method. This decorator does the wrapping.
"""
def async_runner(*args, **kwargs):
return asyncio.run(body(*args, **kwargs))
return async_runner
class CommissionDeviceTest(MatterBaseTest):
"""Test class auto-injected at the start of test list to commission a device when requested"""
def test_run_commissioning(self):
conf = self.matter_test_config
logging.info("Starting commissioning for root index %d, fabric ID 0x%016X, node ID 0x%016X" %
(conf.root_of_trust_index, conf.fabric_id, conf.dut_node_id))
logging.info("Commissioning method: %s" % conf.commissioning_method)
if not self._commission_device():
raise signals.TestAbortAll("Failed to commission node")
def _commission_device(self) -> bool:
dev_ctrl = self.default_controller
conf = self.matter_test_config
# TODO: support by manual code and QR
if conf.commissioning_method == "on-network":
return dev_ctrl.CommissionOnNetwork(nodeId=conf.dut_node_id, setupPinCode=conf.setup_passcode, filterType=DiscoveryFilterType.LONG_DISCRIMINATOR, filter=conf.discriminator)
elif conf.commissioning_method == "ble-wifi":
return dev_ctrl.CommissionWiFi(conf.discriminator, conf.setup_passcode, conf.dut_node_id, conf.wifi_ssid, conf.wifi_passphrase)
elif conf.commissioning_method == "ble-thread":
return dev_ctrl.CommissionThread(conf.discriminator, conf.setup_passcode, conf.dut_node_id, conf.thread_operational_dataset)
elif conf.commissioning_method == "on-network-ip":
logging.warning("==== USING A DIRECT IP COMMISSIONING METHOD NOT SUPPORTED IN THE LONG TERM ====")
return dev_ctrl.CommissionIP(ipaddr=conf.commissionee_ip_address_just_for_testing, setupPinCode=conf.setup_passcode, nodeid=conf.dut_node_id)
else:
raise ValueError("Invalid commissioning method %s!" % conf.commissioning_method)
def default_matter_test_main(argv=None, **kwargs):
"""Execute the test class in a test module.
This is the default entry point for running a test script file directly.
In this case, only one test class in a test script is allowed.
To make your test script executable, add the following to your file:
.. code-block:: python
from matter_testing_support.py import default_matter_test_main
...
if __name__ == '__main__':
default_matter_test_main.main()
Args:
argv: A list that is then parsed as command line args. If None, defaults to sys.argv
"""
matter_test_config = parse_matter_test_args(argv)
# Allow override of command line from optional arguments
if not matter_test_config.controller_cat_tags and "controller_cat_tags" in kwargs:
matter_test_config.controller_cat_tags = kwargs["controller_cat_tags"]
# Find the test class in the test script.
test_class = _find_test_class()
# Load test config file.
test_config = generate_mobly_test_config(matter_test_config)
# Parse test specifiers if exist.
tests = None
if len(matter_test_config.tests) > 0:
tests = matter_test_config.tests
# This is required in case we need any testing with maximized certificate chains.
# We need *all* issuers from the start, even for default controller, to use
# maximized chains, before MatterStackState init, others some stale certs
# may not chain properly.
if "maximize_cert_chains" in kwargs:
matter_test_config.maximize_cert_chains = kwargs["maximize_cert_chains"]
stack = MatterStackState(matter_test_config)
test_config.user_params["matter_stack"] = stash_globally(stack)
# TODO: Steer to right FabricAdmin!
# TODO: If CASE Admin Subject is a CAT tag range, then make sure to issue NOC with that CAT tag
default_controller = stack.certificate_authorities[0].adminList[0].NewController(nodeId=matter_test_config.controller_node_id,
paaTrustStorePath=str(matter_test_config.paa_trust_store_path), catTags=matter_test_config.controller_cat_tags)
test_config.user_params["default_controller"] = stash_globally(default_controller)
test_config.user_params["matter_test_config"] = stash_globally(matter_test_config)
test_config.user_params["certificate_authority_manager"] = stash_globally(stack.certificate_authority_manager)
# Execute the test class with the config
ok = True
runner = TestRunner(log_dir=test_config.log_path,
testbed_name=test_config.testbed_name)
with runner.mobly_logger():
if matter_test_config.commissioning_method is not None:
runner.add_test_class(test_config, CommissionDeviceTest, None)
# Add the tests selected unless we have a commission-only request
if not matter_test_config.commission_only:
runner.add_test_class(test_config, test_class, tests)
try:
runner.run()
ok = runner.results.is_all_pass and ok
except signals.TestAbortAll:
ok = False
except Exception:
logging.exception('Exception when executing %s.', test_config.testbed_name)
ok = False
# Shutdown the stack when all done
stack.Shutdown()
if ok:
logging.info("Final result: PASS !")
else:
logging.error("Final result: FAIL !")
sys.exit(1)