Merge "gd: Implement cert test for Scanning module" (398a7116) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/matchers.py

+9 −1

Original line number	Diff line number	Diff line
		@@ -211,7 +211,7 @@ class HciMatchers(object):
		class AdvertisingMatchers(object):

		@staticmethod
		def CallbackMsg(type, advertiser_id=None, status=None, data=None):
		def AdvertisingCallbackMsg(type, advertiser_id=None, status=None, data=None):
		return lambda event: True if event.message_type == type and (advertiser_id == None or advertiser_id == event.advertiser_id) \
		and (status == None or status == event.status) and (data == None or data == event.data) else False

		@@ -221,6 +221,14 @@ class AdvertisingMatchers(object):
		and (address == None or address == event.address) else False


		class ScanningMatchers(object):

		@staticmethod
		def ScanningCallbackMsg(type, status=None, data=None):
		return lambda event: True if event.message_type == type and (status == None or status == event.status) \
		and (data == None or data == event.data) else False


		class NeighborMatchers(object):

		@staticmethod

system/gd/hci/cert/le_advertising_manager_test_lib.py

+23 −20

Original line number	Diff line number	Diff line
		@@ -32,8 +32,8 @@ from hci.facade import hci_facade_pb2 as hci_facade
		from hci.facade import \
		le_advertising_manager_facade_pb2 as le_advertising_facade
		from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade
		from hci.facade.le_advertising_manager_facade_pb2 import AdvertisingCallbackMsgType
		from hci.facade.le_advertising_manager_facade_pb2 import AdvertisingStatus
		from hci.facade.le_advertising_manager_facade_pb2 import CallbackMsgType


		class LeAdvertisingManagerTestBase():
		@@ -171,8 +171,9 @@ class LeAdvertisingManagerTestBase():
		self.set_address_policy_with_static_address()
		create_response = self.create_advertiser()
		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_SET_STARTED, create_response.advertiser_id,
		AdvertisingStatus.SUCCESS, 0x00))
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_SET_STARTED,
		create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS,
		0x00))

		def test_enable_advertiser_callback(self):
		self.set_address_policy_with_static_address()
		@@ -182,8 +183,9 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.EnableAdvertiser(enable_advertiser_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_ENABLED, create_response.advertiser_id,
		AdvertisingStatus.SUCCESS, 0x01))
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_ENABLED,
		create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS,
		0x01))

		def test_disable_advertiser_callback(self):
		self.set_address_policy_with_static_address()
		@@ -193,8 +195,9 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.EnableAdvertiser(disable_advertiser_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_ENABLED, create_response.advertiser_id,
		AdvertisingStatus.SUCCESS, 0x00))
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_ENABLED,
		create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS,
		0x00))

		def test_set_advertising_data_callback(self):
		self.set_address_policy_with_static_address()
		@@ -209,8 +212,8 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.SetData(set_data_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_DATA_SET, create_response.advertiser_id,
		AdvertisingStatus.SUCCESS))
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_DATA_SET,
		create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS))

		def test_set_scan_response_data_callback(self):
		self.set_address_policy_with_static_address()
		@@ -225,8 +228,8 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.SetData(set_data_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.SCAN_RESPONSE_DATA_SET, create_response.advertiser_id,
		AdvertisingStatus.SUCCESS))
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.SCAN_RESPONSE_DATA_SET,
		create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS))

		def test_set_parameters_callback(self):
		self.set_address_policy_with_static_address()
		@@ -250,8 +253,8 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.SetParameters(set_parameters_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.ADVERTISING_PARAMETERS_UPDATED,
		create_response.advertiser_id, AdvertisingStatus.SUCCESS))
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_PARAMETERS_UPDATED,
		create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS))

		def test_set_periodic_parameters_callback(self):
		self.set_address_policy_with_static_address()
		@@ -267,8 +270,8 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.SetPeriodicParameters(set_periodic_parameters_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.PERIODIC_ADVERTISING_PARAMETERS_UPDATED,
		create_response.advertiser_id))
		AdvertisingMatchers.AdvertisingCallbackMsg(
		AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_PARAMETERS_UPDATED, create_response.advertiser_id))

		def test_set_periodic_data_callback(self):
		self.set_address_policy_with_static_address()
		@@ -283,7 +286,7 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.SetPeriodicData(set_periodic_data_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.PERIODIC_ADVERTISING_DATA_SET,
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_DATA_SET,
		create_response.advertiser_id))

		def test_enable_periodic_advertising_callback(self):
		@@ -294,7 +297,7 @@ class LeAdvertisingManagerTestBase():
		self.dut.hci_le_advertising_manager.EnablePeriodicAdvertising(enable_periodic_advertising_request)

		assertThat(self.dut.callback_event_stream).emits(
		AdvertisingMatchers.CallbackMsg(CallbackMsgType.PERIODIC_ADVERTISING_ENABLED,
		AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_ENABLED,
		create_response.advertiser_id))

		def test_get_own_address(self):
		@@ -306,5 +309,5 @@ class LeAdvertisingManagerTestBase():
		address_with_type = common.BluetoothAddressWithType(
		address=common.BluetoothAddress(address=bytes(b'd0:05:04:03:02:01')), type=common.RANDOM_DEVICE_ADDRESS)
		assertThat(self.dut.address_event_stream).emits(
		AdvertisingMatchers.AddressMsg(CallbackMsgType.OWN_ADDRESS_READ, create_response.advertiser_id,
		AdvertisingMatchers.AddressMsg(AdvertisingCallbackMsgType.OWN_ADDRESS_READ, create_response.advertiser_id,
		address_with_type))

system/gd/hci/cert/le_scanning_manager_test.py

+8 −0

Original line number	Diff line number	Diff line
		@@ -22,3 +22,11 @@ class LeScanningManagerTest(GdBaseTestClass, LeScanningManagerTestBase):

		def setup_class(self):
		GdBaseTestClass.setup_class(self, dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')

		def setup_test(self):
		GdBaseTestClass.setup_test(self)
		LeScanningManagerTestBase.setup_test(self, self.cert)

		def teardown_test(self):
		LeScanningManagerTestBase.teardown_test(self)
		GdBaseTestClass.teardown_test(self)

system/gd/hci/cert/le_scanning_manager_test_lib.py

+151 −29

Original line number	Diff line number	Diff line
		@@ -18,19 +18,40 @@ import os
		import sys
		import logging

		from bluetooth_packets_python3 import hci_packets
		from cert.closable import safeClose
		from cert.event_stream import EventStream
		from google.protobuf import empty_pb2 as empty_proto
		from cert.matchers import ScanningMatchers
		from cert.truth import assertThat
		from facade import common_pb2 as common
		from facade import rootservice_pb2 as facade_rootservice
		from google.protobuf import empty_pb2 as empty_proto
		from hci.facade import hci_facade_pb2 as hci_facade
		from hci.facade import le_scanning_manager_facade_pb2 as le_scanning_facade
		from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade
		from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade
		from bluetooth_packets_python3 import hci_packets
		from facade import common_pb2 as common
		from hci.facade.le_scanning_manager_facade_pb2 import ScanningCallbackMsgType
		from hci.facade.le_scanning_manager_facade_pb2 import ScanningStatus


		class LeScanningManagerTestBase():

		def setup_test(self, cert):
		self.dut.callback_event_stream = EventStream(
		self.dut.hci_le_scanning_manager.FetchCallbackEvents(empty_proto.Empty()))
		self.dut.advertising_report_stream = EventStream(
		self.dut.hci_le_scanning_manager.FetchAdvertisingReports(empty_proto.Empty()))

		def teardown_test(self):
		if self.dut.callback_event_stream is not None:
		safeClose(self.dut.callback_event_stream)
		else:
		logging.info("DUT: Callback Event Stream is None!")
		if self.dut.advertising_report_stream is not None:
		safeClose(self.dut.advertising_report_stream)
		else:
		logging.info("DUT: Advertising Report Stream is None!")

		def register_for_event(self, event_code):
		msg = hci_facade.EventCodeMsg(code=int(event_code))
		self.cert.hci.RegisterEventHandler(msg)
		@@ -47,7 +68,7 @@ class LeScanningManagerTestBase():
		else:
		self.cert.hci.EnqueueCommandWithStatus(cmd)

		def test_le_ad_scan_dut_scans(self):
		def set_address_policy_with_static_address(self):
		privacy_policy = le_initiator_address_facade.PrivacyPolicy(
		address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
		address_with_type=common.BluetoothAddressWithType(
		@@ -66,10 +87,9 @@ class LeScanningManagerTestBase():
		minimum_rotation_time=0,
		maximum_rotation_time=0)
		self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(cert_privacy_policy)
		with EventStream(
		# DUT Scans
		self.dut.hci_le_scanning_manager.StartScan(empty_proto.Empty())) as advertising_event_stream:

		def test_le_ad_scan_dut_scans(self):
		self.set_address_policy_with_static_address()
		# CERT Advertises
		gap_name = hci_packets.GapData()
		gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
		@@ -87,7 +107,109 @@ class LeScanningManagerTestBase():

		create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)

		advertising_event_stream.assert_event_occurs(lambda packet: b'Im_The_CERT' in packet.event)
		scan_request = le_scanning_facade.ScanRequest(start=True)
		self.dut.hci_le_scanning_manager.Scan(scan_request)

		self.dut.advertising_report_stream.assert_event_occurs(lambda packet: b'Im_The_CERT' in packet.event)

		remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)
		self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)

		def test_register_scanner_callback(self):
		self.set_address_policy_with_static_address()
		register_request = le_scanning_facade.RegisterScannerRequest(uuid=123)
		self.dut.hci_le_scanning_manager.RegisterScanner(register_request)
		assertThat(self.dut.callback_event_stream).emits(
		ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SCANNER_REGISTERED,
		ScanningStatus.SCAN_SUCCESS, 123))

		def test_register_scanner_with_same_uuid(self):
		self.set_address_policy_with_static_address()
		register_request = le_scanning_facade.RegisterScannerRequest(uuid=123)
		self.dut.hci_le_scanning_manager.RegisterScanner(register_request)
		assertThat(self.dut.callback_event_stream).emits(
		ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SCANNER_REGISTERED,
		ScanningStatus.SCAN_SUCCESS, 123))
		self.dut.hci_le_scanning_manager.RegisterScanner(register_request)
		assertThat(self.dut.callback_event_stream).emits(
		ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SCANNER_REGISTERED,
		ScanningStatus.SCAN_INTERNAL_ERROR, 123))

		def test_set_scan_parameters_callback(self):
		self.set_address_policy_with_static_address()
		set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(
		scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.ACTIVE, scan_interval=0x10, scan_window=0x04)
		self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)

		assertThat(self.dut.callback_event_stream).emits(
		ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SET_SCANNER_PARAMETER_COMPLETE,
		ScanningStatus.SCAN_SUCCESS, 0x01))

		def test_set_scan_parameters_with_invalid_parameter(self):
		self.set_address_policy_with_static_address()
		set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(
		scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.ACTIVE, scan_interval=0x00, scan_window=0x00)
		self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)

		assertThat(self.dut.callback_event_stream).emits(
		ScanningMatchers.ScanningCallbackMsg(ScanningCallbackMsgType.SET_SCANNER_PARAMETER_COMPLETE,
		ScanningStatus.SCAN_ILLEGAL_PARAMETER))

		def test_active_scan(self):
		self.set_address_policy_with_static_address()
		# CERT Advertises
		gap_name = hci_packets.GapData()
		gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
		gap_name.data = list(bytes(b'Scan response data'))
		gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
		config = le_advertising_facade.AdvertisingConfig(
		scan_response=[gap_data],
		interval_min=512,
		interval_max=768,
		advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
		own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
		channel_map=7,
		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)

		set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(
		scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.ACTIVE, scan_interval=0x10, scan_window=0x04)
		self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)
		scan_request = le_scanning_facade.ScanRequest(start=True)
		self.dut.hci_le_scanning_manager.Scan(scan_request)

		self.dut.advertising_report_stream.assert_event_occurs(lambda packet: b'Scan response data' in packet.event)

		remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)
		self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)

		def test_passive_scan(self):
		self.set_address_policy_with_static_address()
		# CERT Advertises
		gap_name = hci_packets.GapData()
		gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
		gap_name.data = list(bytes(b'Scan response data'))
		gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
		config = le_advertising_facade.AdvertisingConfig(
		scan_response=[gap_data],
		interval_min=512,
		interval_max=768,
		advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
		own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
		channel_map=7,
		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)

		set_scan_parameters_request = le_scanning_facade.SetScanParametersRequest(
		scanner_id=0x01, scan_type=le_scanning_facade.LeScanType.PASSIVE, scan_interval=0x10, scan_window=0x04)
		self.dut.hci_le_scanning_manager.SetScanParameters(set_scan_parameters_request)
		scan_request = le_scanning_facade.ScanRequest(start=True)
		self.dut.hci_le_scanning_manager.Scan(scan_request)

		self.dut.advertising_report_stream.assert_event_occurs_at_most(
		lambda packet: b'Scan response data' in packet.event, 0)

		remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)
		self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)
		No newline at end of file

