Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

TC-CNET-4_12 python test #38074

Draft
wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

TC-CNET-4_12 python test #38074

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
355 changes: 355 additions & 0 deletions src/python_testing/TC_CNET_4_12.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,355 @@
#
# Copyright (c) 2025 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.
#

# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
# for details about the block below.
#
# === BEGIN CI TEST ARGUMENTS ===
# test-runner-runs:
# run1:
# app: ${ALL_CLUSTERS_APP}
# app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
# script-args: >
# --storage-path admin_storage.json
# --commissioning-method on-network
# --discriminator 1234
# --passcode 20202021
# --PICS src/app/tests/suites/certification/ci-pics-values
# --string-arg PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET:${THREAD_1ST}
# --string-arg PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET:${THREAD_2ND}
# --trace-to json:${TRACE_TEST_JSON}.json
# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
# factory-reset: true
# quiet: true
# === END CI TEST ARGUMENTS ===

import logging

import chip.clusters as Clusters
from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
from mobly import asserts

logger = logging.getLogger(__name__)


class TC_CNET_4_12(MatterBaseTest):

CLUSTER_CNET = Clusters.NetworkCommissioning
CLUSTER_DESC = Clusters.Descriptor
CLUSTER_CGEN = Clusters.GeneralCommissioning
failsafe_expiration_seconds = 900
default_network_id = 'Thread'

def def_TC_CNET_4_12(self):
return '[TC-CNET-4.12] [Thread] Verification for ConnectNetwork Command [DUT-Server]'

def pics_TC_CNET_4_12(self):
"""Return the PICS definitions associated with this test."""
pics = [
"CNET.S"
]
return pics

def steps_TC_CNET_4_12(self) -> list[TestStep]:
steps = [
TestStep('precondition-1', 'DUT supports CNET.S.F01(TH)'),
TestStep('precondition-2', 'DUT has a Network Commissioning cluster on endpoint PIXIT.CNET.ENDPOINT_THREAD with FeatureMap attribute of 2', is_commissioning=True),
TestStep('precondition-3', 'DUT is commissioned on PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET'),
TestStep('precondition-4', 'TH has can communicate to two valid thread PANs: PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET and PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET'),
TestStep('precondition-5', 'XPANID of PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET is saved as th_xpan and the XPANID of PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET is saved as th_xpan1'),
TestStep(1, 'TH sends ArmFailSafe command to the DUT with ExpiryLengthSeconds set to 900'),
TestStep(2, 'TH reads Networks attribute from the DUT and saves the number of entries as NumNetworks'),
TestStep(3, 'TH saves the index of the Networks list entry from step 2 as Userth_netidx'),
TestStep(4, '''TH sends RemoveNetwork Command to the DUT
with NetworkID field set to th_xpan
and Breadcrumb field set to 1'''),
TestStep(5, '''TH sends AddOrUpdateThreadNetwork command to the DUT
with operational dataset field set to PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET
and Breadcrumb field set to 1'''),
TestStep(6, 'TH reads Networks attribute from the DUT'),
TestStep(7, '''TH sends ConnectNetwork command to the DUT
with NetworkID field set to PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET
and Breadcrumb field set to 2'''),
TestStep(8, 'TH discovers and connects to DUT on the PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET operational network'),
TestStep(9, 'TH reads Breadcrumb attribute from the General Commissioning cluster of the DUT'),
TestStep(10, 'TH sends ArmFailSafe command to the DUT with ExpiryLengthSeconds set to 0'),
TestStep(11, 'TH ensures it can communicate on PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET'),
TestStep(12, 'TH discovers and connects to DUT on the PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET operational network'),
TestStep(13, 'TH sends ArmFailSafe command to the DUT with ExpiryLengthSeconds set to 900'),
TestStep(14, '''TH sends RemoveNetwork Command to the DUT with NetworkID field set to th_xpan
and Breadcrumb field set to 1'''),
TestStep(15, '''TH sends AddOrUpdateThreadNetwork command to the DUT
with the OperationalDataset field set to PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET
and Breadcrumb field set to 1'''),
TestStep(16, '''TH sends ConnectNetwork command to the DUT
with NetworkID field set to the extended PAN ID of PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET
and Breadcrumb field set to 3'''),
TestStep(17, 'TH discovers and connects to DUT on the PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET operational network'),
TestStep(18, 'TH reads Breadcrumb attribute from the General Commissioning cluster of the DUT'),
TestStep(19, 'TH sends the CommissioningComplete command to the DUT'),
TestStep(20, 'TH reads Networks attribute from the DUT')
]
return steps

@async_test_body
async def test_TC_CNET_4_12(self):

if self.is_pics_sdk_ci_only:
logger.info('Test is not running in CI.')
self.skip_all_remaining_steps('precondition-1')
return

# Pre-Conditions
self.step('precondition-1')
self.step('precondition-2')
# By running this test from the terminal, it commissions the device.
logger.info('Pre-Conditions #1: DUT has a Network Commissioning cluster on endpoint PIXIT.CNET.ENDPOINT_THREAD ')

