tests: bluetooth: add test case that sdp client l2cap connecting fail

test the sdp request callback is called when sdp l2cap connecting fail.

Signed-off-by: Mark Wang <yichang.wang@nxp.com>

Author: MarkWangChinese

tests/bluetooth/classic/sdp_c/pytest/test_sdp.py

@@ -23,6 +23,7 @@
 )
 from bumble.device import Device
 from bumble.hci import Address, HCI_Write_Page_Timeout_Command
+from bumble.l2cap import ClassicChannel, L2CAP_Connection_Response
 from bumble.sdp import (
     SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
     SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
@@ -888,6 +889,59 @@ async def sdp_sa_discover_multiple_records(hci_port, shell, dut, address) -> Non
             assert found is True
 
 
+async def sdp_ssa_discover_fail(hci_port, shell, dut, address) -> None:
+    def on_app_connection_request(self, request) -> None:
+        logger.info('Force L2CAP connection failure')
+        self.destination_cid = request.source_cid
+        self.send_control_frame(
+            L2CAP_Connection_Response(
+                identifier=request.identifier,
+                destination_cid=self.source_cid,
+                source_cid=self.destination_cid,
+                result=L2CAP_Connection_Response.CONNECTION_REFUSED_NO_RESOURCES_AVAILABLE,
+                status=0x0000,
+            )
+        )
+
+    # Save the origin method
+    on_connection_request = ClassicChannel.on_connection_request
+    # Replace the origin method with a new one to force L2CAP connection failure
+    ClassicChannel.on_connection_request = on_app_connection_request
+
+    logger.info('<<< SDP Discovery ...')
+    async with await open_transport_or_link(hci_port) as hci_transport:
+        device = Device.with_hci(
+            'Bumble',
+            Address('F0:F1:F2:F3:F4:F5'),
+            hci_transport.source,
+            hci_transport.sink,
+        )
+        device.classic_enabled = True
+        device.le_enabled = False
+        with open(f"bumble_hci_{sys._getframe().f_code.co_name}.log", "wb") as snoop_file:
+            device.host.snooper = BtSnooper(snoop_file)
+            await device_power_on(device)
+            await device.send_command(HCI_Write_Page_Timeout_Command(page_timeout=0xFFFF))
+
+            target_address = address.split(" ")[0]
+            logger.info(f'=== Connecting to {target_address}...')
+            try:
+                connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT)
+                logger.info(f'=== Connected to {connection.peer_address}!')
+            except Exception as e:
+                logger.error(f'Fail to connect to {target_address}!')
+                raise e
+
+            # Discover SDP Record
+            shell.exec_command("sdp_client ssa_discovery_fail")
+            found, lines = await wait_for_shell_response(dut, "test pass")
+            logger.info(f'{lines}')
+            assert found is True
+
+    # Restore the origin method
+    ClassicChannel.on_connection_request = on_connection_request
+
+
 class TestSdpServer:
     def test_sdp_ssa_discover_no_record(self, shell: Shell, dut: DeviceAdapter, sdp_client_dut):
         """Test case to request SDP records. No SDP record registered."""
@@ -966,3 +1020,9 @@ def test_sdp_sa_discover_multiple_records(
         logger.info(f'test_sdp_sa_discover_multiple_records {sdp_client_dut}')
         hci, iut_address = sdp_client_dut
         asyncio.run(sdp_sa_discover_multiple_records(hci, shell, dut, iut_address))
+
+    def test_sdp_ssa_discover_fail(self, shell: Shell, dut: DeviceAdapter, sdp_client_dut):
+        """Test case to request SDP records. but the L2CAP connecting fail."""
+        logger.info(f'test_sdp_ssa_discover_fail {sdp_client_dut}')
+        hci, iut_address = sdp_client_dut
+        asyncio.run(sdp_ssa_discover_fail(hci, shell, dut, iut_address))

tests/bluetooth/classic/sdp_c/src/sdp_client.c

@@ -280,10 +280,40 @@ static int cmd_sa_discovery(const struct shell *sh, size_t argc, char *argv[])
 	return 0;
 }
 
+static uint8_t sdp_discover_fail_func(struct bt_conn *conn, struct bt_sdp_client_result *result,
+				      const struct bt_sdp_discover_params *params)
+{
+	if ((result == NULL) || (result->resp_buf == NULL) || (result->resp_buf->len == 0)) {
+		printk("test pass\n");
+	} else {
+		printk("test fail\n");
+	}
+
+	return BT_SDP_DISCOVER_UUID_STOP;
+}
+
+static int cmd_ssa_discovery_fail(const struct shell *sh, size_t argc, char *argv[])
+{
+	int err;
+
+	sdp_discover.uuid = BT_UUID_DECLARE_16(BT_SDP_HANDSFREE_SVCLASS),
+	sdp_discover.func = sdp_discover_fail_func;
+	sdp_discover.pool = &sdp_client_pool;
+	sdp_discover.type = BT_SDP_DISCOVER_SERVICE_SEARCH_ATTR;
+
+	err = bt_sdp_discover(default_conn, &sdp_discover);
+	if (err) {
+		shell_error(sh, "Fail to start SDP Discovery (err %d)", err);
+		return err;
+	}
+	return 0;
+}
+
 SHELL_STATIC_SUBCMD_SET_CREATE(sdp_client_cmds,
 	SHELL_CMD_ARG(ss_discovery, NULL, "<UUID>", cmd_ss_discovery, 2, 0),
 	SHELL_CMD_ARG(sa_discovery, NULL, "<Service Record Handle>", cmd_sa_discovery, 2, 0),
 	SHELL_CMD_ARG(ssa_discovery, NULL, "<UUID>", cmd_ssa_discovery, 2, 0),
+	SHELL_CMD_ARG(ssa_discovery_fail, NULL, "", cmd_ssa_discovery_fail, 1, 0),
 	SHELL_SUBCMD_SET_END
 );