Skip to content

Commit e4502c3

Browse files
yufengwangcayufengwangca
authored
[Fabric-Sync] Implement the test plan for TC_CCTRL_2_3 (project-chip#35202)
* [Fabric-Sync] Implement the test plan for TC_CCTRL_2_3 * Remove unused import * Add TC_CCTRL_2_3.py to exclude * Address review comments * Remove the CI skip condition
1 parent 5bff3a5 commit e4502c3

File tree

2 files changed

+177
-0
lines changed

2 files changed

+177
-0
lines changed

‎src/python_testing/TC_CCTRL_2_3.py

+176
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
#
2+
# Copyright (c) 2024 Project CHIP Authors
3+
# All rights reserved.
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
#
17+
18+
# See https://github.com/project-chip/connectedhomeip/blob/master/docs/testing/python.md#defining-the-ci-test-arguments
19+
# for details about the block below.
20+
#
21+
# TODO: Skip CI for now, we don't have any way to run this. Needs setup. See test_TC_CCTRL.py
22+
23+
# This test requires a TH_SERVER application. Please specify with --string-arg th_server_app_path:<path_to_app>
24+
25+
import logging
26+
import os
27+
import random
28+
import signal
29+
import subprocess
30+
import time
31+
import uuid
32+
33+
import chip.clusters as Clusters
34+
from chip import ChipDeviceCtrl
35+
from chip.interaction_model import InteractionModelError, Status
36+
from matter_testing_support import (MatterBaseTest, TestStep, async_test_body, default_matter_test_main, has_cluster,
37+
per_endpoint_test)
38+
from mobly import asserts
39+
40+
41+
class TC_CCTRL_2_3(MatterBaseTest):
42+
43+
@async_test_body
44+
async def setup_class(self):
45+
super().setup_class()
46+
self.app_process = None
47+
app = self.user_params.get("th_server_app_path", None)
48+
if not app:
49+
asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
50+
51+
self.kvs = f'kvs_{str(uuid.uuid4())}'
52+
self.port = 5543
53+
discriminator = random.randint(0, 4095)
54+
passcode = 20202021
55+
app_args = f'--secured-device-port {self.port} --discriminator {discriminator} --passcode {passcode} --KVS {self.kvs}'
56+
cmd = f'{app} {app_args}'
57+
# TODO: Determine if we want these logs cooked or pushed to somewhere else
58+
logging.info("Starting TH_SERVER")
59+
self.app_process = subprocess.Popen(cmd, bufsize=0, shell=True)
60+
logging.info("TH_SERVER started")
61+
time.sleep(3)
62+
63+
logging.info("Commissioning from separate fabric")
64+
65+
# Create a second controller on a new fabric to communicate to the server
66+
new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
67+
new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
68+
paa_path = str(self.matter_test_config.paa_trust_store_path)
69+
self.TH_server_controller = new_fabric_admin.NewController(nodeId=112233, paaTrustStorePath=paa_path)
70+
self.server_nodeid = 1111
71+
await self.TH_server_controller.CommissionOnNetwork(nodeId=self.server_nodeid, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
72+
logging.info("Commissioning TH_SERVER complete")
73+
74+
def teardown_class(self):
75+
# In case the th_server_app_path does not exist, then we failed the test
76+
# and there is nothing to remove
77+
if self.app_process is not None:
78+
logging.warning("Stopping app with SIGTERM")
79+
self.app_process.send_signal(signal.SIGTERM.value)
80+
self.app_process.wait()
81+
82+
if os.path.exists(self.kvs):
83+
os.remove(self.kvs)
84+
85+
super().teardown_class()
86+
87+
def steps_TC_CCTRL_2_3(self) -> list[TestStep]:
88+
steps = [TestStep(1, "Get number of fabrics from TH_SERVER", is_commissioning=True),
89+
TestStep(2, "Reading Attribute VendorId from TH_SERVER"),
90+
TestStep(3, "Reading Attribute ProductId from TH_SERVER"),
91+
TestStep(4, "Send RequestCommissioningApproval command to DUT with CASE session with correct VendorId and ProductId"),
92+
TestStep(5, "(Manual Step) Approve Commissioning Approval Request on DUT using method indicated by the manufacturer"),
93+
TestStep(6, "Reading Event CommissioningRequestResult from DUT, confirm one new event"),
94+
TestStep(7, "Send another RequestCommissioningApproval command to DUT with CASE session with same RequestId as the previous one"),
95+
TestStep(8, "Send CommissionNode command to DUT with CASE session, with valid parameters"),
96+
TestStep(9, "Send another CommissionNode command to DUT with CASE session, with with same RequestId as the previous one"),
97+
TestStep(10, "Send OpenCommissioningWindow command on Administrator Commissioning Cluster sent to TH_SERVER"),
98+
TestStep(11, "Wait for DUT to successfully commission TH_SERVER, 30 seconds"),
99+
TestStep(12, "Get number of fabrics from TH_SERVER, verify DUT successfully commissioned TH_SERVER")]
100+
101+
return steps
102+
103+
@per_endpoint_test(has_cluster(Clusters.CommissionerControl))
104+
async def test_TC_CCTRL_2_3(self):
105+
self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
106+
107+
self.step(1)
108+
th_server_fabrics = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False)
109+
110+
self.step(2)
111+
th_server_vid = await self.read_single_attribute_check_success(cluster=Clusters.BasicInformation, attribute=Clusters.BasicInformation.Attributes.VendorID, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
112+
113+
self.step(3)
114+
th_server_pid = await self.read_single_attribute_check_success(cluster=Clusters.BasicInformation, attribute=Clusters.BasicInformation.Attributes.ProductID, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
115+
116+
self.step(4)
117+
good_request_id = 0x1234567812345678
118+
cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(
119+
requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid, label="Test Ecosystem")
120+
await self.send_single_cmd(cmd=cmd)
121+
122+
self.step(5)
123+
if not self.is_ci:
124+
self.wait_for_user_input("Approve Commissioning approval request using manufacturer specified mechanism")
125+
126+
self.step(6)
127+
event_path = [(self.matter_test_config.endpoint, Clusters.CommissionerControl.Events.CommissioningRequestResult, 1)]
128+
events = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=event_path)
129+
asserts.assert_equal(len(events), 1, "Unexpected event list len")
130+
asserts.assert_equal(events[0].Data.statusCode, 0, "Unexpected status code")
131+
asserts.assert_equal(events[0].Data.clientNodeId,
132+
self.matter_test_config.controller_node_id, "Unexpected client node id")
133+
asserts.assert_equal(events[0].Data.requestId, good_request_id, "Unexpected request ID")
134+
135+
self.step(7)
136+
cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(
137+
requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid)
138+
try:
139+
await self.send_single_cmd(cmd=cmd)
140+
asserts.fail("Unexpected success on CommissionNode")
141+
except InteractionModelError as e:
142+
asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
143+
144+
self.step(8)
145+
cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=good_request_id, responseTimeoutSeconds=30)
146+
resp = await self.send_single_cmd(cmd)
147+
asserts.assert_equal(type(resp), Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow,
148+
"Incorrect response type")
149+
150+
self.step(9)
151+
cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=good_request_id, responseTimeoutSeconds=30)
152+
try:
153+
await self.send_single_cmd(cmd=cmd)
154+
asserts.fail("Unexpected success on CommissionNode")
155+
except InteractionModelError as e:
156+
asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
157+
158+
self.step(10)
159+
# min commissioning timeout is 3*60 seconds, so use that even though the command said 30.
160+
cmd = Clusters.AdministratorCommissioning.Commands.OpenCommissioningWindow(commissioningTimeout=3*60,
161+
PAKEPasscodeVerifier=resp.PAKEPasscodeVerifier,
162+
discriminator=resp.discriminator,
163+
iterations=resp.iterations, salt=resp.salt)
164+
await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000)
165+
166+
self.step(11)
167+
time.sleep(30)
168+
169+
self.step(12)
170+
th_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, fabric_filtered=False)
171+
asserts.assert_equal(len(th_server_fabrics) + 1, len(th_server_fabrics_new),
172+
"Unexpected number of fabrics on TH_SERVER")
173+
174+
175+
if __name__ == "__main__":
176+
default_matter_test_main()

‎src/python_testing/execute_python_tests.py

+1
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ def main(search_directory, env_file):
5959
"TC_CNET_4_4.py",
6060
"TC_CCTRL_2_1.py",
6161
"TC_CCTRL_2_2.py",
62+
"TC_CCTRL_2_3.py",
6263
"TC_DGGEN_3_2.py",
6364
"TC_EEVSE_Utils.py",
6465
"TC_ECOINFO_2_1.py",

0 commit comments

Comments
 (0)