# The FeatureMap attribute value is 2
feature_map = await self.read_single_attribute_check_success(
cluster=self.CLUSTER_CNET,
attribute=self.CLUSTER_CNET.Attributes.FeatureMap)
asserts.assert_true(feature_map == 2,
msg="Verify that feature_map is equal to 1")
logger.info(f'Pre-Conditions #3: The FeatureMap attribute value is: {feature_map}')

self.step('precondition-3')
# TODO: Implement precondition-3
self.step('precondition-4')
# TODO: Implement precondition-4
self.step('precondition-5')
# TODO: Implement precondition-5
th_xpan = self.user_params.get('PIXIT.CNET.THREAD_1ST_OPERATIONALDATASET', self.default_network_id)
th_xpan_1 = self.user_params.get('PIXIT.CNET.THREAD_2ND_OPERATIONALDATASET', self.default_network_id)

# Steps

self.step(1)
cmd = Clusters.GeneralCommissioning.Commands.ArmFailSafe(expiryLengthSeconds=self.failsafe_expiration_seconds, breadcrumb=1)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
# Verify that the DUT responds with ArmFailSafeResponse with ErrorCode as 'OK'(0)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk,
"Failure status returned from arm failsafe")
logger.info(f'Step #1b - ArmFailSafeResponse with ErrorCode as OK({resp.errorCode})')

self.step(2)
networks = await self.read_single_attribute_check_success(
cluster=Clusters.NetworkCommissioning,
attribute=Clusters.NetworkCommissioning.Attributes.Networks
)
logger.info(f'Step #2: Networks attribute: {networks}')

num_networks = len(networks)
logger.info(f'Step #2: Number of Networks entries (NumNetworks): {num_networks}')
asserts.assert_true(num_networks > 0, "Error: No networks found")

# TODO: Implement proper validation to verify the the Networks attribute "NetworkID" and "Connected"
for cnet in networks:
if cnet.networkID.decode('utf-8') == th_xpan and cnet.connected:
network_found = True
break
logger.info(f'Step #2: Found network with ID {th_xpan} and connected={network_found}.')
asserts.assert_true(
network_found, f"Error: Network with ID {th_xpan} and connected=True not found.")

self.step(3)
# TODO: Implement proper validation to verify the the Networks index
userth_netidx = None
for index, network in enumerate(networks):
if network['networkID'] == th_xpan:
userth_netidx = index
break
asserts.assert_true(userth_netidx is not None, "")
logger.info(f'Step #3: Networks attribute: {userth_netidx}')

self.step(4)
cmd = Clusters.NetworkCommissioning.Commands.RemoveNetwork(networkID=th_xpan, breadcrumb=1)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
network_index = resp.networkIndex
logger.info(f'Step #4: RemoveNetwork Status is success ({resp.networkingStatus})')
logger.info(f'Step #4: Network index: ({network_index})')

# Verify that the DUT responds with Remove Network with NetworkingStatus as 'Success'(0)
asserts.assert_equal(resp.networkingStatus, Clusters.NetworkCommissioning.Enums.NetworkCommissioningStatusEnum.kSuccess,
"Failure status returned from ReordeRemove Network")
asserts.assert_equal(network_index, 0, "The network index is not as expected.")

self.step(5)
cmd = Clusters.NetworkCommissioning.Commands.AddOrUpdateThreadNetwork(operationalDataset=th_xpan_1, breadcrumb=1)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
logger.info(f'Step #5: AddOrUpdateThreadNetwork Status is success ({resp.networkingStatus})')
# Verify that the DUT responds with AddThreadNetwork with NetworkingStatus as 'Success'(0)
asserts.assert_equal(resp.networkingStatus, Clusters.NetworkCommissioning.Enums.NetworkCommissioningStatusEnum.kSuccess,
"Failure status returned from AddThreadNetwork")
debug_text = resp.debugText
# TODO: Check if None is part of the validation
asserts.assert_true(debug_text is None or debug_text == '' or len(debug_text) <= 512,
"debugText must be None, empty or have a maximum length of 512 characters.")

self.step(6)
networks = await self.read_single_attribute_check_success(
cluster=Clusters.NetworkCommissioning,
attribute=Clusters.NetworkCommissioning.Attributes.Networks
)
logger.info(f'Step #6: Networks attribute: {networks}')
# TODO; Implement the Verify that the Networks attribute list has an entry NetworkID=th_xpan, Connected=FALSE
# TODO: Why th_xpan? should be th_xpan_1

self.step(7)
cmd = Clusters.NetworkCommissioning.Commands.ConnectNetwork(operationalDataset=th_xpan_1, breadcrumb=2)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
logger.info(f'Step #7: ConnectNetwork Status is success ({resp.networkingStatus})')
# Verify that the DUT responds with AddThreadNConnectNetworketwork with NetworkingStatus as 'Success'(0)
asserts.assert_equal(resp.networkingStatus, Clusters.NetworkCommissioning.Enums.NetworkCommissioningStatusEnum.kSuccess,
"Failure status returned from ConnectNetwork")

