TC-TMP-2.1: Fix (project-chip#32881)

* TC-TMP-2.1: Fix

The constraints here are actually relatively complex because the
min and max can be null and fall back to default values.

Updated the automation to python so we can do the appropriate tests,
added tests for the various scenarios becuase there are a LOT and
we don't cover any of them with our current examples.

Please see
CHIP-Specifications/chip-test-plans#4096
for test plan updates

* remove yaml

* missed a change

* Restyled by autopep8

* remove unused import

* Apply suggestions from code review

---------

Co-authored-by: Restyled.io <commits@restyled.io>

Author: cecille

src/app/tests/suites/certification/Test_TC_TMP_2_1.yaml

@@ -0,0 +1,97 @@
+#
+#    Copyright (c) 2024 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.
+
+import chip.clusters as Clusters
+from chip.clusters.Types import NullValue
+from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from mobly import asserts
+
+
+class TC_TMP_2_1(MatterBaseTest):
+    def desc_TC_TMP_2_1(self) -> str:
+        return "[TC-TMP-2.1] Attributes with Server as DUT"
+
+    def pics_TC_TMP_2_1(self):
+        return ["TMP.S"]
+
+    def steps_TC_TMP_2_1(self) -> list[TestStep]:
+        return [
+            TestStep(1, "Commissioning, already done", is_commissioning=True),
+            TestStep(
+                2, "Set default bounds `min_bound` = -27315, `max_bound` = 32767"),
+            TestStep(3, "TH reads the MinMeasuredValue attribute from the DUT and saves as `min_measured_value`. If `min_measured_value` is not null, set `min_bound` to `min_measured_value`"),
+            TestStep(4, "TH reads the MaxMeasuredValue attribute from the DUT and saves as `max_measured_value`. If `max_measured_value` is not null, set `max_bound` to `max_measured_value",
+                     "Verify that `max_measured_value` is either null or an int16 where min_bound < `max_measured_value` ≤ 32767."),
+            TestStep(5, "If `min_measured_value` is not null, verify min measured value range",
+                     "Verify that -27315 ≤ `min_measured_value` < `max_bound`"),
+            TestStep(6, "TH reads the MeasuredValue attribute from the DUT",
+                     "Verify that the DUT response contains either null or a int16 where `min_bound` ≤ MeasuredValue ≤ `max_bound`."),
+            TestStep(7, "TH reads the Tolerance attribute from the DUT",
+                     "Verify that Tolerance is in the range of 0 to 2048"),
+        ]
+
+    @async_test_body
+    async def test_TC_TMP_2_1(self):
+        cluster = Clusters.TemperatureMeasurement
+        attr = Clusters.TemperatureMeasurement.Attributes
+
+        # Commission DUT - already done
+        self.step(1)
+
+        self.step(2)
+        min_bound = -27315
+        max_bound = 32767
+
+        self.step(3)
+        min_measured_value = await self.read_single_attribute_check_success(cluster=cluster, attribute=attr.MinMeasuredValue)
+        if min_measured_value != NullValue:
+            min_bound = min_measured_value
+
+        self.step(4)
+        max_measured_value = await self.read_single_attribute_check_success(cluster=cluster, attribute=attr.MaxMeasuredValue)
+        if max_measured_value != NullValue:
+            max_bound = max_measured_value
+            asserts.assert_greater(max_measured_value, min_bound,
+                                   "MaxMeasuredValue is not greater than the minimum bound")
+            asserts.assert_less_equal(
+                max_measured_value, 32767, "MaxMeasuredValue is not less than or equal to than 32767")
+
+        self.step(5)
+        if min_measured_value != NullValue:
+            asserts.assert_greater_equal(min_measured_value, -27315, "MinMeasuredValue is out of range")
+            asserts.assert_less(min_measured_value, max_bound, "MinMeasuredValue is out of range")
+        else:
+            self.mark_current_step_skipped()
+
+        self.step(6)
+        measured_value = await self.read_single_attribute_check_success(cluster=cluster, attribute=attr.MeasuredValue)
+        if measured_value != NullValue:
+            print(measured_value)
+            print(min_bound)
+            asserts.assert_greater_equal(
+                measured_value, min_bound, "Measured value is less than min bound")
+            asserts.assert_less_equal(
+                measured_value, max_bound, "Measured value is greater than max bound")
+
+        self.step(7)
+        tolerance = await self.read_single_attribute_check_success(cluster=cluster, attribute=attr.Tolerance)
+        asserts.assert_greater_equal(tolerance, 0, "Tolerance is less than 0")
+        asserts.assert_less_equal(
+            tolerance, 2048, "Tolerance is greater than 2048")
+
+
+if __name__ == "__main__":
+    default_matter_test_main()

src/python_testing/TC_TMP_2_1.py

@@ -1656,7 +1656,7 @@ def get_test_info(test_class: MatterBaseTest, matter_test_config: MatterTestConf
     return info
 
 
-def run_tests(test_class: MatterBaseTest, matter_test_config: MatterTestConfig, hooks: TestRunnerHooks) -> None:
+def run_tests_no_exit(test_class: MatterBaseTest, matter_test_config: MatterTestConfig, hooks: TestRunnerHooks, default_controller=None, external_stack=None) -> bool:
 
     get_test_info(test_class, matter_test_config)
 
@@ -1668,7 +1668,10 @@ def run_tests(test_class: MatterBaseTest, matter_test_config: MatterTestConfig,
     if len(matter_test_config.tests) > 0:
         tests = matter_test_config.tests
 
-    stack = MatterStackState(matter_test_config)
+    if external_stack:
+        stack = external_stack
+    else:
+        stack = MatterStackState(matter_test_config)
 
     with TracingContext() as tracing_ctx:
         for destination in matter_test_config.trace_to:
@@ -1678,12 +1681,12 @@ def run_tests(test_class: MatterBaseTest, matter_test_config: MatterTestConfig,
 
         # TODO: Steer to right FabricAdmin!
         # TODO: If CASE Admin Subject is a CAT tag range, then make sure to issue NOC with that CAT tag
-
-        default_controller = stack.certificate_authorities[0].adminList[0].NewController(
-            nodeId=matter_test_config.controller_node_id,
-            paaTrustStorePath=str(matter_test_config.paa_trust_store_path),
-            catTags=matter_test_config.controller_cat_tags
-        )
+        if not default_controller:
+            default_controller = stack.certificate_authorities[0].adminList[0].NewController(
+                nodeId=matter_test_config.controller_node_id,
+                paaTrustStorePath=str(matter_test_config.paa_trust_store_path),
+                catTags=matter_test_config.controller_cat_tags
+            )
         test_config.user_params["default_controller"] = stash_globally(default_controller)
 
         test_config.user_params["matter_test_config"] = stash_globally(matter_test_config)
@@ -1732,10 +1735,16 @@ def run_tests(test_class: MatterBaseTest, matter_test_config: MatterTestConfig,
         hooks.stop(duration=duration)
 
     # Shutdown the stack when all done
-    stack.Shutdown()
+    if not external_stack:
+        stack.Shutdown()
 
     if ok:
         logging.info("Final result: PASS !")
     else:
         logging.error("Final result: FAIL !")
+    return ok
+
+
+def run_tests(test_class: MatterBaseTest, matter_test_config: MatterTestConfig, hooks: TestRunnerHooks, default_controller=None, external_stack=None) -> None:
+    if not run_tests_no_exit(test_class, matter_test_config, hooks, default_controller, external_stack):
         sys.exit(1)

src/python_testing/matter_testing_support.py

@@ -0,0 +1,57 @@
+#
+#    Copyright (c) 2024 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.
+#
+
+import importlib
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from chip.clusters import Attribute
+
+try:
+    from matter_testing_support import MatterStackState, MatterTestConfig, run_tests_no_exit
+except ImportError:
+    sys.path.append(os.path.abspath(
+        os.path.join(os.path.dirname(__file__), '..')))
+    from matter_testing_support import MatterStackState, MatterTestConfig, run_tests_no_exit
+
+
+class AsyncMock(MagicMock):
+    async def __call__(self, *args, **kwargs):
+        return super(AsyncMock, self).__call__(*args, **kwargs)
+
+
+class MockTestRunner():
+    def __init__(self, filename: str, classname: str, test: str):
+        self.config = MatterTestConfig(
+            tests=[test], endpoint=1, dut_node_ids=[1])
+        self.stack = MatterStackState(self.config)
+        self.default_controller = self.stack.certificate_authorities[0].adminList[0].NewController(
+            nodeId=self.config.controller_node_id,
+            paaTrustStorePath=str(self.config.paa_trust_store_path),
+            catTags=self.config.controller_cat_tags
+        )
+        module = importlib.import_module(Path(os.path.basename(filename)).stem)
+        self.test_class = getattr(module, classname)
+
+    def Shutdown(self):
+        self.stack.Shutdown()
+
+    def run_test_with_mock_read(self,  read_cache: Attribute.AsyncReadTransaction.ReadResponse):
+        self.default_controller.Read = AsyncMock(return_value=read_cache)
+        return run_tests_no_exit(self.test_class, self.config, None, self.default_controller, self.stack)