Add test steps to scripts TC_AVSM_2_3.py and TC_AVSM_2_8.py to subscribe

to attributes and verify on report generation.

Author: pidarped

src/python_testing/TC_AVSMTestBase.py

@@ -22,10 +22,31 @@
 import matter.clusters as Clusters
 from matter.clusters import Globals
 from matter.interaction_model import InteractionModelError, Status
+from matter.testing.matter_testing import AttributeMatcher
 
 logger = logging.getLogger(__name__)
 
 
+def wmark_osd_matcher(attribute_id: int, wmark: bool, osd: bool, wmark_check: bool, osd_check: bool) -> "AttributeMatcher":
+    def predicate(report: "AttributeValue") -> bool:
+        if report.attribute != attribute_id:
+            return False
+
+        if len(report.value) == 0:
+            return False
+
+        stream = report.value[0]
+        wmark_match = (not wmark_check) or (stream.watermarkEnabled == wmark)
+        osd_match = (not osd_check) or (stream.OSDEnabled == osd)
+
+        return wmark_match and osd_match
+
+    return AttributeMatcher.from_callable(
+        description=f"check watermarkEnabled is {wmark} and OSDEnabled is {osd}",
+        matcher=predicate
+    )
+
+
 class AVSMTestBase:
     async def precondition_one_allocated_snapshot_stream(self):
         endpoint = self.get_endpoint(default=1)

src/python_testing/TC_AVSM_2_3.py

@@ -38,10 +38,11 @@
 import logging
 
 from mobly import asserts
-from TC_AVSMTestBase import AVSMTestBase
+from TC_AVSMTestBase import AVSMTestBase, wmark_osd_matcher
 
 import matter.clusters as Clusters
 from matter.interaction_model import InteractionModelError, Status
+from matter.testing.event_attribute_reporting import AttributeSubscriptionHandler
 from matter.testing.matter_testing import MatterBaseTest, TestStep, default_matter_test_main, has_feature, run_if_endpoint_matches
 
 logger = logging.getLogger(__name__)
@@ -80,12 +81,20 @@ def steps_TC_AVSM_2_3(self) -> list[TestStep]:
             ),
             TestStep(
                 5,
+                "Establish a subscription to the AllocatedSnapshotStreams attribute"
+            ),
+            TestStep(
+                6,
                 "TH sends the SnapshotStreamModify command with SnapshotStreamID set to aStreamID. If WMARK is supported, set WaterMarkEnabled to !aWmark`and if OSD is supported, set OSDEnabled to `!aOSD in the command.",
                 "If HardwareEncoder is True, DUT responds with a SUCCESS status code.",
-                "If HardwareEncoder is False, DUT responds with an INVALID_INSTATE status code."
+                "If HardwareEncoder is False, DUT responds with an INVALID_INSTATE status code.",
             ),
             TestStep(
-                6,
+                7,
+                "If HardwareEncoder is True, wait until a subscription report with AllocatedSnapshotStreams attribute is received",
+            ),
+            TestStep(
+                8,
                 "TH reads AllocatedSnapshotStreams attribute from CameraAVStreamManagement Cluster on DUT",
                 "If HardwareEncoder is True, verify the following: If WMARK is supported, verify WaterMarkEnabled == !aWmark. If OSD is supported, verify OSDEnabled == !aOSD.",
             ),
@@ -157,6 +166,13 @@ async def test_TC_AVSM_2_3(self):
             pass
 
         self.step(5)
