[Camera] Add Matter Camera TCs (TC-WEBRTC-1.1, TC-WEBRTC-1.2, TC-WEBRTC-1.3, TC-WEBRTC-1.4, TC-WEBRTC-1.5) (#38584)
* Add TC_WEBRTC_1_1 to python_testing
Signed-off-by: abhishek.h <abhishek.h@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Add TC_WEBRTC_1_2 to python_testing
Signed-off-by: abhishek.h <abhishek.h@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Add TC_WEBRTC_1_3 to python_testing
Signed-off-by: abhishek.h <abhishek.h@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Add TC_WEBRTC_1_4 to python_testing
Signed-off-by: abhishek.h <abhishek.h@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Add TC_WEBRTC_1_5 to python_testing
Signed-off-by: abhishek.h <abhishek.h@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Restyled by isort
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* ruff linting fix
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Don't wait for user inetraction in CI - check peer connection state instead for session establishment
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Exclude WEBRTC TCs from the CI.
Exclude WEBRTC TCs from the CI, until python webrtc bindings and
python controller changes are landed.
Signed-off-by: suyambu.rm <suyambu.rm@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Update TCs based on controller refactor
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Restyled by prettier-yaml
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
* Allocate video stream based on values from DUT
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
---------
Signed-off-by: abhishek.h <abhishek.h@samsung.com>
Signed-off-by: Charles Kim <chulspro.kim@samsung.com>
Signed-off-by: suyambu.rm <suyambu.rm@samsung.com>
Co-authored-by: abhishek.h <abhishek.h@samsung.com>
Co-authored-by: Restyled.io <commits@restyled.io>
Co-authored-by: suyambu.rm <suyambu.rm@samsung.com>
diff --git a/src/python_testing/TC_WEBRTC_1_1.py b/src/python_testing/TC_WEBRTC_1_1.py
new file mode 100644
index 0000000..764efe7
--- /dev/null
+++ b/src/python_testing/TC_WEBRTC_1_1.py
@@ -0,0 +1,130 @@
+#
+# Copyright (c) 2025 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.
+
+from chip.ChipDeviceCtrl import TransportPayloadCapability
+from chip.clusters import WebRTCTransportProvider
+from chip.clusters.Types import NullValue
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.webrtc import PeerConnection, WebRTCManager
+from mobly import asserts
+from TC_WEBRTC_Utils import WebRTCTestHelper
+from test_plan_support import commission_if_required
+
+
+class TC_WEBRTC_1_1(MatterBaseTest, WebRTCTestHelper):
+ def steps_TC_WEBRTC_1_1(self) -> list[TestStep]:
+ steps = [
+ TestStep("precondition-1", commission_if_required(), is_commissioning=True),
+ TestStep("precondition-2", "Confirm no active WebRTC sessions exist in DUT"),
+ TestStep(1, "TH sends the ProvideOffer command with an SDP Offer and null WebRTCSessionID to the DUT."),
+ TestStep(2, "DUT sends Answer command to the TH/WEBRTCR."),
+ TestStep(3, "TH sends the SUCCESS status code to the DUT."),
+ TestStep(4, "TH sends the ICECandidates command with a its ICE candidates to the DUT."),
+ TestStep(5, "DUT sends ProvideICECandidates command to the TH/WEBRTCR."),
+ TestStep(6, "TH waits for 5 seconds. Verify the WebRTC session has been successfully established."),
+ ]
+
+ return steps
+
+ def desc_TC_WEBRTC_1_1(self) -> str:
+ return "[TC-WEBRTC-1.1] Validate that setting an SDP Offer successfully initiates a new WebRTC session."
+
+ def pics_TC_WEBRTC_1_1(self) -> list[str]:
+ return ["WEBRTCR", "WEBRTCP"]
+
+ @async_test_body
+ async def test_TC_WEBRTC_1_1(self):
+ self.step("precondition-1")
+
+ endpoint = self.get_endpoint(default=1)
+ webrtc_manager = WebRTCManager()
+ webrtc_peer: PeerConnection = webrtc_manager.create_peer(
+ node_id=self.dut_node_id, fabric_index=self.default_controller.GetFabricIndexInternal(), endpoint=endpoint
+ )
+
+ self.step("precondition-2")
+ current_sessions = await self.read_single_attribute_check_success(
+ cluster=WebRTCTransportProvider, attribute=WebRTCTransportProvider.Attributes.CurrentSessions, endpoint=endpoint
+ )
+ asserts.assert_equal(len(current_sessions), 0, "Found an existing WebRTC session")
+
+ # Allocate video stream in DUT if possible, to receive actual video stream
+ # This step is not part of test plan
+ aVideoStreamID = await self.allocate_video_stream(endpoint)
+
+ # Test Invokation
+
+ self.step(1)
+
+ offer = webrtc_peer.get_local_offer()
+
+ provide_offer_response: WebRTCTransportProvider.Commands.ProvideOfferResponse = await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideOffer(
+ webRTCSessionID=NullValue,
+ sdp=offer,
+ streamUsage=WebRTCTransportProvider.Enums.StreamUsageEnum.kLiveView,
+ videoStreamID=aVideoStreamID,
+ audioStreamID=NullValue,
+ originatingEndpointID=1,
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+ session_id = provide_offer_response.webRTCSessionID
+ asserts.assert_true(session_id >= 0, "Invalid response")
+
+ webrtc_manager.session_id_created(session_id, self.dut_node_id)
+
+ self.step(2)
+
+ answer_sessionId, answer = webrtc_peer.get_remote_answer()
+
+ asserts.assert_equal(session_id, answer_sessionId, "ProvideAnswer invoked with wrong session id")
+ asserts.assert_true(len(answer) > 0, "Invalid answer SDP received")
+
+ self.step(3)
+
+ webrtc_peer.set_remote_answer(answer)
+
+ self.step(4)
+ local_candidates = webrtc_peer.get_local_ice_candidates()
+
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideICECandidates(
+ webRTCSessionID=answer_sessionId, ICECandidates=local_candidates
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(5)
+ ice_session_id, remote_candidates = webrtc_peer.get_remote_ice_candidates()
+ asserts.assert_equal(session_id, ice_session_id, "ProvideIceCandidates invoked with wrong session id")
+ asserts.assert_true(len(remote_candidates) > 0, "Invalid remote ice candidates received")
+
+ webrtc_peer.set_remote_ice_candidates(remote_candidates)
+
+ self.step(6)
+ if not self.is_pics_sdk_ci_only and aVideoStreamID != NullValue:
+ self.wait_for_user_input("Verify WebRTC session is established")
+ else:
+ webrtc_peer.wait_for_session_establishment()
+
+ webrtc_manager.close_all()
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_WEBRTC_1_2.py b/src/python_testing/TC_WEBRTC_1_2.py
new file mode 100644
index 0000000..d8533e8
--- /dev/null
+++ b/src/python_testing/TC_WEBRTC_1_2.py
@@ -0,0 +1,167 @@
+#
+# Copyright (c) 2025 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.
+
+from chip.ChipDeviceCtrl import TransportPayloadCapability
+from chip.clusters import WebRTCTransportProvider
+from chip.clusters.Types import NullValue
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.webrtc import PeerConnection, WebRTCManager
+from mobly import asserts
+from test_plan_support import commission_if_required
+
+
+class TC_WEBRTC_1_2(MatterBaseTest):
+ def steps_TC_WEBRTC_1_2(self) -> list[TestStep]:
+ steps = [
+ TestStep("precondition-1", commission_if_required(), is_commissioning=True),
+ TestStep("precondition-2", "Confirm there is an active WebRTC sessions exist in DUT"),
+ TestStep(0, "TH Reads CurrentSessions attribute from WEBRTCP (DUT)"),
+ TestStep(1, "TH sends the ProvideOffer command with an SDP Offer with the WebRTCSessionID saved in step 1 to the DUT."),
+ TestStep(2, "TH waits up to 30 seconds for Answer command from the DUT."),
+ TestStep(3, "TH sends the SUCCESS status code to the DUT."),
+ TestStep(4, "TH sends the ICECandidates command with a its ICE candidates to the DUT."),
+ TestStep(5, "TH waits up to 30 seconds for ProvideICECandidates command from the DUT."),
+ TestStep(6, "TH waits for 10 seconds."),
+ TestStep(7, "TH sends the EndSession command with the WebRTCSessionID saved in step 1 to the DUT."),
+ ]
+
+ return steps
+
+ def desc_TC_WEBRTC_1_2(self) -> str:
+ return "[TC-WEBRTC-1.2] Validate that providing an existing WebRTC session ID with an SDP Offer successfully triggers the re-offer flow."
+
+ def pics_TC_WEBRTC_1_2(self) -> list[str]:
+ return ["WEBRTCR", "WEBRTCP"]
+
+ @async_test_body
+ async def test_TC_WEBRTC_1_2(self):
+ self.step("precondition-1")
+
+ endpoint = self.get_endpoint(default=1)
+ webrtc_manager = WebRTCManager()
+ webrtc_peer: PeerConnection = webrtc_manager.create_peer(
+ node_id=self.dut_node_id, fabric_index=self.default_controller.GetFabricIndexInternal(), endpoint=endpoint
+ )
+ # Test Invokation
+ self.step("precondition-2")
+ await establish_webrtc_session(webrtc_manager, webrtc_peer, endpoint, self)
+
+ self.step(0)
+ current_sessions = await self.read_single_attribute_check_success(
+ cluster=WebRTCTransportProvider, attribute=WebRTCTransportProvider.Attributes.CurrentSessions, endpoint=endpoint
+ )
+ asserts.assert_equal(len(current_sessions), 1, "No Pre-Existing session found")
+ prev_sessionid = current_sessions[0].id
+
+ self.step(1)
+ offer = webrtc_peer.get_local_offer()
+
+ provide_offer_response: WebRTCTransportProvider.Commands.ProvideOfferResponse = await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideOffer(
+ webRTCSessionID=prev_sessionid,
+ sdp=offer,
+ streamUsage=WebRTCTransportProvider.Enums.StreamUsageEnum.kLiveView,
+ videoStreamID=NullValue,
+ audioStreamID=NullValue,
+ originatingEndpointID=1,
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ asserts.assert_equal(
+ provide_offer_response.webRTCSessionID, prev_sessionid, "Session id does not match with the previous session"
+ )
+
+ self.step(2)
+ answer_sessionId, answer = webrtc_peer.get_remote_answer(timeout=30)
+
+ asserts.assert_equal(prev_sessionid, answer_sessionId, "Session id does not match with the previous session")
+ asserts.assert_true(len(answer) > 0, "Invalid answer SDP received")
+
+ self.step(3)
+ webrtc_peer.set_remote_answer(answer)
+
+ self.step(4)
+ local_candidates = webrtc_peer.get_local_ice_candidates()
+
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideICECandidates(
+ webRTCSessionID=answer_sessionId, ICECandidates=local_candidates
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(5)
+ ice_session_id, remote_candidates = webrtc_peer.get_remote_ice_candidates()
+ asserts.assert_equal(prev_sessionid, ice_session_id, "ProvideIceCandidates invoked with wrong session id")
+ asserts.assert_true(len(remote_candidates) > 0, "Invalid remote ice candidates received")
+ webrtc_peer.set_remote_ice_candidates(remote_candidates)
+
+ self.step(6)
+ if not self.is_pics_sdk_ci_only:
+ self.wait_for_user_input("Verify WebRTC session is established")
+ else:
+ webrtc_peer.wait_for_session_establishment()
+
+ self.step(7)
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.EndSession(
+ webRTCSessionID=prev_sessionid, reason=WebRTCTransportProvider.Enums.WebRTCEndReasonEnum.kUserHangup
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ webrtc_manager.close_all()
+
+
+async def establish_webrtc_session(webrtc_manager, webrtc_peer, endpoint, ctrl):
+ offer = webrtc_peer.get_local_offer()
+ provide_offer_response: WebRTCTransportProvider.Commands.ProvideOfferResponse = await ctrl.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideOffer(
+ webRTCSessionID=NullValue,
+ sdp=offer,
+ streamUsage=WebRTCTransportProvider.Enums.StreamUsageEnum.kLiveView,
+ videoStreamID=NullValue,
+ audioStreamID=NullValue,
+ originatingEndpointID=1,
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+ asserts.assert_true(provide_offer_response.webRTCSessionID >= 0, "Invalid response")
+ webrtc_manager.session_id_created(provide_offer_response.webRTCSessionID, ctrl.dut_node_id)
+
+ answer_sessionId, answer = webrtc_peer.get_remote_answer(timeout=30)
+ webrtc_peer.set_remote_answer(answer)
+
+ local_candidates = webrtc_peer.get_local_ice_candidates()
+ await ctrl.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideICECandidates(webRTCSessionID=answer_sessionId, ICECandidates=local_candidates),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ ice_session_id, remote_candidates = webrtc_peer.get_remote_ice_candidates()
+ webrtc_peer.set_remote_ice_candidates(remote_candidates)
+
+ return webrtc_peer.wait_for_session_establishment()
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_WEBRTC_1_3.py b/src/python_testing/TC_WEBRTC_1_3.py
new file mode 100644
index 0000000..e14f6d9
--- /dev/null
+++ b/src/python_testing/TC_WEBRTC_1_3.py
@@ -0,0 +1,130 @@
+#
+# Copyright (c) 2025 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.
+
+from chip.ChipDeviceCtrl import TransportPayloadCapability
+from chip.clusters import WebRTCTransportProvider
+from chip.clusters.Types import NullValue
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.webrtc import PeerConnection, WebRTCManager
+from mobly import asserts
+from test_plan_support import commission_if_required
+
+
+class TC_WEBRTC_1_3(MatterBaseTest):
+ def steps_TC_WEBRTC_1_3(self) -> list[TestStep]:
+ steps = [
+ TestStep("precondition-1", commission_if_required(), is_commissioning=True),
+ TestStep("precondition-2", "Confirm no active WebRTC sessions exist in DUT"),
+ TestStep(1, "TH sends the SolicitOffer command without ICEServers and ICETransportPolicy to the DUT."),
+ TestStep(2, "TH waits up to 30 seconds for Offer command from the DUT."),
+ TestStep(3, "TH sends the SUCCESS status code to the DUT."),
+ TestStep(4, "TH sends the ProvideAnswer command with the WebRTCSessionID saved in step 1 and a SDP Offer to the DUT."),
+ TestStep(5, "TH sends the ICECandidates command with a its ICE candidates to the DUT."),
+ TestStep(6, "TH waits up to 30 seconds for ProvideICECandidates command from the DUT."),
+ TestStep(7, "TH waits for 10 seconds."),
+ TestStep(8, "TH sends the EndSession command with the WebRTCSessionID saved in step 1 to the DUT."),
+ ]
+
+ return steps
+
+ def desc_TC_WEBRTC_1_3(self) -> str:
+ return "[TC-WEBRTC-1.3] Validate Deferred Offer Flow for Battery-Powered Camera in Standby Mode."
+
+ def pics_TC_WEBRTC_1_3(self) -> list[str]:
+ return ["WEBRTCR", "WEBRTCP"]
+
+ @async_test_body
+ async def test_TC_WEBRTC_1_3(self):
+ self.step("precondition-1")
+
+ endpoint = self.get_endpoint(default=1)
+ webrtc_manager = WebRTCManager()
+ webrtc_peer: PeerConnection = webrtc_manager.create_peer(
+ node_id=self.dut_node_id, fabric_index=self.default_controller.GetFabricIndexInternal(), endpoint=endpoint
+ )
+
+ self.step("precondition-2")
+ current_sessions = await self.read_single_attribute_check_success(
+ cluster=WebRTCTransportProvider, attribute=WebRTCTransportProvider.Attributes.CurrentSessions, endpoint=endpoint
+ )
+ asserts.assert_equal(len(current_sessions), 0, "Found an existing WebRTC session")
+
+ self.step(1)
+ solicit_offer_response: WebRTCTransportProvider.Commands.SolicitOfferResponse = await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.SolicitOffer(
+ streamUsage=WebRTCTransportProvider.Enums.StreamUsageEnum.kLiveView,
+ videoStreamID=NullValue,
+ audioStreamID=NullValue,
+ originatingEndpointID=1,
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+ session_id = solicit_offer_response.webRTCSessionID
+ asserts.assert_true(session_id >= 0, "Invalid response")
+ asserts.assert_true(solicit_offer_response.deferredOffer, "Expected deferredOffer = True")
+ webrtc_manager.session_id_created(session_id, self.dut_node_id)
+
+ self.step(2)
+ offer_sessionId, remote_offer_sdp = webrtc_peer.get_remote_offer(timeout=30)
+ asserts.assert_equal(offer_sessionId, session_id, "Invalid session id")
+ asserts.assert_true(len(remote_offer_sdp) > 0, "Invalid offer sdp received")
+
+ self.step(3)
+ webrtc_peer.set_remote_offer(remote_offer_sdp)
+
+ self.step(4)
+ local_answer = webrtc_peer.get_local_answer()
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideAnswer(webRTCSessionID=session_id, sdp=local_answer),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(5)
+ local_candidates = webrtc_peer.get_local_ice_candidates()
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideICECandidates(webRTCSessionID=session_id, ICECandidates=local_candidates),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(6)
+ ice_session_id, remote_candidates = webrtc_peer.get_remote_ice_candidates(timeout=30)
+ asserts.assert_equal(ice_session_id, session_id, "Invalid session id")
+ asserts.assert_true(len(remote_candidates) > 0, "Invalid ice candidates received")
+ webrtc_peer.set_remote_ice_candidates(remote_candidates)
+
+ self.step(7)
+ if not self.is_pics_sdk_ci_only:
+ self.wait_for_user_input("Verify WebRTC session is established")
+ else:
+ webrtc_peer.wait_for_session_establishment()
+
+ self.step(8)
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.EndSession(
+ webRTCSessionID=session_id, reason=WebRTCTransportProvider.Enums.WebRTCEndReasonEnum.kUserHangup
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ webrtc_manager.close_all()
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_WEBRTC_1_4.py b/src/python_testing/TC_WEBRTC_1_4.py
new file mode 100644
index 0000000..3bf02d2
--- /dev/null
+++ b/src/python_testing/TC_WEBRTC_1_4.py
@@ -0,0 +1,133 @@
+#
+# Copyright (c) 2025 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.
+
+
+from chip.ChipDeviceCtrl import TransportPayloadCapability
+from chip.clusters import WebRTCTransportProvider
+from chip.clusters.Types import NullValue
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.webrtc import PeerConnection, WebRTCManager
+from mobly import asserts
+from test_plan_support import commission_if_required
+
+
+class TC_WEBRTC_1_4(MatterBaseTest):
+ def steps_TC_WEBRTC_1_4(self) -> list[TestStep]:
+ steps = [
+ TestStep("precondition-1", commission_if_required(), is_commissioning=True),
+ TestStep("precondition-2", "Confirm no active WebRTC sessions exist in DUT"),
+ TestStep(1, "TH sends the SolicitOffer command without ICEServers and ICETransportPolicy to the DUT."),
+ TestStep(2, "DUT sends Offer command to TH/WEBRTCR."),
+ TestStep(3, "TH sends the SUCCESS status code to the DUT."),
+ TestStep(4, "TH sends the ProvideAnswer command with the WebRTCSessionID saved in step 1 and a SDP Offer to the DUT."),
+ TestStep(5, "TH sends the ICECandidates command with a its ICE candidates to the DUT."),
+ TestStep(6, "TH waits up to 30 seconds for ProvideICECandidates command from the DUT."),
+ TestStep(7, "TH waits for 10 seconds."),
+ TestStep(8, "TH sends the EndSession command with the WebRTCSessionID saved in step 1 to the DUT."),
+ ]
+
+ return steps
+
+ def desc_TC_WEBRTC_1_4(self) -> str:
+ return "[TC-WEBRTC-1.3] Validate Deferred Offer Flow for Battery-Powered Camera in Standby Mode."
+
+ def pics_TC_WEBRTC_1_4(self) -> list[str]:
+ return ["WEBRTCR", "WEBRTCP"]
+
+ @async_test_body
+ async def test_TC_WEBRTC_1_4(self):
+ self.step("precondition-1")
+
+ endpoint = self.get_endpoint(default=1)
+ webrtc_manager = WebRTCManager()
+ webrtc_peer: PeerConnection = webrtc_manager.create_peer(
+ node_id=self.dut_node_id, fabric_index=self.default_controller.GetFabricIndexInternal(), endpoint=endpoint
+ )
+
+ self.step("precondition-2")
+ current_sessions = await self.read_single_attribute_check_success(
+ cluster=WebRTCTransportProvider, attribute=WebRTCTransportProvider.Attributes.CurrentSessions, endpoint=endpoint
+ )
+ asserts.assert_equal(len(current_sessions), 0, "Found an existing WebRTC session")
+
+ self.step(1)
+ solicit_offer_response: WebRTCTransportProvider.Commands.SolicitOfferResponse = await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.SolicitOffer(
+ streamUsage=WebRTCTransportProvider.Enums.StreamUsageEnum.kLiveView,
+ videoStreamID=NullValue,
+ audioStreamID=NullValue,
+ originatingEndpointID=1,
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+ session_id = solicit_offer_response.webRTCSessionID
+ asserts.assert_true(session_id >= 0, "Invalid response")
+ asserts.assert_false(solicit_offer_response.deferredOffer, "Expected deferredOffer = False")
+
+ webrtc_manager.session_id_created(session_id, self.dut_node_id)
+
+ self.step(2)
+ # TH should immediately get Offer from DUT
+ offer_sessionId, remote_offer_sdp = webrtc_peer.get_remote_offer(timeout=5)
+ asserts.assert_equal(offer_sessionId, session_id, "Invalid session id")
+ asserts.assert_true(len(remote_offer_sdp) > 0, "Invalid offer sdp received")
+
+ self.step(3)
+ webrtc_peer.set_remote_offer(remote_offer_sdp)
+
+ self.step(4)
+ local_answer = webrtc_peer.get_local_answer()
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideAnswer(webRTCSessionID=session_id, sdp=local_answer),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(5)
+ local_candidates = webrtc_peer.get_local_ice_candidates()
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideICECandidates(webRTCSessionID=session_id, ICECandidates=local_candidates),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(6)
+ ice_session_id, remote_candidates = webrtc_peer.get_remote_ice_candidates(timeout=30)
+ asserts.assert_equal(ice_session_id, session_id, "Invalid session id")
+ asserts.assert_true(len(remote_candidates) > 0, "Invalid ice candidates received")
+ webrtc_peer.set_remote_ice_candidates(remote_candidates)
+
+ self.step(7)
+ if not self.is_pics_sdk_ci_only:
+ self.wait_for_user_input("Verify WebRTC session is established")
+ else:
+ webrtc_peer.wait_for_session_establishment()
+
+ self.step(8)
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.EndSession(
+ webRTCSessionID=session_id, reason=WebRTCTransportProvider.Enums.WebRTCEndReasonEnum.kUserHangup
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ webrtc_manager.close_all()
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_WEBRTC_1_5.py b/src/python_testing/TC_WEBRTC_1_5.py
new file mode 100644
index 0000000..0fc945f
--- /dev/null
+++ b/src/python_testing/TC_WEBRTC_1_5.py
@@ -0,0 +1,113 @@
+#
+# Copyright (c) 2025 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.
+
+
+from chip.ChipDeviceCtrl import TransportPayloadCapability
+from chip.clusters import WebRTCTransportProvider
+from chip.testing.matter_testing import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
+from chip.webrtc import PeerConnection, WebRTCManager
+from mobly import asserts
+from test_plan_support import commission_if_required
+
+
+class TC_WEBRTC_1_5(MatterBaseTest):
+ def steps_TC_WEBRTC_1_5(self) -> list[TestStep]:
+ steps = [
+ TestStep("precondition-1", commission_if_required(), is_commissioning=True),
+ TestStep("precondition-2", "Confirm no active WebRTC sessions exist in DUT"),
+ TestStep(1, "Follow manufacturer provided instructions to have DUT send the Offer command to the TH."),
+ TestStep(2, "TH sends the SUCCESS status code to the DUT."),
+ TestStep(3, "TH sends the ProvideAnswer command with the WebRTCSessionID saved in step 1 and a SDP Offer to the DUT."),
+ TestStep(4, "TH sends the ICECandidates command with a its ICE candidates to the DUT."),
+ TestStep(5, "DUT sends ProvideICECandidates command to TH/WEBRTCR."),
+ TestStep(6, "TH waits for 5 seconds."),
+ TestStep(7, "TH sends the EndSession command with the WebRTCSessionID saved in step 1 to the DUT."),
+ ]
+
+ return steps
+
+ def desc_TC_WEBRTC_1_5(self) -> str:
+ return "[TC-WEBRTC-1.3] Validate Deferred Offer Flow for Battery-Powered Camera in Standby Mode."
+
+ def pics_TC_WEBRTC_1_5(self) -> list[str]:
+ return ["WEBRTCR", "WEBRTCP"]
+
+ @async_test_body
+ async def test_TC_WEBRTC_1_5(self):
+ self.step("precondition-1")
+ endpoint = self.get_endpoint(default=1)
+ webrtc_manager = WebRTCManager()
+ webrtc_peer: PeerConnection = webrtc_manager.create_peer(
+ node_id=self.dut_node_id, fabric_index=self.default_controller.GetFabricIndexInternal(), endpoint=endpoint
+ )
+
+ self.step("precondition-2")
+ current_sessions = await self.read_single_attribute_check_success(
+ cluster=WebRTCTransportProvider, attribute=WebRTCTransportProvider.Attributes.CurrentSessions, endpoint=endpoint
+ )
+ asserts.assert_equal(len(current_sessions), 0, "Found an existing WebRTC session")
+
+ self.step(1)
+ session_id, remote_offer_sdp = webrtc_peer.get_remote_offer(timeout=30)
+ asserts.assert_true(session_id >= 0, "Invalid session id")
+ asserts.assert_true(len(remote_offer_sdp) > 0, "Invalid offer sdp received")
+ webrtc_manager.session_id_created(session_id, self.dut_node_id)
+
+ self.step(2)
+ webrtc_peer.set_remote_offer(remote_offer_sdp)
+
+ self.step(3)
+ local_answer = webrtc_peer.get_local_answer()
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideAnswer(webRTCSessionID=session_id, sdp=local_answer),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(4)
+ local_candidates = webrtc_peer.get_local_ice_candidates()
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.ProvideICECandidates(webRTCSessionID=session_id, ICECandidates=local_candidates),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ self.step(5)
+ ice_session_id, remote_candidates = webrtc_peer.get_remote_ice_candidates(timeout=30)
+ asserts.assert_equal(ice_session_id, session_id, "Invalid session id")
+ asserts.assert_true(len(remote_candidates) > 0, "Invalid ice candidates received")
+ webrtc_peer.set_remote_ice_candidates(remote_candidates)
+
+ self.step(6)
+ if not self.is_pics_sdk_ci_only:
+ self.wait_for_user_input("Verify WebRTC session is established")
+ else:
+ webrtc_peer.wait_for_session_establishment()
+
+ self.step(7)
+ await self.send_single_cmd(
+ cmd=WebRTCTransportProvider.Commands.EndSession(
+ webRTCSessionID=session_id, reason=WebRTCTransportProvider.Enums.WebRTCEndReasonEnum.kUserHangup
+ ),
+ endpoint=endpoint,
+ payloadCapability=TransportPayloadCapability.LARGE_PAYLOAD,
+ )
+
+ webrtc_manager.close_all()
+
+
+if __name__ == "__main__":
+ default_matter_test_main()
diff --git a/src/python_testing/TC_WEBRTC_Utils.py b/src/python_testing/TC_WEBRTC_Utils.py
new file mode 100644
index 0000000..a79bcac
--- /dev/null
+++ b/src/python_testing/TC_WEBRTC_Utils.py
@@ -0,0 +1,68 @@
+#
+# Copyright (c) 2025 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 logging
+
+from chip.clusters import CameraAvStreamManagement
+from chip.clusters.Types import NullValue
+
+
+class WebRTCTestHelper:
+ async def read_avstr_attribute_expect_success(self, endpoint, attribute):
+ return await self.read_single_attribute_check_success(
+ endpoint=endpoint, cluster=CameraAvStreamManagement, attribute=attribute
+ )
+
+ async def allocate_video_stream(self, endpoint):
+ attrs = CameraAvStreamManagement.Attributes
+ try:
+ # Check for watermark and OSD features
+ feature_map = await self.read_avstr_attribute_expect_success(endpoint, attrs.FeatureMap)
+ watermark = True if (feature_map & CameraAvStreamManagement.Bitmaps.Feature.kWatermark) != 0 else None
+ osd = True if (feature_map & CameraAvStreamManagement.Bitmaps.Feature.kOnScreenDisplay) != 0 else None
+
+ # Get the parms from the device (those which are available)
+ aStreamUsagePriorities = await self.read_avstr_attribute_expect_success(endpoint, attrs.StreamUsagePriorities)
+ aRateDistortionTradeOffPoints = await self.read_avstr_attribute_expect_success(
+ endpoint, attrs.RateDistortionTradeOffPoints
+ )
+ aMinViewport = await self.read_avstr_attribute_expect_success(endpoint, attrs.MinViewport)
+ aVideoSensorParams = await self.read_avstr_attribute_expect_success(endpoint, attrs.VideoSensorParams)
+
+ response = await self.send_single_cmd(
+ cmd=CameraAvStreamManagement.Commands.VideoStreamAllocate(
+ streamUsage=aStreamUsagePriorities[0],
+ videoCodec=aRateDistortionTradeOffPoints[0].codec,
+ minFrameRate=30,
+ maxFrameRate=aVideoSensorParams.maxFPS,
+ minResolution=aMinViewport,
+ maxResolution=CameraAvStreamManagement.Structs.VideoResolutionStruct(
+ width=aVideoSensorParams.sensorWidth, height=aVideoSensorParams.sensorHeight
+ ),
+ minBitRate=aRateDistortionTradeOffPoints[0].minBitRate,
+ maxBitRate=aRateDistortionTradeOffPoints[0].minBitRate,
+ minKeyFrameInterval=2000,
+ maxKeyFrameInterval=8000,
+ watermarkEnabled=watermark,
+ OSDEnabled=osd,
+ ),
+ endpoint=endpoint,
+ )
+ return response.videoStreamID
+
+ except Exception as e:
+ logging.error(f"Failed to allocate video stream. {e}")
+ return NullValue
diff --git a/src/python_testing/test_metadata.yaml b/src/python_testing/test_metadata.yaml
index afd9131..49100bc 100644
--- a/src/python_testing/test_metadata.yaml
+++ b/src/python_testing/test_metadata.yaml
@@ -123,6 +123,28 @@
- name: TC_AVSUMTestBase.py
reason:
Shared code for Camera AV Settings (TC_AVSUM*), not a standalone test.
+ - name: TC_WEBRTC_1_1.py
+ reason:
+ Depends on Python WebRTC bindings. Will be enabled once local tests
+ are fully verified.
+ - name: TC_WEBRTC_1_2.py
+ reason:
+ Depends on Python WebRTC bindings. Will be enabled once local tests
+ are fully verified.
+ - name: TC_WEBRTC_1_3.py
+ reason:
+ Depends on Python WebRTC bindings. Will be enabled once local tests
+ are fully verified.
+ - name: TC_WEBRTC_1_4.py
+ reason:
+ Depends on Python WebRTC bindings. Will be enabled once local tests
+ are fully verified.
+ - name: TC_WEBRTC_1_5.py
+ reason:
+ Depends on Python WebRTC bindings. Will be enabled once local tests
+ are fully verified.
+ - name: TC_WEBRTC_Utils.py
+ reason: Helper methods for WebRTC TCs, not a standalone test.
# This is a list of slow tests (just arbitrarily picked around 20 seconds)
# used in some script reporting for "be patient" messages as well as potentially
