Skip to content

Commit f2f3d0e

Browse files
tehampsontehampson
authored
Add TC_MCORE_FS_1_3.py test implementation (project-chip#34650)
1 parent 1a7d048 commit f2f3d0e

File tree

1 file changed

+175
-0
lines changed

1 file changed

+175
-0
lines changed

‎src/python_testing/TC_MCORE_FS_1_3.py

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
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+
# This test requires a TH_SERVER application that returns UnsupportedAttribute when reading UniqueID from BasicInformation Cluster. Please specify with --string-arg th_server_app_path:<path_to_app>
19+
20+
import logging
21+
import os
22+
import random
23+
import signal
24+
import subprocess
25+
import time
26+
import uuid
27+
28+
import chip.clusters as Clusters
29+
from chip import ChipDeviceCtrl
30+
from chip.interaction_model import Status
31+
from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, type_matches
32+
from mobly import asserts
33+
34+
35+
class TC_MCORE_FS_1_3(MatterBaseTest):
36+
@async_test_body
37+
async def setup_class(self):
38+
super().setup_class()
39+
self.device_for_th_eco_nodeid = 1111
40+
self.device_for_th_eco_kvs = None
41+
self.device_for_th_eco_port = 5543
42+
self.app_process_for_th_eco = None
43+
44+
self.device_for_dut_eco_nodeid = 1112
45+
self.device_for_dut_eco_kvs = None
46+
self.device_for_dut_eco_port = 5544
47+
self.app_process_for_dut_eco = None
48+
49+
# Create a second controller on a new fabric to communicate to the server
50+
new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
51+
new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
52+
paa_path = str(self.matter_test_config.paa_trust_store_path)
53+
self.TH_server_controller = new_fabric_admin.NewController(nodeId=112233, paaTrustStorePath=paa_path)
54+
55+
def teardown_class(self):
56+
if self.app_process_for_dut_eco is not None:
57+
logging.warning("Stopping app with SIGTERM")
58+
self.app_process_for_dut_eco.send_signal(signal.SIGTERM.value)
59+
self.app_process_for_dut_eco.wait()
60+
if self.app_process_for_th_eco is not None:
61+
logging.warning("Stopping app with SIGTERM")
62+
self.app_process_for_th_eco.send_signal(signal.SIGTERM.value)
63+
self.app_process_for_th_eco.wait()
64+
65+
os.remove(self.device_for_dut_eco_kvs)
66+
if self.device_for_th_eco_kvs is not None:
67+
os.remove(self.device_for_th_eco_kvs)
68+
super().teardown_class()
69+
70+
async def create_device_and_commission_to_th_fabric(self, kvs, port, node_id_for_th, device_info):
71+
# TODO: confirm whether we can open processes like this on the TH
72+
app = self.user_params.get("th_server_app_path", None)
73+
if not app:
74+
asserts.fail('This test requires a TH_SERVER app. Specify app path with --string-arg th_server_app_path:<path_to_app>')
75+
76+
discriminator = random.randint(0, 4095)
77+
passcode = 20202021
78+
app_args = f'--secured-device-port {port} --discriminator {discriminator} --passcode {passcode} --KVS {kvs}'
79+
cmd = f'{app} {app_args}'
80+
# TODO: Determine if we want these logs cooked or pushed to somewhere else
81+
logging.info(f"Starting TH device for {device_info}")
82+
self.app_process_for_dut_eco = subprocess.Popen(cmd, bufsize=0, shell=True)
83+
logging.info(f"Started TH device for {device_info}")
84+
time.sleep(3)
85+
86+
logging.info("Commissioning from separate fabric")
87+
await self.TH_server_controller.CommissionOnNetwork(nodeId=node_id_for_th, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
88+
logging.info("Commissioning device for DUT ecosystem onto TH for managing")
89+
90+
async def create_and_commission_device_for_th_ecosystem(self):
91+
# TODO: confirm whether we can open processes like this on the TH
92+
app = self.user_params.get("th_server_app_path", None)
93+
94+
self.device_for_th_eco_kvs = f'kvs_{str(uuid.uuid4())}'
95+
discriminator = random.randint(0, 4095)
96+
passcode = 20202021
97+
app_args = f'--secured-device-port {self.device_for_th_eco_port} --discriminator {discriminator} --passcode {passcode} --KVS {self.device_for_th_eco_kvs}'
98+
cmd = f'{app} {app_args}'
99+
# TODO: Determine if we want these logs cooked or pushed to somewhere else
100+
logging.info("Starting TH device for TH ecosystem")
101+
self.app_process_for_th_eco = subprocess.Popen(cmd, bufsize=0, shell=True)
102+
logging.info("Started TH device for TH ecosystem")
103+
time.sleep(3)
104+
105+
logging.info("Commissioning from separate fabric")
106+
self.server_nodeid = 1112
107+
await self.TH_server_controller.CommissionOnNetwork(nodeId=self.server_nodeid, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
108+
logging.info("Commissioning TH device for TH ecosystem")
109+
110+
def steps_TC_MCORE_FS_1_3(self) -> list[TestStep]:
111+
steps = [TestStep(1, "DUT_FSA commissions TH_SED_DUT to DUT_FSAs fabric and generates a UniqueID", is_commissioning=True),
112+
TestStep(2, "TH_FSA commissions TH_SED_TH onto TH_FSAs fabric and generates a UniqueID."),
113+
TestStep(3, "Follow manufacturer provided instructions to enable DUT_FSA to synchronize TH_SED_TH onto DUT_FSAs fabric."),
114+
TestStep(4, "DUT_FSA synchronizes TH_SED_TH onto DUT_FSAs fabric and copies the UniqueID presented by TH_FSAs Bridged Device Basic Information Cluster.")]
115+
return steps
116+
117+
@async_test_body
118+
async def test_TC_MCORE_FS_1_3(self):
119+
self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
120+
self.print_step(0, "Commissioning DUT to TH, already done")
121+
self.step(1)
122+
# These steps are not explicitly in step 1, but they help identify the dynamically added endpoint in step 1.
123+
root_node_endpoint = 0
124+
root_part_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=Clusters.Descriptor.Attributes.PartsList, endpoint=root_node_endpoint)
125+
set_of_endpoints_before_adding_device = set(root_part_list)
126+
127+
kvs = f'kvs_{str(uuid.uuid4())}'
128+
device_info = "for DUT ecosystem"
129+
await self.create_device_and_commission_to_th_fabric(kvs, self.device_for_dut_eco_port, self.device_for_dut_eco_nodeid, device_info)
130+
self.device_for_dut_eco_kvs = kvs
131+
read_result = await self.TH_server_controller.ReadAttribute(self.device_for_dut_eco_nodeid, [(root_node_endpoint, Clusters.BasicInformation.Attributes.UniqueID)])
132+
result = read_result[root_node_endpoint][Clusters.BasicInformation][Clusters.BasicInformation.Attributes.UniqueID]
133+
asserts.assert_true(type_matches(result, Clusters.Attribute.ValueDecodeFailure), "We were expecting a value decode failure")
134+
asserts.assert_equal(result.Reason.status, Status.UnsupportedAttribute, "Incorrect error returned from reading UniqueID")
135+
136+
params = await self.openCommissioningWindow(dev_ctrl=self.TH_server_controller, node_id=self.device_for_dut_eco_nodeid)
137+
138+
self.wait_for_user_input(
139+
prompt_msg=f"Using the DUT vendor's provided interface, commission the device using the following parameters:\n"
140+
f"- discriminator: {params.randomDiscriminator}\n"
141+
f"- setupPinCode: {params.commissioningParameters.setupPinCode}\n"
142+
f"- setupQRCode: {params.commissioningParameters.setupQRCode}\n"
143+
f"- setupManualcode: {params.commissioningParameters.setupManualCode}\n"
144+
f"If using FabricSync Admin, you may type:\n"
145+
f">>> pairing onnetwork <desired_node_id> {params.commissioningParameters.setupPinCode}")
146+
147+
root_part_list = await self.read_single_attribute_check_success(cluster=Clusters.Descriptor, attribute=Clusters.Descriptor.Attributes.PartsList, endpoint=root_node_endpoint)
148+
set_of_endpoints_after_adding_device = set(root_part_list)
149+
150+
asserts.assert_true(set_of_endpoints_after_adding_device.issuperset(
151+
set_of_endpoints_before_adding_device), "Expected only new endpoints to be added")
152+
unique_endpoints_set = set_of_endpoints_after_adding_device - set_of_endpoints_before_adding_device
153+
asserts.assert_equal(len(unique_endpoints_set), 1, "Expected only one new endpoint")
154+
newly_added_endpoint = list(unique_endpoints_set)[0]
155+
156+
th_sed_dut_unique_id = await self.read_single_attribute_check_success(cluster=Clusters.BridgedDeviceBasicInformation, attribute=Clusters.BridgedDeviceBasicInformation.Attributes.UniqueID, endpoint=newly_added_endpoint)
157+
asserts.assert_true(type_matches(th_sed_dut_unique_id, str), "UniqueID should be a string")
158+
asserts.assert_true(th_sed_dut_unique_id, "UniqueID should not be an empty string")
159+
160+
self.step(2)
161+
kvs = f'kvs_{str(uuid.uuid4())}'
162+
device_info = "for TH_FSA ecosystem"
163+
await self.create_device_and_commission_to_th_fabric(kvs, self.device_for_th_eco_port, self.device_for_th_eco_nodeid, device_info)
164+
self.device_for_th_eco_kvs = kvs
165+
# TODO(https://github.com/CHIP-Specifications/chip-test-plans/issues/4375) During setup we need to create the TH_FSA device
166+
# where we would commission device created in create_device_and_commission_to_th_fabric to be commissioned into TH_FSA.
167+
168+
# TODO(https://github.com/CHIP-Specifications/chip-test-plans/issues/4375) Because we cannot create a TH_FSA and there is
169+
# no way to mock it the following 2 test steps are skipped for now.
170+
self.skip_step(3)
171+
self.skip_step(4)
172+
173+
174+
if __name__ == "__main__":
175+
default_matter_test_main()

0 commit comments

Comments
 (0)