Merge "LE L2CAP cert tests: don't repeat the address" (b22c20ff) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/py_l2cap.py

+12 −18

Original line number	Diff line number	Diff line
		@@ -119,8 +119,9 @@ class PyLeL2capFixedChannel(IEventStream):

		class PyLeL2capDynamicChannel(IEventStream):

		def __init__(self, device, psm, l2cap_stream):
		def __init__(self, device, cert_address, psm, l2cap_stream):
		self._device = device
		self._cert_address = cert_address
		self._psm = psm
		self._le_l2cap_stream = l2cap_stream
		self._our_le_l2cap_view = FilteringEventStream(
		@@ -138,10 +139,7 @@ class PyLeL2capDynamicChannel(IEventStream):
		def close_channel(self):
		self._device.l2cap_le.CloseDynamicChannel(
		l2cap_le_facade_pb2.CloseDynamicChannelRequest(
		remote=common.BluetoothAddressWithType(
		address=common.BluetoothAddress(
		address=b"22:33:ff:ff:11:00")),
		psm=self._psm))
		remote=self._cert_address, psm=self._psm))


		class _CreditBasedConnectionResponseFutureWrapper(object):
		@@ -150,9 +148,10 @@ class _CreditBasedConnectionResponseFutureWrapper(object):
		create the corresponding PyLeL2capDynamicChannel object later
		"""

		def __init__(self, grpc_response_future, device, psm, le_l2cap_stream):
		def __init__(self, grpc_response_future, device, cert_address, psm, le_l2cap_stream):
		self._grpc_response_future = grpc_response_future
		self._device = device
		self._cert_address = cert_address
		self._psm = psm
		self._le_l2cap_stream = le_l2cap_stream

		@@ -163,7 +162,7 @@ class _CreditBasedConnectionResponseFutureWrapper(object):
		def get_channel(self):
		assertThat(self.get_status()).isEqualTo(
		l2cap_packets.LeCreditBasedConnectionResponseResult.SUCCESS)
		return PyLeL2capDynamicChannel(self._device, self._psm,
		return PyLeL2capDynamicChannel(self._device, self._cert_address, self._psm,
		self._le_l2cap_stream)


		@@ -185,27 +184,22 @@ class PyLeL2cap(Closable):
		def get_fixed_channel(self, cid=4):
		return PyLeL2capFixedChannel(self._device, cid, self._le_l2cap_stream)

		def register_coc(self, psm=0x33):
		def register_coc(self, cert_address, psm=0x33):
		self._device.l2cap_le.SetDynamicChannel(
		l2cap_le_facade_pb2.SetEnableDynamicChannelRequest(
		psm=psm, enable=True))
		return PyLeL2capDynamicChannel(self._device, psm, self._le_l2cap_stream)
		return PyLeL2capDynamicChannel(self._device, cert_address, psm, self._le_l2cap_stream)

		def connect_coc_to_cert(self, psm=0x33):
		def connect_coc_to_cert(self, cert_address, psm=0x33):
		"""
		Send open LE COC request to CERT. Get a future for connection result, to be used after CERT accepts request
		"""
		self.register_coc(psm)
		# TODO: Update CERT device random address in ACL manager
		self.register_coc(cert_address, psm)
		response_future = self._device.l2cap_le.OpenDynamicChannel.future(
		l2cap_le_facade_pb2.OpenDynamicChannelRequest(
		psm=psm,
		remote=common.BluetoothAddressWithType(
		address=common.BluetoothAddress(
		address=b"22:33:ff:ff:11:00"))))
		l2cap_le_facade_pb2.OpenDynamicChannelRequest(psm=psm, remote=cert_address))

		return _CreditBasedConnectionResponseFutureWrapper(
		response_future, self._device, psm, self._le_l2cap_stream)
		response_future, self._device, cert_address, psm, self._le_l2cap_stream)

		def update_connection_parameter(self,
		conn_interval_min=0x10,

system/gd/cert/py_le_acl_manager.py

+3 −4

Original line number	Diff line number	Diff line
		@@ -44,13 +44,12 @@ class PyLeAclManagerAclConnection(IEventStream, Closable):

		if remote_addr:
		remote_addr_bytes = bytes(
		remote_addr,
		'utf8') if type(remote_addr) is str else bytes(remote_addr)
		remote_addr.address.address,
		'utf8') if type(remote_addr.address.address) is str else bytes(remote_addr.address.address)
		self.connection_event_stream = EventStream(
		self.device.hci_le_acl_manager.CreateConnection(
		le_acl_manager_facade.LeConnectionMsg(
		address_type=int(
		hci_packets.AddressType.RANDOM_DEVICE_ADDRESS),
		address_type=int(remote_addr.type),
		address=remote_addr_bytes)))
		else:
		self.connection_event_stream = None

system/gd/l2cap/le/cert/dual_l2cap_test.py

+11 −6

Original line number	Diff line number	Diff line
		@@ -68,27 +68,32 @@ class DualL2capTest(GdBaseTestClass):
		gap_name.data = list(bytes(b'Im_The_DUT'))
		gap_data = le_advertising_facade.GapDataMsg(
		data=bytes(gap_name.Serialize()))
		self.dut_address = common.BluetoothAddressWithType(
		address = common.BluetoothAddress(address=bytes(b'0D:05:04:03:02:01')),
		type=common.RANDOM_DEVICE_ADDRESS)
		self.cert_address = common.BluetoothAddressWithType(
		address=common.BluetoothAddress(address=b"22:33:ff:ff:11:00"),
		type=common.RANDOM_DEVICE_ADDRESS)
		config = le_advertising_facade.AdvertisingConfig(
		advertisement=[gap_data],
		random_address=common.BluetoothAddress(
		address=bytes(b'0D:05:04:03:02:01')),
		random_address=self.dut_address.address,
		interval_min=512,
		interval_max=768,
		event_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
		address_type=common.RANDOM_DEVICE_ADDRESS,
		peer_address_type=common.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS,
		peer_address=common.BluetoothAddress(
		address=bytes(b'A6:A5:A4:A3:A2:A1')),
		address=bytes(b'00:00:00:00:00:00')),
		channel_map=7,
		filter_policy=le_advertising_facade.AdvertisingFilterPolicy.
		ALL_DEVICES)
		request = le_advertising_facade.CreateAdvertiserRequest(config=config)
		create_response = self.dut.hci_le_advertising_manager.CreateAdvertiser(
		request)
		self.cert_le_l2cap.connect_le_acl(bytes(b'0D:05:04:03:02:01'))
		self.cert_le_l2cap.connect_le_acl(self.dut_address)

		def _open_le_coc_from_dut(self, psm=0x33, our_scid=None):
		response_future = self.dut_le_l2cap.connect_coc_to_cert(psm)
		response_future = self.dut_le_l2cap.connect_coc_to_cert(self.cert_address, psm)
		cert_channel = self.cert_le_l2cap.verify_and_respond_open_channel_from_remote(
		psm=psm, our_scid=our_scid)
		dut_channel = response_future.get_channel()
		@@ -133,7 +138,7 @@ class DualL2capTest(GdBaseTestClass):
		mps=100,
		initial_credit=6):

		dut_channel = self.dut_le_l2cap.register_coc(psm)
		dut_channel = self.dut_le_l2cap.register_coc(self.cert_address, psm)
		cert_channel = self.cert_le_l2cap.open_channel(signal_id, psm, scid,
		mtu, mps, initial_credit)

system/gd/l2cap/le/cert/le_l2cap_test.py

+32 −26

Original line number	Diff line number	Diff line
		@@ -53,24 +53,27 @@ class LeL2capTest(GdBaseTestClass):
		gap_name.data = list(bytes(b'Im_The_DUT'))
		gap_data = le_advertising_facade.GapDataMsg(
		data=bytes(gap_name.Serialize()))
		self.dut_address = common.BluetoothAddressWithType(
		address = common.BluetoothAddress(address=bytes(b'0D:05:04:03:02:01')),
		type=common.RANDOM_DEVICE_ADDRESS)
		self.cert_address = common.BluetoothAddressWithType(
		address=common.BluetoothAddress(address=b"22:33:ff:ff:11:00"),
		type=common.RANDOM_DEVICE_ADDRESS)
		config = le_advertising_facade.AdvertisingConfig(
		advertisement=[gap_data],
		random_address=common.BluetoothAddress(
		address=bytes(b'0D:05:04:03:02:01')),
		random_address=self.dut_address.address,
		interval_min=512,
		interval_max=768,
		event_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
		address_type=common.RANDOM_DEVICE_ADDRESS,
		peer_address_type=common.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS,
		peer_address=common.BluetoothAddress(
		address=bytes(b'A6:A5:A4:A3:A2:A1')),
		peer_address_type=common.PUBLIC_DEVICE_ADDRESS,
		peer_address=common.BluetoothAddress(address=bytes(b'00:00:00:00:00:00')),
		channel_map=7,
		filter_policy=le_advertising_facade.AdvertisingFilterPolicy.
		ALL_DEVICES)
		filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
		request = le_advertising_facade.CreateAdvertiserRequest(config=config)
		create_response = self.dut.hci_le_advertising_manager.CreateAdvertiser(
		request)
		self.cert_l2cap.connect_le_acl(bytes(b'0D:05:04:03:02:01'))
		self.cert_l2cap.connect_le_acl(self.dut_address)

		def _set_link_from_dut_and_open_channel(self,
		signal_id=1,
		@@ -85,24 +88,27 @@ class LeL2capTest(GdBaseTestClass):
		gap_name.data = list(bytes(b'Im_The_DUT'))
		gap_data = le_advertising_facade.GapDataMsg(
		data=bytes(gap_name.Serialize()))
		self.dut_address = common.BluetoothAddressWithType(
		address = common.BluetoothAddress(address=bytes(b'0D:05:04:03:02:01')),
		type=common.RANDOM_DEVICE_ADDRESS)
		self.cert_address = common.BluetoothAddressWithType(
		address = common.BluetoothAddress(address=bytes(b'22:33:ff:ff:11:00')),
		type=common.RANDOM_DEVICE_ADDRESS)
		config = le_advertising_facade.AdvertisingConfig(
		advertisement=[gap_data],
		random_address=common.BluetoothAddress(
		address=bytes(b'22:33:ff:ff:11:00')),
		random_address=self.cert_address.address,
		interval_min=512,
		interval_max=768,
		event_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
		address_type=common.RANDOM_DEVICE_ADDRESS,
		peer_address_type=common.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS,
		peer_address=common.BluetoothAddress(
		address=bytes(b'0D:05:04:03:02:01')),
		peer_address_type=common.PUBLIC_DEVICE_ADDRESS,
		peer_address=common.BluetoothAddress(address=bytes(b'00:00:00:00:00:00')),
		channel_map=7,
		filter_policy=le_advertising_facade.AdvertisingFilterPolicy.
		ALL_DEVICES)
		filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
		request = le_advertising_facade.CreateAdvertiserRequest(config=config)
		create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(
		request)
		response_future = self.dut_l2cap.connect_coc_to_cert(psm)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm)
		self.cert_l2cap.wait_for_connection()
		# TODO: Currently we can only connect by using Dynamic channel API. Use fixed channel instead.
		cert_channel = self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		@@ -118,14 +124,14 @@ class LeL2capTest(GdBaseTestClass):
		mps=100,
		initial_credit=6):

		dut_channel = self.dut_l2cap.register_coc(psm)
		dut_channel = self.dut_l2cap.register_coc(self.cert_address, psm)
		cert_channel = self.cert_l2cap.open_channel(signal_id, psm, scid, mtu,
		mps, initial_credit)

		return (dut_channel, cert_channel)

		def _open_channel_from_dut(self, psm=0x33):
		response_future = self.dut_l2cap.connect_coc_to_cert(psm)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm)
		cert_channel = self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm)
		dut_channel = response_future.get_channel()
		@@ -357,7 +363,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that an IUT sending an LE Credit Based Connection Request to a legacy peer and receiving a Command Reject does not establish the channel.
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_reject_open_channel_from_remote(psm=0x33)
		assertThat(response_future.get_status()).isNotEqualTo(
		LeCreditBasedConnectionResponseResult.SUCCESS)
		@@ -397,7 +403,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that an IUT sending an LE Credit Based Connection Request on an unsupported LE_PSM will not establish a channel upon receiving an LE Credit Based Connection Response refusing the connection.
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm=0x33,
		result=LeCreditBasedConnectionResponseResult.LE_PSM_NOT_SUPPORTED)
		@@ -492,7 +498,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that the IUT does not establish the channel upon receipt of an LE Credit Based Connection Response indicating the connection was refused with Result “0x0005 – Connection Refused – Insufficient Authentication".
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm=0x33,
		result=LeCreditBasedConnectionResponseResult.
		@@ -508,7 +514,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that the IUT does not establish the channel upon receipt of an LE Credit Based Connection Response indicating the connection was refused with Result “0x0006 – Connection Refused – Insufficient Authorization”.
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm=0x33,
		result=LeCreditBasedConnectionResponseResult.
		@@ -526,7 +532,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x0009 – Connection refused – Invalid Source CID".
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm=0x33,
		result=LeCreditBasedConnectionResponseResult.INVALID_SOURCE_CID)
		@@ -544,7 +550,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x000A – Connection refused – Source CID already allocated".
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm=0x33,
		result=LeCreditBasedConnectionResponseResult.
		@@ -565,7 +571,7 @@ class LeL2capTest(GdBaseTestClass):
		self._setup_link_from_cert()
		(dut_channel, cert_channel) = self._open_channel_from_cert(
		psm=0x33, scid=0x0101)
		self.dut_l2cap.register_coc(psm=0x35)
		self.dut_l2cap.register_coc(self.cert_address, psm=0x35)
		self.cert_l2cap.get_control_channel().send(
		l2cap_packets.LeCreditBasedConnectionRequestBuilder(
		2, 0x35, 0x0101, 1000, 1000, 1000))
		@@ -582,7 +588,7 @@ class LeL2capTest(GdBaseTestClass):
		Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x000B – Connection refused – Unacceptable Parameters".
		"""
		self._setup_link_from_cert()
		response_future = self.dut_l2cap.connect_coc_to_cert(psm=0x33)
		response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
		self.cert_l2cap.verify_and_respond_open_channel_from_remote(
		psm=0x33,
		result=LeCreditBasedConnectionResponseResult.UNACCEPTABLE_PARAMETERS