+        # Establish subscription to AllocatedSnapshotStreams
+        sub_handler = AttributeSubscriptionHandler(cluster, attr.AllocatedSnapshotStreams)
+        await sub_handler.start(self.default_controller, self.dut_node_id, endpoint=endpoint, min_interval_sec=0, max_interval_sec=30, keepSubscriptions=False)
+
+        sub_handler.reset()
+
+        self.step(6)
         try:
             cmd = commands.SnapshotStreamModify(
                 snapshotStreamID=aStreamID,
@@ -171,7 +187,14 @@ async def test_TC_AVSM_2_3(self):
                 asserts.assert_equal(e.status, Status.InvalidInState, "Wrong error code returned")
             pass
 
-        self.step(6)
+        self.step(7)
+        if aHardwareEncoder:
+            expected_wmark = not aWmark
+            expected_osd = not aOSD
+            sub_handler.await_all_expected_report_matches(expected_matchers=[wmark_osd_matcher(
+                attr.AllocatedSnapshotStreams, expected_wmark, expected_osd, wmarkSupport, osdSupport)], timeout_sec=20)
+
+        self.step(8)
         aAllocatedSnapshotStreams = await self.read_single_attribute_check_success(
             endpoint=endpoint, cluster=cluster, attribute=attr.AllocatedSnapshotStreams
         )

src/python_testing/TC_AVSM_2_8.py

@@ -38,10 +38,11 @@
 import logging
 
 from mobly import asserts
-from TC_AVSMTestBase import AVSMTestBase
+from TC_AVSMTestBase import AVSMTestBase, wmark_osd_matcher
 
 import matter.clusters as Clusters
 from matter.interaction_model import InteractionModelError, Status
+from matter.testing.event_attribute_reporting import AttributeSubscriptionHandler
 from matter.testing.matter_testing import MatterBaseTest, TestStep, default_matter_test_main, has_feature, run_if_endpoint_matches
 
 logger = logging.getLogger(__name__)
@@ -79,11 +80,19 @@ def steps_TC_AVSM_2_8(self) -> list[TestStep]:
             ),
             TestStep(
                 5,
+                "Establish a subscription to the AllocatedVideoStreams attribute"
+            ),
+            TestStep(
+                6,
                 "TH sends the VideoStreamModify command with VideoStreamID set to aStreamID. If WMARK is supported, set WaterMarkEnabled to !aWmark and if OSD is supported, set OSDEnabled to !aOSD in the command.",
                 "DUT responds with a SUCCESS status code.",
             ),
             TestStep(
-                6,
+                7,
+                "Wait until a subscription report with AllocatedVideoStreams attribute is received",
+            ),
+            TestStep(
+                8,
                 "TH reads AllocatedVideoStreams attribute from CameraAVStreamManagement Cluster on DUT",
                 "Verify the following: If WMARK is supported, verify WaterMarkEnabled == !aWmark. If OSD is supported, verify OSDEnabled == !aOSD.",
             ),
@@ -151,6 +160,13 @@ async def test_TC_AVSM_2_8(self):
             pass
 
         self.step(5)
+        # Establish subscription to AllocatedSnapshotStreams
+        sub_handler = AttributeSubscriptionHandler(cluster, attr.AllocatedVideoStreams)
+        await sub_handler.start(self.default_controller, self.dut_node_id, endpoint=endpoint, min_interval_sec=0, max_interval_sec=30, keepSubscriptions=False)
+
+        sub_handler.reset()
+
+        self.step(6)
         try:
             videoStreamModifyCmd = commands.VideoStreamModify(
                 videoStreamID=aStreamID,
@@ -162,7 +178,13 @@ async def test_TC_AVSM_2_8(self):
             asserts.assert_equal(e.status, Status.Success, "Unexpected error returned")
             pass
 
-        self.step(6)
+        self.step(7)
+        expected_wmark = not aWmark
+        expected_osd = not aOSD
+        sub_handler.await_all_expected_report_matches(expected_matchers=[wmark_osd_matcher(
+            attr.AllocatedVideoStreams, expected_wmark, expected_osd, wmarkSupport, osdSupport)], timeout_sec=20)
+
+        self.step(8)
         aAllocatedVideoStreams = await self.read_single_attribute_check_success(
             endpoint=endpoint, cluster=cluster, attribute=attr.AllocatedVideoStreams
         )