CommandCode.
CONNECTION_RESPONSE] = self._on_connection_response_configuration_request_with_unknown_options_and_hint
+ # more of a hack for the moment
+ def reply_with_continuation_flag(self):
+ self.control_table[
+ CommandCode.
+ CONNECTION_RESPONSE] = self._on_connection_response_configuration_request_with_continuation_flag
+
def _on_connection_request_default(self, l2cap_control_view):
connection_request_view = l2cap_packets.ConnectionRequestView(
l2cap_control_view)
self._acl.send(bytes(byte_array))
return True
+ def _on_connection_response_configuration_request_with_continuation_flag(
+ self, l2cap_control_view):
+ connection_response_view = l2cap_packets.ConnectionResponseView(
+ l2cap_control_view)
+ sid = connection_response_view.GetIdentifier()
+ scid = connection_response_view.GetSourceCid()
+ dcid = connection_response_view.GetDestinationCid()
+ self.scid_to_dcid[scid] = dcid
+
+ mtu_opt = l2cap_packets.MtuConfigurationOption()
+ mtu_opt.mtu = 0x1234
+
+ options = [mtu_opt]
+ config_request = l2cap_packets.ConfigurationRequestBuilder(
+ sid + 1, dcid, l2cap_packets.Continuation.CONTINUE, options)
+
+ self.control_channel.send(config_request)
+
+ flush_timeout_option = l2cap_packets.FlushTimeoutConfigurationOption()
+ flush_timeout_option.flush_timeout = 65535
+ option = [flush_timeout_option]
+ config_request = l2cap_packets.ConfigurationRequestBuilder(
+ sid + 2, dcid, l2cap_packets.Continuation.END, option)
+ self.get_control_channel().send(config_request)
+ return True
+
def _on_configuration_request_default(self, l2cap_control_view):
configuration_request = l2cap_packets.ConfigurationRequestView(
l2cap_control_view)
def cert_send_b_frame(self, b_frame):
self.cert_l2cap.send_acl(b_frame)
- def _send_configuration_request(self,
- sid,
- dcid,
- continuation=l2cap_packets.Continuation.END,
- options=[],
- payload=[]):
-
- config_request = l2cap_packets.ConfigurationRequestBuilder(
- sid, dcid, continuation, options)
-
- config_request_l2cap = l2cap_packets.BasicFrameBuilder(
- 1, config_request)
-
- config_request_l2cap = config_request_l2cap.Serialize()
- config_request_l2cap.extend(payload)
- config_request_l2cap[0] += len(payload)
- config_request_l2cap[6] += len(payload)
- self.cert_device.hci_acl_manager.SendAclData(
- acl_manager_facade.AclData(
- handle=self.cert_acl_handle,
- payload=bytes(config_request_l2cap)))
-
def _setup_link_from_cert(self):
self.dut.neighbor.EnablePageScan(
neighbor_facade.EnableMsg(enabled=True))
Verify the IUT is able to receive configuration requests that have the continuation flag set.
"""
cert_acl_handle = self._setup_link_from_cert()
- with EventCallbackStream(
- self.cert_device.hci_acl_manager.FetchAclData(
- empty_proto.Empty())) as cert_acl_data_stream:
- cert_acl_data_asserts = EventAsserts(cert_acl_data_stream)
- scid = 0x41
- psm = 0x33
- cert_acl_data_stream.register_callback(self._handle_control_packet)
-
- # Send configuration request with CONTINUE
- self.on_connection_response = lambda log: self._on_connection_response_use_mtu(log, continuation=l2cap_packets.Continuation.CONTINUE, mtu_value=48)
-
- self._open_channel(cert_acl_data_stream, 1, cert_acl_handle, scid,
- psm)
- cert_acl_data_asserts.assert_event_occurs(
- self.is_correct_configuration_response)
- flush_timeout_option = l2cap_packets.FlushTimeoutConfigurationOption(
- )
- flush_timeout_option.flush_timeout = 65535
- option = [flush_timeout_option]
- self._send_configuration_request(
- 3, self.scid_to_dcid[scid], options=option)
- cert_acl_data_asserts.assert_event_occurs(
- self.is_correct_configuration_response)
+
+ # Send configuration request with CONTINUE
+ self.cert_l2cap.reply_with_continuation_flag()
+
+ (dut_channel, cert_channel) = self._open_unvalidated_channel(
+ scid=0x41, psm=0x33)
+
+ assertThat(self.cert_l2cap.get_control_channel()).emits(
+ L2capMatchers.ConfigurationResponse(), at_least_times=2)
def test_retry_config_after_rejection(self):
"""