20
20
21
21
import chip .clusters as Clusters
22
22
from matter_testing_support import MatterBaseTest , async_test_body , default_matter_test_main
23
- from mobly import asserts
24
23
from mdns_discovery .mdns_discovery import MdnsDiscovery , MdnsServiceType
25
24
25
+ from mobly import asserts
26
26
27
27
'''
28
28
Category
36
36
https://github.com/CHIP-Specifications/chip-test-plans/blob/master/src/securechannel.adoc#343-tc-sc-43-discovery-dut_commissionee
37
37
'''
38
38
39
+
39
40
class TC_SC_4_3 (MatterBaseTest ):
40
41
41
42
ONE_HOUR_IN_MS = 3600000
@@ -49,29 +50,34 @@ async def get_descriptor_server_list(self):
49
50
cluster = Clusters .Descriptor ,
50
51
attribute = Clusters .Descriptor .Attributes .ServerList
51
52
)
53
+
52
54
async def get_idle_mode_threshhold_ms (self ):
53
55
return await self .read_single_attribute_check_success (
54
56
endpoint = 0 ,
55
57
dev_ctrl = self .default_controller ,
56
58
cluster = Clusters .IcdManagement ,
57
59
attribute = Clusters .IcdManagement .Attributes .ActiveModeThreshold
58
60
)
61
+
59
62
async def get_icd_feature_map (self ):
60
63
return await self .read_single_attribute_check_success (
61
64
endpoint = 0 ,
62
65
dev_ctrl = self .default_controller ,
63
66
cluster = Clusters .IcdManagement ,
64
67
attribute = Clusters .IcdManagement .Attributes .FeatureMap
65
68
)
69
+
66
70
def get_dut_instance_name (self ) -> str :
67
71
node_id = self .dut_node_id
68
72
compressed_fabric_id = self .default_controller .GetCompressedFabricId ()
69
73
instance_name = f'{ compressed_fabric_id :016X} -{ node_id :016X} '
70
74
return instance_name
75
+
71
76
def get_operational_subtype (self ) -> str :
72
77
compressed_fabric_id = self .default_controller .GetCompressedFabricId ()
73
78
service_name = f'_I{ compressed_fabric_id :016X} ._sub.{ MdnsServiceType .OPERATIONAL .value } '
74
79
return service_name
80
+
75
81
@staticmethod
76
82
def verify_decimal_value (input_value , comparison_value : int ):
77
83
try :
@@ -90,6 +96,7 @@ def verify_decimal_value(input_value, comparison_value: int):
90
96
return (False , f"Input ({ input_value } ) exceeds the allowed value { comparison_value } ." )
91
97
except ValueError :
92
98
return (False , f"Input ({ input_value } ) is not a valid decimal number." )
99
+
93
100
def verify_t_value (self , t_value ):
94
101
# Verify t_value is a decimal number without leading zeros and less than or equal to 6
95
102
try :
@@ -100,14 +107,15 @@ def verify_t_value(self, t_value):
100
107
return False , f"T value ({ t_value } ) has leading zeros."
101
108
if T_int != float (t_value ):
102
109
return False , f"T value ({ t_value } ) is not an integer."
103
-
110
+
104
111
# Convert to bitmap and verify bit 0 is clear
105
112
if T_int & 1 == 0 :
106
113
return True , f"T value ({ t_value } ) is valid and bit 0 is clear."
107
114
else :
108
115
return False , f"Bit 0 is not clear. T value ({ t_value } )"
109
116
except ValueError :
110
117
return False , "T value ({t_value}) is not a valid decimal number."
118
+
111
119
@staticmethod
112
120
def contains_ipv6_address (addresses ):
113
121
# IPv6 pattern for basic validation
@@ -118,7 +126,7 @@ def contains_ipv6_address(addresses):
118
126
return True , "At least one IPv6 address is present."
119
127
120
128
return False , "No IPv6 addresses found."
121
-
129
+
122
130
@async_test_body
123
131
async def test_TC_SC_4_3 (self ):
124
132
print (f"\n " * 10 )
@@ -132,7 +140,7 @@ async def test_TC_SC_4_3(self):
132
140
133
141
# *** STEP 1 ***
134
142
self .print_step ("1" , "DUT is commissioned on the same fabric as TH." )
135
-
143
+
136
144
# *** STEP 2 ***
137
145
self .print_step ("2" , "TH reads ServerList attribute from the Descriptor cluster on EP0. If the ICD Management cluster ID (70,0x46) is present in the list, set supports_icd to true, otherwise set supports_icd to false." )
138
146
ep0_servers = await self .get_descriptor_server_list ()
@@ -142,7 +150,8 @@ async def test_TC_SC_4_3(self):
142
150
logging .info (f"\n \n \t supports_icd: { supports_icd } \n \n " )
143
151
144
152
# *** STEP 3 ***
145
- self .print_step ("3" , "If supports_icd is true, TH reads ActiveModeThreshold from the ICD Management cluster on EP0 and saves as active_mode_threshold." )
153
+ self .print_step (
154
+ "3" , "If supports_icd is true, TH reads ActiveModeThreshold from the ICD Management cluster on EP0 and saves as active_mode_threshold." )
146
155
if supports_icd :
147
156
active_mode_threshold_ms = await self .get_idle_mode_threshhold_ms ()
148
157
logging .info (f"\n \n \t active_mode_threshold_ms: { active_mode_threshold_ms } \n \n " )
@@ -159,11 +168,8 @@ async def test_TC_SC_4_3(self):
159
168
self .print_step ("5" , "TH constructs the instance name for the DUT as the 64-bit compressed Fabric identifier, and the assigned 64-bit Node identifier, each expressed as a fixed-length sixteen-character hexadecimal string, encoded as ASCII (UTF-8) text using capital letters, separated by a hyphen." )
160
169
instance_name = self .get_dut_instance_name ()
161
170
162
-
163
-
164
171
# PENDING STEPS 6-8
165
172
166
-
167
173
mdns = MdnsDiscovery ()
168
174
operational = await mdns .get_operational_service (
169
175
service_name = f"{ instance_name } .{ MdnsServiceType .OPERATIONAL .value } " ,
@@ -222,7 +228,8 @@ async def test_TC_SC_4_3(self):
222
228
223
229
# SAT TXT KEY
224
230
if 'SAT' in operational .txt_record :
225
- logging .info (f"SAT key is present in TXT record, verify that it is a decimal value with no leading zeros and is less than or equal to 65535." )
231
+ logging .info (
232
+ f"SAT key is present in TXT record, verify that it is a decimal value with no leading zeros and is less than or equal to 65535." )
226
233
sat_value = operational .txt_record ['SAT' ]
227
234
result , message = self .verify_decimal_value (sat_value , self .MAX_SAT_VALUE )
228
235
asserts .assert_true (result , message )
0 commit comments