Skip to content

Commit dedef1f

Browse files
yufengwangcaandy31415
yufengwangca
and
andy31415
authored
Implement python test for TC_DGSW_2_3 (project-chip#36918)
* Implement python test for TC_DGSW_2_3 * Update src/python_testing/TC_DGSW_2_3.py Co-authored-by: Andrei Litvin <andy314@gmail.com> * Update src/python_testing/TC_DGSW_2_3.py Co-authored-by: Andrei Litvin <andy314@gmail.com> * Update src/python_testing/TC_DGSW_2_3.py Co-authored-by: Andrei Litvin <andy314@gmail.com> * Update src/python_testing/TC_DGSW_2_3.py Co-authored-by: Andrei Litvin <andy314@gmail.com> * Update src/python_testing/TC_DGSW_2_3.py Co-authored-by: Andrei Litvin <andy314@gmail.com> * Address review comments * Use is_valid_str_value to check str * Add the missing changes * Fix lint error --------- Co-authored-by: Andrei Litvin <andy314@gmail.com>
1 parent 36eddf5 commit dedef1f

File tree

1 file changed

+209
-0
lines changed

1 file changed

+209
-0
lines changed

‎src/python_testing/TC_DGSW_2_3.py

+209
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
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+
# === BEGIN CI TEST ARGUMENTS ===
22+
# test-runner-runs:
23+
# run1:
24+
# app: ${ALL_CLUSTERS_APP}
25+
# app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
26+
# script-args: >
27+
# --storage-path admin_storage.json
28+
# --commissioning-method on-network
29+
# --discriminator 1234
30+
# --passcode 20202021
31+
# --trace-to json:${TRACE_TEST_JSON}.json
32+
# --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
33+
# factory-reset: true
34+
# quiet: true
35+
# === END CI TEST ARGUMENTS ===
36+
#
37+
38+
import logging
39+
40+
import chip.clusters as Clusters
41+
from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
42+
from mobly import asserts
43+
44+
45+
class TC_DGSW_2_3(MatterBaseTest):
46+
47+
@staticmethod
48+
def is_valid_uint64_value(value):
49+
return isinstance(value, int) and 0 <= value <= 0xFFFFFFFFFFFFFFFF
50+
51+
@staticmethod
52+
def is_valid_uint32_value(value):
53+
return isinstance(value, int) and 0 <= value <= 0xFFFFFFFF
54+
55+
@staticmethod
56+
def is_valid_str_value(value):
57+
return isinstance(value, str) and len(value) > 0
58+
59+
async def read_dgsw_attribute_expect_success(self, endpoint, attribute):
60+
cluster = Clusters.Objects.SoftwareDiagnostics
61+
return await self.read_single_attribute_check_success(endpoint=endpoint, cluster=cluster, attribute=attribute)
62+
63+
async def send_reset_watermarks_command(self):
64+
endpoint = 0
65+
diags_cluster = Clusters.Objects.SoftwareDiagnostics
66+
return await self.send_single_cmd(cmd=diags_cluster.Commands.ResetWatermarks(), endpoint=endpoint)
67+
68+
def desc_TC_DGSW_2_3(self) -> str:
69+
"""Returns a description of this test"""
70+
return "[TC-DGSW-2.1] Attributes with Server as DUT"
71+
72+
def pics_TC_DGSW_2_3(self) -> list[str]:
73+
return ["DGSW.S"]
74+
75+
def steps_TC_DGSW_2_3(self) -> list[TestStep]:
76+
steps = [
77+
TestStep(1, "Commissioning, already done", is_commissioning=True),
78+
TestStep(2, "Read the AttributeList attribute"),
79+
TestStep(3, "Read the ThreadMetrics attribute"),
80+
TestStep(4, "Read the CurrentHeapHighWatermark attribute"),
81+
TestStep(5, "Read the CurrentHeapUsed attribute"),
82+
TestStep(6, "Send ResetWatermarks command"),
83+
TestStep(7, "Read the CurrentHeapHighWatermark attribute"),
84+
TestStep(8, "Read the ThreadMetrics attribute"),
85+
]
86+
return steps
87+
88+
@async_test_body
89+
async def test_TC_DGSW_2_3(self):
90+
91+
endpoint = self.get_endpoint(default=0)
92+
93+
# STEP 1: Commission DUT (already done)
94+
self.step(1)
95+
96+
# STEP 2: TH reads from the DUT the AttributeList attribute
97+
self.step(2)
98+
attributes = Clusters.SoftwareDiagnostics.Attributes
99+
attribute_list = await self.read_dgsw_attribute_expect_success(endpoint=endpoint, attribute=attributes.AttributeList)
100+
101+
# STEP 3: TH reads from the DUT the ThreadMetrics attribute
102+
self.step(3)
103+
if self.pics_guard(attributes.ThreadMetrics.attribute_id in attribute_list):
104+
thread_metrics_original = await self.read_dgsw_attribute_expect_success(endpoint=endpoint, attribute=attributes.ThreadMetrics)
105+
106+
# Iterate over all items in the list and validate each one
107+
for metric in thread_metrics_original:
108+
# The Id field is mandatory
109+
asserts.assert_true(self.is_valid_uint64_value(metric.id), "Id field should be a uint64 type")
110+
111+
if metric.name is not None:
112+
asserts.assert_true(self.is_valid_str_value(metric.name), "Name field should be a string type")
113+
114+
if metric.stackFreeCurrent is not None:
115+
asserts.assert_true(self.is_valid_uint32_value(metric.stackFreeCurrent),
116+
"StackFreeCurrent field should be a uint32 type")
117+
118+
if metric.stackFreeMinimum is not None:
119+
asserts.assert_true(self.is_valid_uint32_value(metric.stackFreeMinimum),
120+
"StackFreeMinimum field should be a uint32 type")
121+
122+
if metric.stackSize is not None:
123+
asserts.assert_true(self.is_valid_uint32_value(metric.stackSize), "StackSize field should be a uint32 type")
124+
125+
# STEP 4: TH reads from the DUT the CurrentHeapHighWatermark attribute
126+
self.step(4)
127+
if self.pics_guard(attributes.CurrentHeapHighWatermark.attribute_id in attribute_list):
128+
high_watermark_original = await self.read_dgsw_attribute_expect_success(endpoint=endpoint, attribute=attributes.CurrentHeapHighWatermark)
129+
asserts.assert_true(self.is_valid_uint64_value(high_watermark_original),
130+
"CurrentHeapHighWatermark field should be a uint64 type")
131+
132+
# STEP 5: TH reads from the DUT the CurrentHeapUsed attribute
133+
self.step(5)
134+
if self.pics_guard(attributes.CurrentHeapUsed.attribute_id in attribute_list):
135+
current_heap_used_original = await self.read_dgsw_attribute_expect_success(endpoint=endpoint, attribute=attributes.CurrentHeapUsed)
136+
asserts.assert_true(self.is_valid_uint64_value(current_heap_used_original),
137+
"CurrentHeapUsed field should be a uint64 type")
138+
if high_watermark_original is not None:
139+
asserts.assert_true(current_heap_used_original <= high_watermark_original,
140+
"CurrentHeapUsed should be less than or equal to CurrentHeapHighWatermark")
141+
142+
# STEP 6: TH sends to the DUT the command
143+
self.step(6)
144+
response = await self.send_reset_watermarks_command()
145+
logging.info(f"ResetWatermarks response: {response}")
146+
147+
# STEP 7: TH reads from the DUT the CurrentHeapHighWatermark attribute
148+
self.step(7)
149+
if self.pics_guard(attributes.CurrentHeapHighWatermark.attribute_id in attribute_list):
150+
current_heap_high_watermark = await self.read_dgsw_attribute_expect_success(endpoint=endpoint, attribute=attributes.CurrentHeapHighWatermark)
151+
asserts.assert_true(self.is_valid_uint64_value(current_heap_high_watermark),
152+
"CurrentHeapHighWatermark field should be a uint64 type")
153+
154+
# Verify that the returned value is <= high_watermark_original
155+
asserts.assert_true(current_heap_high_watermark <= high_watermark_original,
156+
f"CurrentHeapHighWatermark ({current_heap_high_watermark}) should be <= high_watermark_original ({high_watermark_original})")
157+
158+
# If CurrentHeapUsed is supported
159+
if current_heap_used_original is not None and current_heap_used_original < high_watermark_original:
160+
# Verify that the returned value is < high_watermark_original
161+
asserts.assert_true(current_heap_high_watermark < high_watermark_original,
162+
f"CurrentHeapHighWatermark ({current_heap_high_watermark}) should be < high_watermark_original ({high_watermark_original}) when CurrentHeapUsed ({current_heap_used_original}) is less than high_watermark_original")
163+
164+
# STEP 8: TH reads from the DUT the ThreadMetrics attribute
165+
self.step(8)
166+
if self.pics_guard(attributes.ThreadMetrics.attribute_id in attribute_list):
167+
thread_metrics_reset = await self.read_dgsw_attribute_expect_success(endpoint=endpoint, attribute=attributes.ThreadMetrics)
168+
169+
# Validate all elements in the list
170+
for metric in thread_metrics_reset:
171+
# The Id field is mandatory
172+
asserts.assert_true(self.is_valid_uint64_value(metric.id), "Id field should be a uint64 type")
173+
174+
if metric.name is not None:
175+
asserts.assert_true(self.is_valid_str_value(metric.name), "Name field should be a string type")
176+
177+
if metric.stackFreeCurrent is not None:
178+
asserts.assert_true(self.is_valid_uint32_value(metric.stackFreeCurrent),
179+
"StackFreeCurrent field should be a uint32 type")
180+
181+
if metric.stackFreeMinimum is not None:
182+
asserts.assert_true(self.is_valid_uint32_value(metric.stackFreeMinimum),
183+
"StackFreeMinimum field should be a uint32 type")
184+
185+
if metric.stackSize is not None:
186+
asserts.assert_true(self.is_valid_uint32_value(metric.stackSize), "StackSize field should be a uint32 type")
187+
188+
# Ensure the list length matches thread_metrics_original to simplify matching
189+
asserts.assert_equal(len(thread_metrics_reset), len(thread_metrics_original),
190+
"Mismatch in the number of items between thread_metrics_reset and thread_metrics_original")
191+
192+
# Compare each item in the list
193+
for reset_metric in thread_metrics_reset:
194+
# Find the corresponding item in thread_metrics_original by ID
195+
original_metric = next((m for m in thread_metrics_original if m.id == reset_metric.id), None)
196+
asserts.assert_is_not_none(original_metric,
197+
f"No matching ThreadMetrics entry found in thread_metrics_original for ID {reset_metric.id}")
198+
199+
# Compare StackFreeMinimum if present in both
200+
if original_metric.stackFreeMinimum is not None and reset_metric.stackFreeMinimum is not None:
201+
asserts.assert_true(
202+
reset_metric.stackFreeMinimum >= original_metric.stackFreeMinimum,
203+
f"StackFreeMinimum for ID {reset_metric.id} in thread_metrics_reset ({reset_metric.stackFreeMinimum}) "
204+
f"should be >= corresponding value in thread_metrics_original ({original_metric.stackFreeMinimum})"
205+
)
206+
207+
208+
if __name__ == "__main__":
209+
default_matter_test_main()

0 commit comments

Comments
 (0)