system/gd/hci/cert/le_scanning_with_security_test_lib.py

+6 −1

Original line number	Diff line number	Diff line
		@@ -20,6 +20,7 @@ from google.protobuf import empty_pb2 as empty_proto
		from hci.facade import hci_facade_pb2 as hci_facade
		from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade
		from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade
		from hci.facade import le_scanning_manager_facade_pb2 as le_scanning_facade
		from bluetooth_packets_python3 import hci_packets
		from facade import common_pb2 as common

		@@ -57,7 +58,8 @@ class LeScanningWithSecurityTestBase():
		self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(cert_privacy_policy)
		with EventStream(
		# DUT Scans
		self.dut.hci_le_scanning_manager.StartScan(empty_proto.Empty())) as advertising_event_stream:
		self.dut.hci_le_scanning_manager.FetchAdvertisingReports(
		empty_proto.Empty())) as advertising_event_stream:

		# CERT Advertises
		gap_name = hci_packets.GapData()
		@@ -76,6 +78,9 @@ class LeScanningWithSecurityTestBase():

		create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)

		scan_request = le_scanning_facade.ScanRequest(start=True)
		self.dut.hci_le_scanning_manager.Scan(scan_request)

		assertThat(advertising_event_stream).emits(lambda packet: b'Im_The_CERT' in packet.event)

		remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)