self.step(8)
# TODO: Verify that the TH successfully connects to the DUT from previous step

self.step(9)
breadcrumb_info = await self.read_single_attribute_check_success(
cluster=Clusters.GeneralCommissioning,
attribute=Clusters.GeneralCommissioning.Attributes.Breadcrumb
)
logger.info(f'Step #9: Breadcrumb attribute: {breadcrumb_info}')
asserts.assert_equal(breadcrumb_info, 1,
"The Breadcrumb attribute is not 1")

self.step(10)
cmd = Clusters.GeneralCommissioning.Commands.ArmFailSafe(expiryLengthSeconds=0)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
# Verify that the DUT responds with ArmFailSafeResponse with ErrorCode as 'OK'(0)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk,
"Failure status returned from arm failsafe")
logger.info(f'Step #10 - ArmFailSafeResponse with ErrorCode as OK({resp.errorCode})')

self.step(11)
# TODO: Verify that TH can communicate on th_xpan

self.step(12)
# TODO: Verify that TH can discover and connect on th_xpan

self.step(13)
cmd = Clusters.GeneralCommissioning.Commands.ArmFailSafe(expiryLengthSeconds=self.failsafe_expiration_seconds)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
# Verify that the DUT responds with ArmFailSafeResponse with ErrorCode as 'OK'(0)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk,
"Failure status returned from arm failsafe")
logger.info(f'Step #1b - ArmFailSafeResponse with ErrorCode as OK({resp.errorCode})')

self.step(14)
cmd = Clusters.NetworkCommissioning.Commands.RemoveNetwork(networkID=th_xpan, breadcrumb=1)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
network_index = resp.networkIndex
logger.info(f'Step #14: RemoveNetwork Status is success ({resp.networkingStatus})')
logger.info(f'Step #14: Network index: ({network_index})')

# Verify that the DUT responds with Remove Network with NetworkingStatus as 'Success'(0)
asserts.assert_equal(resp.networkingStatus, Clusters.NetworkCommissioning.Enums.NetworkCommissioningStatusEnum.kSuccess,
"Failure status returned from ReordeRemove Network")
asserts.assert_equal(network_index, 0, "The network index is not as expected.")

self.step(15)
cmd = Clusters.NetworkCommissioning.Commands.AddOrUpdateThreadNetwork(operationalDataset=th_xpan_1, breadcrumb=3)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
logger.info(f'Step #15: AddOrUpdateThreadNetwork Status is success ({resp.networkingStatus})')
# Verify that the DUT responds with AddThreadNetwork with NetworkingStatus as 'Success'(0)
asserts.assert_equal(resp.networkingStatus, Clusters.NetworkCommissioning.Enums.NetworkCommissioningStatusEnum.kSuccess,
"Failure status returned from AddThreadNetwork")
debug_text = resp.debugText
# TODO: Check if None is part of the validation
asserts.assert_true(debug_text is None or debug_text == '' or len(debug_text) <= 512,
"debugText must be None, empty or have a maximum length of 512 characters.")

self.step(16)
cmd = Clusters.NetworkCommissioning.Commands.ConnectNetwork(operationalDataset=th_xpan_1, breadcrumb=3)
resp = await self.send_single_cmd(
dev_ctrl=self.default_controller,
node_id=self.dut_node_id,
cmd=cmd
)
logger.info(f'Step #16: ConnectNetwork Status is success ({resp.networkingStatus})')
# Verify that the DUT responds with AddThreadNConnectNetworketwork with NetworkingStatus as 'Success'(0)
asserts.assert_equal(resp.networkingStatus, Clusters.NetworkCommissioning.Enums.NetworkCommissioningStatusEnum.kSuccess,
"Failure status returned from ConnectNetwork")

self.step(17)
# TODO: Verify that the TH successfully connects to the DUT from previous step

self.step(18)
breadcrumb_info = await self.read_single_attribute_check_success(
cluster=Clusters.GeneralCommissioning,
attribute=Clusters.GeneralCommissioning.Attributes.Breadcrumb
)
logger.info(f'Step #18: Breadcrumb attribute: {breadcrumb_info}')
asserts.assert_equal(breadcrumb_info, 3,
"The Breadcrumb attribute is not 3")

self.step(19)
# TODO: Implement TH sends the CommissioningComplete and CommissioningCompleteResponse with the ErrorCode OK (0)

self.step(20)
networks = await self.read_single_attribute_check_success(
cluster=Clusters.NetworkCommissioning,
attribute=Clusters.NetworkCommissioning.Attributes.Networks
)
logger.info(f'Step #2: Networks attribute: {networks}')

# TODO: Implement proper validation to verify the the Networks attribute "NetworkID" and "Connected"
for cnet in networks:
if cnet.networkID.decode('utf-8') == th_xpan_1 and cnet.connected:
network_found = True
break
logger.info(f'Step #2: Found network with ID {th_xpan_1} and connected={network_found}.')
asserts.assert_true(
network_found, f"Error: Network with ID {th_xpan_1} and connected=True not found.")


if __name__ == "__main__":
default_matter_test_main()
Loading