self._device,
1,
1,
- self.get_acl_stream(),
+ self._get_acl_stream(),
self._acl,
control_channel=None)
- self.get_acl_stream().register_callback(self._handle_control_packet)
+ self._get_acl_stream().register_callback(self._handle_control_packet)
def open_channel(self, signal_id, psm, scid):
self.control_channel.send(
assertThat(self.control_channel).emits(response)
return CertL2capChannel(self._device, scid,
response.get().GetDestinationCid(),
- self.get_acl_stream(), self._acl,
+ self._get_acl_stream(), self._acl,
self.control_channel)
# prefer to use channel abstraction instead, if at all possible
def get_control_channel(self):
return self.control_channel
- # temporary until clients migrated
- def get_acl_stream(self):
+ def _get_acl_stream(self):
return self._acl_manager.get_acl_stream()
- # temporary until clients migrated
- def get_acl(self):
- return self._acl
-
- # temporary until clients migrated
- def get_dcid(self, scid):
- return self.scid_to_dcid[scid]
-
def turn_on_ertm(self, tx_window_size=10, max_transmit=20):
self.ertm_option = l2cap_packets.RetransmissionAndFlowControlConfigurationOption(
)
def setup_test(self):
super().setup_test()
- self.dut.address = self.dut.hci_controller.GetMacAddress(
+ self.dut.address = self.dut.hci_controller.GetMacAddressSimple()
+ self.cert.address = self.cert.controller_read_only_property.ReadLocalAddress(
empty_proto.Empty()).address
- cert_address = self.cert.controller_read_only_property.ReadLocalAddress(
- empty_proto.Empty()).address
- self.cert.address = cert_address
- self.dut_address = common_pb2.BluetoothAddress(address=self.dut.address)
self.cert_address = common_pb2.BluetoothAddress(
address=self.cert.address)
- self.dut.neighbor.EnablePageScan(
- neighbor_facade.EnableMsg(enabled=True))
-
self.dut_l2cap = PyL2cap(self.dut)
self.cert_l2cap = CertL2cap(self.cert)
- self.cert_acl = None
def teardown_test(self):
self.cert_l2cap.close()
super().teardown_test()
def cert_send_b_frame(self, b_frame):
- self.cert_acl.send(b_frame.Serialize())
+ self.cert_l2cap.send_acl(b_frame)
def _setup_link_from_cert(self):
-
self.dut.neighbor.EnablePageScan(
neighbor_facade.EnableMsg(enabled=True))
self.cert_l2cap.connect_acl(self.dut.address)
- self.cert_acl = self.cert_l2cap.get_acl()
def _open_unvalidated_channel(self,
signal_id=1,
self._open_unvalidated_channel(scid=0x41, psm=0x33)
assertThat(self.cert_l2cap.get_control_channel()).emits(
- L2capMatchers.ConfigurationResponse())
- assertThat(self.cert_l2cap.get_control_channel()).emits(
- L2capMatchers.ConfigurationRequest(), at_least_times=2)
+ L2capMatchers.ConfigurationResponse(),
+ L2capMatchers.ConfigurationRequest(),
+ L2capMatchers.ConfigurationRequest()).inAnyOrder()
def test_config_unknown_options_with_hint(self):
"""