src/ble_test.py - platform/test/vts-testcase/nbu - Git at Google

 """BLE tests."""

 import time


 from mobly import asserts
 from mobly import test_runner
 from mobly import utils
 from utils import android_base_test

 # Number of seconds for the target to stay BLE advertising.
 ADVERTISING_TIME = 120
 # The number of seconds to wait for receiving scan results.
 SCAN_TIMEOUT = 5
 # The number of seconds to wair for connection established.
 CONNECTION_TIMEOUT = 20
 # The number of seconds to wait before cancel connection.
 CANCEL_CONNECTION_WAIT_TIME = 0.1
 # UUID for test service.
 TEST_BLE_SERVICE_UUID = '0000fe23-0000-1000-8000-00805f9b34fb'
 # UUID for write characteristic.
 TEST_WRITE_UUID = '0000e632-0000-1000-8000-00805f9b34fb'
 # UUID for second write characteristic.
 TEST_SECOND_WRITE_UUID = '0000e633-0000-1000-8000-00805f9b34fb'
 # UUID for read test.
 TEST_READ_UUID = '0000e631-0000-1000-8000-00805f9b34fb'
 # UUID for second read characteristic.
 TEST_SECOND_READ_UUID = '0000e634-0000-1000-8000-00805f9b34fb'
 # UUID for third read characteristic.
 TEST_THIRD_READ_UUID = '0000e635-0000-1000-8000-00805f9b34fb'
 # UUID for scan response.
 TEST_SCAN_RESPONSE_UUID = '0000e639-0000-1000-8000-00805f9b34fb'
 # Advertise settings in json format for Ble Advertise.
 ADVERTISE_SETTINGS = {
     'AdvertiseMode': 'ADVERTISE_MODE_LOW_LATENCY',
     'Timeout': ADVERTISING_TIME * 1000,
     'Connectable': True,
     'TxPowerLevel': 'ADVERTISE_TX_POWER_ULTRA_LOW'
 }
 # Ramdom data to represent device stored in advertise data.
 DATA = utils.rand_ascii_str(16)
 # Random data for scan response.
 SCAN_RESPONSE_DATA = utils.rand_ascii_str(16)
 # Random data for read operation.
 READ_DATA = utils.rand_ascii_str(8)
 # Random data for second read operation.
 SECOND_READ_DATA = utils.rand_ascii_str(8)
 # Random data for third read operation.
 THIRD_READ_DATA = utils.rand_ascii_str(8)
 # Random data for write operation.
 WRITE_DATA = utils.rand_ascii_str(8)
 # Random data for second write operation.
 SECOND_WRITE_DATA = utils.rand_ascii_str(8)
 # Advertise data in json format for BLE advertise.
 ADVERTISE_DATA = {
     'IncludeDeviceName': False,
     'ServiceData': [{
         'UUID': TEST_BLE_SERVICE_UUID,
         'Data': DATA
     }]
 }
 # Advertise data in json format representing scan response for BLE advertise.
 SCAN_RESPONSE = {
     'InlcudeDeviceName':
         False,
     'ServiceData': [{
         'UUID': TEST_SCAN_RESPONSE_UUID,
         'Data': SCAN_RESPONSE_DATA
     }]
 }
 # Scan filter in json format for BLE scan.
 SCAN_FILTER = {'ServiceUuid': TEST_BLE_SERVICE_UUID}
 # Scan settings in json format for BLE scan.
 SCAN_SETTINGS = {'ScanMode': 'SCAN_MODE_LOW_LATENCY'}
 # Characteristics for write in json format.
 WRITE_CHARACTERISTIC = {
     'UUID': TEST_WRITE_UUID,
     'Property': 'PROPERTY_WRITE',
     'Permission': 'PERMISSION_WRITE'
 }
 SECOND_WRITE_CHARACTERISTIC = {
     'UUID': TEST_SECOND_WRITE_UUID,
     'Property': 'PROPERTY_WRITE',
     'Permission': 'PERMISSION_WRITE'
 }
 # Characteristics for read in json format.
 READ_CHARACTERISTIC = {
     'UUID': TEST_READ_UUID,
     'Property': 'PROPERTY_READ',
     'Permission': 'PERMISSION_READ',
     'Data': READ_DATA
 }
 SECOND_READ_CHARACTERISTIC = {
     'UUID': TEST_SECOND_READ_UUID,
     'Property': 'PROPERTY_READ',
     'Permission': 'PERMISSION_READ',
     'Data': SECOND_READ_DATA
 }
 THIRD_READ_CHARACTERISTIC = {
     'UUID': TEST_THIRD_READ_UUID,
     'Property': 'PROPERTY_READ',
     'Permission': 'PERMISSION_READ',
     'Data': THIRD_READ_DATA
 }
 # Service data in json format for Ble Server.
 SERVICE = {
     'UUID':
         TEST_BLE_SERVICE_UUID,
     'Type':
         'SERVICE_TYPE_PRIMARY',
     'Characteristics': [
         WRITE_CHARACTERISTIC, SECOND_WRITE_CHARACTERISTIC, READ_CHARACTERISTIC,
         SECOND_READ_CHARACTERISTIC, THIRD_READ_CHARACTERISTIC
     ]
 }
 # Macros for literal string.
 UUID = 'UUID'
 GATT_SUCCESS = 'GATT_SUCCESS'
 STATE = 'newState'
 STATUS = 'status'


 def IsRequiredScanResult(scan_result):
   result = scan_result.data['result']
   for service in result['ScanRecord']['Services']:
     if service[UUID] == TEST_BLE_SERVICE_UUID and service['Data'] == DATA:
       return True
   return False


 def Discover(scanner, advertiser):
   """Logic for BLE scan and advertise.

   Args:
     scanner: AndroidDevice. The device that starts ble scan to find target.
     advertiser: AndroidDevice. The device that keeps advertising so other
     devices acknowledge it.

   Steps:
     1. Advertiser starts advertising and gets a startSuccess callback.
     2. Scanner starts scanning and finds advertiser from scan results.

   Verifies:
     Advertiser is discovered within 5s by scanner.
   """
   advertiser.advertise_callback = advertiser.android.bleStartAdvertising(
       ADVERTISE_SETTINGS, ADVERTISE_DATA, SCAN_RESPONSE)
   scanner.scan_callback = scanner.android.bleStartScan([SCAN_FILTER],
                                                        SCAN_SETTINGS)
   advertiser.advertise_callback.waitAndGet('onStartSuccess', 30)
   advertiser.log.info('BLE advertising started')
   time.sleep(SCAN_TIMEOUT)
   scan_result = scanner.scan_callback.waitForEvent(
       'onScanResult', IsRequiredScanResult, SCAN_TIMEOUT)
   scan_success = False
   scan_response_found = False
   result = scan_result.data['result']
   for service in result['ScanRecord']['Services']:
     if service[UUID] == TEST_BLE_SERVICE_UUID and service['Data'] == DATA:
       scanner.connect_to_address = result['Device']['Address']
       scan_success = True
     if (service[UUID] == TEST_SCAN_RESPONSE_UUID and
         service['Data'] == SCAN_RESPONSE_DATA):
       scan_response_found = True
   asserts.assert_true(
       scan_success, 'Advertiser is not found inside %d seconds' % SCAN_TIMEOUT)
   asserts.assert_true(scan_response_found, 'Scan response is not found')
   scanner.log.info('Advertiser is found')


 def StopDiscover(scanner, advertiser):
   """Logic for stopping BLE scan and advertise.

   Args:
     scanner: AndroidDevice. The device that starts ble scan to find target.
     advertiser: AndroidDevice. The device that keeps advertising so other
     devices acknowledge it.

   Steps:
     1. Scanner stops scanning.
     2. Advertiser stops advertising.
   """
   scanner.android.bleStopScan(scanner.scan_callback.callback_id)
   scanner.log.info('BLE scanning stopped')
   advertiser.android.bleStopAdvertising(
       advertiser.advertise_callback.callback_id)
   advertiser.log.info('BLE advertising stopped')


 def Connect(client, server):
   """Logic for starting BLE client and server.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.
     server: AndroidDevice. The device that behaves as GATT server.

   Steps:
     1. Server starts and service added properly.
     2. Client connects to server via Gatt, connection completes with
     GATT_SUCCESS within some TIMEOUT, onConnectionStateChange/STATE_CONNECTED
     called EXACTLY once.

   Verifies:
     Client gets corresponding callback.
   """
   server.server_callback = server.android.bleStartServer([SERVICE])
   start_server_result = server.server_callback.waitAndGet('onServiceAdded', 30)
   asserts.assert_equal(start_server_result.data[STATUS], GATT_SUCCESS)
   uuids = [
       characteristic[UUID]
       for characteristic in start_server_result.data['Service'][
           'Characteristics']
   ]
   for uuid in [
       characteristic[UUID] for characteristic in SERVICE['Characteristics']
   ]:
     asserts.assert_true(uuid in uuids, 'Failed to find uuid %s.' % uuid)
   server.log.info('BLE server started')
   client.client_callback = client.android.bleConnectGatt(
       client.connect_to_address)
   time.sleep(CONNECTION_TIMEOUT)
   start_client_results = client.client_callback.getAll(
       'onConnectionStateChange')
   asserts.assert_equal(len(start_client_results), 1)
   for start_client_result in start_client_results:
     asserts.assert_equal(start_client_result.data[STATUS], GATT_SUCCESS)
     asserts.assert_equal(start_client_result.data[STATE], 'STATE_CONNECTED')
   client.log.info('BLE client connected')


 def CancelOpen(client, server):
   """Logic for BLE client cancel open and reconnect.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.
     server: AndroidDevice. The device that behaves as GATT server.

   Steps:
     1. Server starts and service added properly.
     2. Client stars to connect to server via Gatt, but the connection has not
     been established.
     3. Client calls disconnect to cancel the connection.
     4. Client connects to server, connection completes with GATT_SUCCESS within
     some TIMEOUT, onConnectionStateChange/STATE_CONNECTEDcalled EXACTLY once.

   Verifies:
     Client gets corresponding callback.
   """
   server.server_callback = server.android.bleStartServer([SERVICE])
   start_server_result = server.server_callback.waitAndGet('onServiceAdded', 30)
   asserts.assert_equal(start_server_result.data[STATUS], GATT_SUCCESS)
   uuids = [
       characteristic[UUID]
       for characteristic in start_server_result.data['Service'][
           'Characteristics']
   ]
   for uuid in [
       characteristic[UUID] for characteristic in SERVICE['Characteristics']
   ]:
     asserts.assert_true(uuid in uuids, 'Failed to find uuid %s.' % uuid)
   server.log.info('BLE server started')
   client.client_callback = client.android.bleConnectGatt(
       client.connect_to_address)
   time.sleep(CANCEL_CONNECTION_WAIT_TIME)
   start_client_results = client.client_callback.getAll(
       'onConnectionStateChange')
   if not start_client_results:
     client.android.bleDisconnect()
     client.log.info('BLE client cancel open')
     time.sleep(CANCEL_CONNECTION_WAIT_TIME)
     client.client_callback = client.android.bleConnectGatt(
         client.connect_to_address)
     time.sleep(CONNECTION_TIMEOUT)
     start_client_results = client.client_callback.getAll(
         'onConnectionStateChange')
     asserts.assert_equal(len(start_client_results), 1)
     for start_client_result in start_client_results:
       asserts.assert_equal(start_client_result.data[STATUS], GATT_SUCCESS)
       asserts.assert_equal(start_client_result.data[STATE], 'STATE_CONNECTED')
       client.log.info('BLE client connected')


 def Disconnect(client, server):
   """Logic for stopping BLE client and server.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.
     server: AndroidDevice. The device that behaves as GATT server.

   Steps:
     1. Client calls disconnect, gets a callback with STATE_DISCONNECTED
     and GATT_SUCCESS.
     2. Server closes.

   Verifies:
     Client gets corresponding callback.
   """
   client.android.bleDisconnect()
   stop_client_result = client.client_callback.waitAndGet(
       'onConnectionStateChange', 30)
   asserts.assert_equal(stop_client_result.data[STATUS], GATT_SUCCESS)
   asserts.assert_equal(stop_client_result.data[STATE], 'STATE_DISCONNECTED')
   client.log.info('BLE client disconnected')
   server.android.bleStopServer()
   server.log.info('BLE server stopped')


 def DiscoverServices(client):
   """Logic for BLE services discovery.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.

   Steps:
     1. Client successfully completes service discovery & gets
     onServicesDiscovered callback within some TIMEOUT, onServicesDiscovered/
     GATT_SUCCESS called EXACTLY once.
     2. Client discovers the readable and writable characteristics.

   Verifies:
     Client gets corresponding callback.
   """
   client.android.bleDiscoverServices()
   time.sleep(CONNECTION_TIMEOUT)
   discover_services_results = client.client_callback.getAll(
       'onServiceDiscovered')
   asserts.assert_equal(len(discover_services_results), 1)
   service_discovered = False
   asserts.assert_equal(discover_services_results[0].data[STATUS],
                        GATT_SUCCESS)
   for service in discover_services_results[0].data['Services']:
     if service['UUID'] == TEST_BLE_SERVICE_UUID:
       service_discovered = True
       uuids = [
           characteristic[UUID]
           for characteristic in service['Characteristics']
       ]
       for uuid in [
           characteristic[UUID]
           for characteristic in SERVICE['Characteristics']
       ]:
         asserts.assert_true(uuid in uuids, 'Failed to find uuid %s.' % uuid)
   asserts.assert_true(service_discovered,
                       'Failed to discover the customize service')
   client.log.info('BLE discover services finished')


 def ReadCharacteristic(client):
   """Logic for BLE characteristic read.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.

   Steps:
     1. Client reads a characteristic from server & gets true.
     2. Server calls sendResponse & client gets onCharacteristicRead.

   Verifies:
     Client gets corresponding callback.
   """
   read_operation_result = client.android.bleReadOperation(
       TEST_BLE_SERVICE_UUID, TEST_READ_UUID)
   asserts.assert_true(read_operation_result,
                       'BLE read operation failed to start')
   read_operation_result = client.client_callback.waitAndGet(
       'onCharacteristicRead', 30)
   asserts.assert_equal(read_operation_result.data[STATUS], GATT_SUCCESS)
   asserts.assert_equal(read_operation_result.data['Data'], READ_DATA)
   client.log.info('Read operation finished')
   read_operation_result = client.android.bleReadOperation(
       TEST_BLE_SERVICE_UUID, TEST_SECOND_READ_UUID)
   asserts.assert_true(read_operation_result,
                       'BLE read operation failed to start')
   read_operation_result = client.client_callback.waitAndGet(
       'onCharacteristicRead', 30)
   asserts.assert_equal(read_operation_result.data[STATUS], GATT_SUCCESS)
   asserts.assert_equal(read_operation_result.data['Data'], SECOND_READ_DATA)
   client.log.info('Second read operation finished')
   read_operation_result = client.android.bleReadOperation(
       TEST_BLE_SERVICE_UUID, TEST_THIRD_READ_UUID)
   asserts.assert_true(read_operation_result,
                       'BLE read operation failed to start')
   read_operation_result = client.client_callback.waitAndGet(
       'onCharacteristicRead', 30)
   asserts.assert_equal(read_operation_result.data[STATUS], GATT_SUCCESS)
   asserts.assert_equal(read_operation_result.data['Data'], THIRD_READ_DATA)
   client.log.info('Third read operation finished')


 def WriteCharacteristic(client, server):
   """Logic for BLE characteristic write.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.
     server: AndroidDevice. The device that behaves as GATT server.

   Steps:
     1. Client writes a characteristic to server & gets true.
     2. Server calls sendResponse & client gets onCharacteristicWrite.

   Verifies:
     Client gets corresponding callback.
   """
   write_operation_result = client.android.bleWriteOperation(
       TEST_BLE_SERVICE_UUID, TEST_WRITE_UUID, WRITE_DATA)
   asserts.assert_true(write_operation_result,
                       'BLE write operation failed to start')
   server_write_operation_result = server.server_callback.waitAndGet(
       'onCharacteristicWriteRequest', 30)
   asserts.assert_equal(server_write_operation_result.data['Data'], WRITE_DATA)
   client.client_callback.waitAndGet('onCharacteristicWrite', 30)
   client.log.info('Write operation finished')
   write_operation_result = client.android.bleWriteOperation(
       TEST_BLE_SERVICE_UUID, TEST_SECOND_WRITE_UUID, SECOND_WRITE_DATA)
   asserts.assert_true(write_operation_result,
                       'BLE write operation failed to start')
   server_write_operation_result = server.server_callback.waitAndGet(
       'onCharacteristicWriteRequest', 30)
   asserts.assert_equal(server_write_operation_result.data['Data'],
                        SECOND_WRITE_DATA)
   client.client_callback.waitAndGet('onCharacteristicWrite', 30)
   client.log.info('Second write operation finished')


 def ReliableWrite(client, server):
   """Logic for BLE reliable write.

   Args:
     client: AndroidDevice. The device that behaves as GATT client.
     server: AndroidDevice. The device that behaves as GATT server.

   Steps:
     1. Client calls beginReliableWrite & gets true.
     2. Client writes a characteristic to server & gets true.
     3. Server calls sendResponse & client gets onCharacteristicWrite.
     4. Client calls executeReliableWrite & gets true
     5. Server calls sendResponse & client gets onReliableWriteCompleted.

   Verifies:
     Client get corresponding callbacks.
   """
   begin_reliable_write_result = client.android.bleBeginReliableWrite()
   asserts.assert_true(begin_reliable_write_result,
                       'BLE reliable write failed to start')
   client.log.info('BLE reliable write started')
   WriteCharacteristic(client, server)
   execute_reliable_write_result = client.android.bleExecuteReliableWrite()
   asserts.assert_true(execute_reliable_write_result,
                       'BLE reliable write failed to execute')
   client.log.info('BLE reliable write execute started')
   client.client_callback.waitAndGet('onReliableWriteCompleted', 30)
   client.log.info('BLE reliable write finished')


 class BleTest(android_base_test.AndroidBaseTest):
   """BLE tests."""

   def setup_class(self):
     super(BleTest, self).setup_class()
     self.initiator = self.dut_a
     self.receiver = self.dut_b
     # Sets the tag that represents this device in logs.
     # The device used to scan BLE devices and behave as a GATT client.
     self.initiator.debug_tag = 'initiator'
     # The device used to BLE advertise and behave as a GATT server.
     self.receiver.debug_tag = 'receiver'

   def setup_test(self):
     # Make sure bluetooth is on.
     self.initiator.android.btEnable()
     self.receiver.android.btEnable()

   def test_basic_ble_process(self):
     """Test for basic BLE process flow.

     Steps:
       1. Initiator discovers receiver.
       2. Initiator connects to receiver.
       3. Initiator discovers the BLE service receiver provided.
       4. Initiator reads a message from receiver.
       5. Initiator sends a message to receiver.
       6. Initiator disconnects from receiver.
       7. BLE scan and advertise stopped.

     Verifies:
       In each step, initiator and receiver get corresponding callbacks.
     """
     Discover(self.initiator, self.receiver)
     Connect(self.initiator, self.receiver)
     DiscoverServices(self.initiator)
     ReadCharacteristic(self.initiator)
     WriteCharacteristic(self.initiator, self.receiver)
     Disconnect(self.initiator, self.receiver)
     StopDiscover(self.initiator, self.receiver)

   def test_cancel_open_ble_process(self):
     """Test for BLE process flow involving cancel open.

     Steps:
       1. Initiator discovers receiver.
       2. Initiator initials connection to receiver, cancels connection and
       reconnects.
       3. Initiator discovers the BLE service receiver provided.
       6. Initiator disconnects from receiver.
       7. BLE scan and advertise stopped.

     Verifies:
       In each step, initiator and receiver get corresponding callbacks.
     """
     Discover(self.initiator, self.receiver)
     CancelOpen(self.initiator, self.receiver)
     DiscoverServices(self.initiator)
     Disconnect(self.initiator, self.receiver)
     StopDiscover(self.initiator, self.receiver)

   def test_reliable_write_ble_process(self):
     """Test for BLE process flow involving reliable write.

     Steps:
       1. Initiator discovers receiver.
       2. Initiator connects to receiver.
       3. Initiator discovers the BLE service receiver provided.
       4. Initiator starts reliable write to receiver and finishes successfully.
       5. Initiator disconnects from receiver.
       6. BLE scan and advertise stopped.

     Verifies:
       In each step, initiator and receiver get corresponding callbacks.
     """
     Discover(self.initiator, self.receiver)
     Connect(self.initiator, self.receiver)
     DiscoverServices(self.initiator)
     ReliableWrite(self.initiator, self.receiver)
     Disconnect(self.initiator, self.receiver)
     StopDiscover(self.initiator, self.receiver)

   def teardown_test(self):
     # Turn Bluetooth off on both devices after test finishes.
     self.initiator.android.btDisable()
     self.receiver.android.btDisable()


 if __name__ == '__main__':
   test_runner.main()
	"""BLE tests."""

	import time



	from mobly import asserts
	from mobly import test_runner
	from mobly import utils
	from utils import android_base_test

	# Number of seconds for the target to stay BLE advertising.
	ADVERTISING_TIME = 120
	# The number of seconds to wait for receiving scan results.
	SCAN_TIMEOUT = 5
	# The number of seconds to wair for connection established.
	CONNECTION_TIMEOUT = 20
	# The number of seconds to wait before cancel connection.
	CANCEL_CONNECTION_WAIT_TIME = 0.1
	# UUID for test service.
	TEST_BLE_SERVICE_UUID = '0000fe23-0000-1000-8000-00805f9b34fb'
	# UUID for write characteristic.
	TEST_WRITE_UUID = '0000e632-0000-1000-8000-00805f9b34fb'
	# UUID for second write characteristic.
	TEST_SECOND_WRITE_UUID = '0000e633-0000-1000-8000-00805f9b34fb'
	# UUID for read test.
	TEST_READ_UUID = '0000e631-0000-1000-8000-00805f9b34fb'
	# UUID for second read characteristic.
	TEST_SECOND_READ_UUID = '0000e634-0000-1000-8000-00805f9b34fb'
	# UUID for third read characteristic.
	TEST_THIRD_READ_UUID = '0000e635-0000-1000-8000-00805f9b34fb'
	# UUID for scan response.
	TEST_SCAN_RESPONSE_UUID = '0000e639-0000-1000-8000-00805f9b34fb'
	# Advertise settings in json format for Ble Advertise.
	ADVERTISE_SETTINGS = {
	'AdvertiseMode': 'ADVERTISE_MODE_LOW_LATENCY',
	'Timeout': ADVERTISING_TIME * 1000,
	'Connectable': True,
	'TxPowerLevel': 'ADVERTISE_TX_POWER_ULTRA_LOW'
	}
	# Ramdom data to represent device stored in advertise data.
	DATA = utils.rand_ascii_str(16)
	# Random data for scan response.
	SCAN_RESPONSE_DATA = utils.rand_ascii_str(16)
	# Random data for read operation.
	READ_DATA = utils.rand_ascii_str(8)
	# Random data for second read operation.
	SECOND_READ_DATA = utils.rand_ascii_str(8)
	# Random data for third read operation.
	THIRD_READ_DATA = utils.rand_ascii_str(8)
	# Random data for write operation.
	WRITE_DATA = utils.rand_ascii_str(8)
	# Random data for second write operation.
	SECOND_WRITE_DATA = utils.rand_ascii_str(8)
	# Advertise data in json format for BLE advertise.
	ADVERTISE_DATA = {
	'IncludeDeviceName': False,
	'ServiceData': [{
	'UUID': TEST_BLE_SERVICE_UUID,
	'Data': DATA
	}]
	}
	# Advertise data in json format representing scan response for BLE advertise.
	SCAN_RESPONSE = {
	'InlcudeDeviceName':
	False,
	'ServiceData': [{
	'UUID': TEST_SCAN_RESPONSE_UUID,
	'Data': SCAN_RESPONSE_DATA
	}]
	}
	# Scan filter in json format for BLE scan.
	SCAN_FILTER = {'ServiceUuid': TEST_BLE_SERVICE_UUID}
	# Scan settings in json format for BLE scan.
	SCAN_SETTINGS = {'ScanMode': 'SCAN_MODE_LOW_LATENCY'}
	# Characteristics for write in json format.
	WRITE_CHARACTERISTIC = {
	'UUID': TEST_WRITE_UUID,
	'Property': 'PROPERTY_WRITE',
	'Permission': 'PERMISSION_WRITE'
	}
	SECOND_WRITE_CHARACTERISTIC = {
	'UUID': TEST_SECOND_WRITE_UUID,
	'Property': 'PROPERTY_WRITE',
	'Permission': 'PERMISSION_WRITE'
	}
	# Characteristics for read in json format.
	READ_CHARACTERISTIC = {
	'UUID': TEST_READ_UUID,
	'Property': 'PROPERTY_READ',
	'Permission': 'PERMISSION_READ',
	'Data': READ_DATA
	}
	SECOND_READ_CHARACTERISTIC = {
	'UUID': TEST_SECOND_READ_UUID,
	'Property': 'PROPERTY_READ',
	'Permission': 'PERMISSION_READ',
	'Data': SECOND_READ_DATA
	}
	THIRD_READ_CHARACTERISTIC = {
	'UUID': TEST_THIRD_READ_UUID,
	'Property': 'PROPERTY_READ',
	'Permission': 'PERMISSION_READ',
	'Data': THIRD_READ_DATA
	}
	# Service data in json format for Ble Server.
	SERVICE = {
	'UUID':
	TEST_BLE_SERVICE_UUID,
	'Type':
	'SERVICE_TYPE_PRIMARY',
	'Characteristics': [
	WRITE_CHARACTERISTIC, SECOND_WRITE_CHARACTERISTIC, READ_CHARACTERISTIC,
	SECOND_READ_CHARACTERISTIC, THIRD_READ_CHARACTERISTIC
	]
	}
	# Macros for literal string.
	UUID = 'UUID'
	GATT_SUCCESS = 'GATT_SUCCESS'
	STATE = 'newState'
	STATUS = 'status'


	def IsRequiredScanResult(scan_result):
	result = scan_result.data['result']
	for service in result['ScanRecord']['Services']:
	if service[UUID] == TEST_BLE_SERVICE_UUID and service['Data'] == DATA:
	return True
	return False


	def Discover(scanner, advertiser):
	"""Logic for BLE scan and advertise.

	Args:
	scanner: AndroidDevice. The device that starts ble scan to find target.
	advertiser: AndroidDevice. The device that keeps advertising so other
	devices acknowledge it.

	Steps:
	1. Advertiser starts advertising and gets a startSuccess callback.
	2. Scanner starts scanning and finds advertiser from scan results.

	Verifies:
	Advertiser is discovered within 5s by scanner.
	"""
	advertiser.advertise_callback = advertiser.android.bleStartAdvertising(
	ADVERTISE_SETTINGS, ADVERTISE_DATA, SCAN_RESPONSE)
	scanner.scan_callback = scanner.android.bleStartScan([SCAN_FILTER],
	SCAN_SETTINGS)
	advertiser.advertise_callback.waitAndGet('onStartSuccess', 30)
	advertiser.log.info('BLE advertising started')
	time.sleep(SCAN_TIMEOUT)
	scan_result = scanner.scan_callback.waitForEvent(
	'onScanResult', IsRequiredScanResult, SCAN_TIMEOUT)
	scan_success = False
	scan_response_found = False
	result = scan_result.data['result']
	for service in result['ScanRecord']['Services']:
	if service[UUID] == TEST_BLE_SERVICE_UUID and service['Data'] == DATA:
	scanner.connect_to_address = result['Device']['Address']
	scan_success = True
	if (service[UUID] == TEST_SCAN_RESPONSE_UUID and
	service['Data'] == SCAN_RESPONSE_DATA):
	scan_response_found = True
	asserts.assert_true(
	scan_success, 'Advertiser is not found inside %d seconds' % SCAN_TIMEOUT)
	asserts.assert_true(scan_response_found, 'Scan response is not found')
	scanner.log.info('Advertiser is found')


	def StopDiscover(scanner, advertiser):
	"""Logic for stopping BLE scan and advertise.

	Args:
	scanner: AndroidDevice. The device that starts ble scan to find target.
	advertiser: AndroidDevice. The device that keeps advertising so other
	devices acknowledge it.

	Steps:
	1. Scanner stops scanning.
	2. Advertiser stops advertising.
	"""
	scanner.android.bleStopScan(scanner.scan_callback.callback_id)
	scanner.log.info('BLE scanning stopped')
	advertiser.android.bleStopAdvertising(
	advertiser.advertise_callback.callback_id)
	advertiser.log.info('BLE advertising stopped')


	def Connect(client, server):
	"""Logic for starting BLE client and server.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.
	server: AndroidDevice. The device that behaves as GATT server.

	Steps:
	1. Server starts and service added properly.
	2. Client connects to server via Gatt, connection completes with
	GATT_SUCCESS within some TIMEOUT, onConnectionStateChange/STATE_CONNECTED
	called EXACTLY once.

	Verifies:
	Client gets corresponding callback.
	"""
	server.server_callback = server.android.bleStartServer([SERVICE])
	start_server_result = server.server_callback.waitAndGet('onServiceAdded', 30)
	asserts.assert_equal(start_server_result.data[STATUS], GATT_SUCCESS)
	uuids = [
	characteristic[UUID]
	for characteristic in start_server_result.data['Service'][
	'Characteristics']
	]
	for uuid in [
	characteristic[UUID] for characteristic in SERVICE['Characteristics']
	]:
	asserts.assert_true(uuid in uuids, 'Failed to find uuid %s.' % uuid)
	server.log.info('BLE server started')
	client.client_callback = client.android.bleConnectGatt(
	client.connect_to_address)
	time.sleep(CONNECTION_TIMEOUT)
	start_client_results = client.client_callback.getAll(
	'onConnectionStateChange')
	asserts.assert_equal(len(start_client_results), 1)
	for start_client_result in start_client_results:
	asserts.assert_equal(start_client_result.data[STATUS], GATT_SUCCESS)
	asserts.assert_equal(start_client_result.data[STATE], 'STATE_CONNECTED')
	client.log.info('BLE client connected')


	def CancelOpen(client, server):
	"""Logic for BLE client cancel open and reconnect.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.
	server: AndroidDevice. The device that behaves as GATT server.

	Steps:
	1. Server starts and service added properly.
	2. Client stars to connect to server via Gatt, but the connection has not
	been established.
	3. Client calls disconnect to cancel the connection.
	4. Client connects to server, connection completes with GATT_SUCCESS within
	some TIMEOUT, onConnectionStateChange/STATE_CONNECTEDcalled EXACTLY once.

	Verifies:
	Client gets corresponding callback.
	"""
	server.server_callback = server.android.bleStartServer([SERVICE])
	start_server_result = server.server_callback.waitAndGet('onServiceAdded', 30)
	asserts.assert_equal(start_server_result.data[STATUS], GATT_SUCCESS)
	uuids = [
	characteristic[UUID]
	for characteristic in start_server_result.data['Service'][
	'Characteristics']
	]
	for uuid in [
	characteristic[UUID] for characteristic in SERVICE['Characteristics']
	]:
	asserts.assert_true(uuid in uuids, 'Failed to find uuid %s.' % uuid)
	server.log.info('BLE server started')
	client.client_callback = client.android.bleConnectGatt(
	client.connect_to_address)
	time.sleep(CANCEL_CONNECTION_WAIT_TIME)
	start_client_results = client.client_callback.getAll(
	'onConnectionStateChange')
	if not start_client_results:
	client.android.bleDisconnect()
	client.log.info('BLE client cancel open')
	time.sleep(CANCEL_CONNECTION_WAIT_TIME)
	client.client_callback = client.android.bleConnectGatt(
	client.connect_to_address)
	time.sleep(CONNECTION_TIMEOUT)
	start_client_results = client.client_callback.getAll(
	'onConnectionStateChange')
	asserts.assert_equal(len(start_client_results), 1)
	for start_client_result in start_client_results:
	asserts.assert_equal(start_client_result.data[STATUS], GATT_SUCCESS)
	asserts.assert_equal(start_client_result.data[STATE], 'STATE_CONNECTED')
	client.log.info('BLE client connected')


	def Disconnect(client, server):
	"""Logic for stopping BLE client and server.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.
	server: AndroidDevice. The device that behaves as GATT server.

	Steps:
	1. Client calls disconnect, gets a callback with STATE_DISCONNECTED
	and GATT_SUCCESS.
	2. Server closes.

	Verifies:
	Client gets corresponding callback.
	"""
	client.android.bleDisconnect()
	stop_client_result = client.client_callback.waitAndGet(
	'onConnectionStateChange', 30)
	asserts.assert_equal(stop_client_result.data[STATUS], GATT_SUCCESS)
	asserts.assert_equal(stop_client_result.data[STATE], 'STATE_DISCONNECTED')
	client.log.info('BLE client disconnected')
	server.android.bleStopServer()
	server.log.info('BLE server stopped')


	def DiscoverServices(client):
	"""Logic for BLE services discovery.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.

	Steps:
	1. Client successfully completes service discovery & gets
	onServicesDiscovered callback within some TIMEOUT, onServicesDiscovered/
	GATT_SUCCESS called EXACTLY once.
	2. Client discovers the readable and writable characteristics.

	Verifies:
	Client gets corresponding callback.
	"""
	client.android.bleDiscoverServices()
	time.sleep(CONNECTION_TIMEOUT)
	discover_services_results = client.client_callback.getAll(
	'onServiceDiscovered')
	asserts.assert_equal(len(discover_services_results), 1)
	service_discovered = False
	asserts.assert_equal(discover_services_results[0].data[STATUS],
	GATT_SUCCESS)
	for service in discover_services_results[0].data['Services']:
	if service['UUID'] == TEST_BLE_SERVICE_UUID:
	service_discovered = True
	uuids = [
	characteristic[UUID]
	for characteristic in service['Characteristics']
	]
	for uuid in [
	characteristic[UUID]
	for characteristic in SERVICE['Characteristics']
	]:
	asserts.assert_true(uuid in uuids, 'Failed to find uuid %s.' % uuid)
	asserts.assert_true(service_discovered,
	'Failed to discover the customize service')
	client.log.info('BLE discover services finished')


	def ReadCharacteristic(client):
	"""Logic for BLE characteristic read.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.

	Steps:
	1. Client reads a characteristic from server & gets true.
	2. Server calls sendResponse & client gets onCharacteristicRead.

	Verifies:
	Client gets corresponding callback.
	"""
	read_operation_result = client.android.bleReadOperation(
	TEST_BLE_SERVICE_UUID, TEST_READ_UUID)
	asserts.assert_true(read_operation_result,
	'BLE read operation failed to start')
	read_operation_result = client.client_callback.waitAndGet(
	'onCharacteristicRead', 30)
	asserts.assert_equal(read_operation_result.data[STATUS], GATT_SUCCESS)
	asserts.assert_equal(read_operation_result.data['Data'], READ_DATA)
	client.log.info('Read operation finished')
	read_operation_result = client.android.bleReadOperation(
	TEST_BLE_SERVICE_UUID, TEST_SECOND_READ_UUID)
	asserts.assert_true(read_operation_result,
	'BLE read operation failed to start')
	read_operation_result = client.client_callback.waitAndGet(
	'onCharacteristicRead', 30)
	asserts.assert_equal(read_operation_result.data[STATUS], GATT_SUCCESS)
	asserts.assert_equal(read_operation_result.data['Data'], SECOND_READ_DATA)
	client.log.info('Second read operation finished')
	read_operation_result = client.android.bleReadOperation(
	TEST_BLE_SERVICE_UUID, TEST_THIRD_READ_UUID)
	asserts.assert_true(read_operation_result,
	'BLE read operation failed to start')
	read_operation_result = client.client_callback.waitAndGet(
	'onCharacteristicRead', 30)
	asserts.assert_equal(read_operation_result.data[STATUS], GATT_SUCCESS)
	asserts.assert_equal(read_operation_result.data['Data'], THIRD_READ_DATA)
	client.log.info('Third read operation finished')


	def WriteCharacteristic(client, server):
	"""Logic for BLE characteristic write.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.
	server: AndroidDevice. The device that behaves as GATT server.

	Steps:
	1. Client writes a characteristic to server & gets true.
	2. Server calls sendResponse & client gets onCharacteristicWrite.

	Verifies:
	Client gets corresponding callback.
	"""
	write_operation_result = client.android.bleWriteOperation(
	TEST_BLE_SERVICE_UUID, TEST_WRITE_UUID, WRITE_DATA)
	asserts.assert_true(write_operation_result,
	'BLE write operation failed to start')
	server_write_operation_result = server.server_callback.waitAndGet(
	'onCharacteristicWriteRequest', 30)
	asserts.assert_equal(server_write_operation_result.data['Data'], WRITE_DATA)
	client.client_callback.waitAndGet('onCharacteristicWrite', 30)
	client.log.info('Write operation finished')
	write_operation_result = client.android.bleWriteOperation(
	TEST_BLE_SERVICE_UUID, TEST_SECOND_WRITE_UUID, SECOND_WRITE_DATA)
	asserts.assert_true(write_operation_result,
	'BLE write operation failed to start')
	server_write_operation_result = server.server_callback.waitAndGet(
	'onCharacteristicWriteRequest', 30)
	asserts.assert_equal(server_write_operation_result.data['Data'],
	SECOND_WRITE_DATA)
	client.client_callback.waitAndGet('onCharacteristicWrite', 30)
	client.log.info('Second write operation finished')


	def ReliableWrite(client, server):
	"""Logic for BLE reliable write.

	Args:
	client: AndroidDevice. The device that behaves as GATT client.
	server: AndroidDevice. The device that behaves as GATT server.

	Steps:
	1. Client calls beginReliableWrite & gets true.
	2. Client writes a characteristic to server & gets true.
	3. Server calls sendResponse & client gets onCharacteristicWrite.
	4. Client calls executeReliableWrite & gets true
	5. Server calls sendResponse & client gets onReliableWriteCompleted.

	Verifies:
	Client get corresponding callbacks.
	"""
	begin_reliable_write_result = client.android.bleBeginReliableWrite()
	asserts.assert_true(begin_reliable_write_result,
	'BLE reliable write failed to start')
	client.log.info('BLE reliable write started')
	WriteCharacteristic(client, server)
	execute_reliable_write_result = client.android.bleExecuteReliableWrite()
	asserts.assert_true(execute_reliable_write_result,
	'BLE reliable write failed to execute')
	client.log.info('BLE reliable write execute started')
	client.client_callback.waitAndGet('onReliableWriteCompleted', 30)
	client.log.info('BLE reliable write finished')


	class BleTest(android_base_test.AndroidBaseTest):
	"""BLE tests."""

	def setup_class(self):
	super(BleTest, self).setup_class()
	self.initiator = self.dut_a
	self.receiver = self.dut_b
	# Sets the tag that represents this device in logs.
	# The device used to scan BLE devices and behave as a GATT client.
	self.initiator.debug_tag = 'initiator'
	# The device used to BLE advertise and behave as a GATT server.
	self.receiver.debug_tag = 'receiver'

	def setup_test(self):
	# Make sure bluetooth is on.
	self.initiator.android.btEnable()
	self.receiver.android.btEnable()

	def test_basic_ble_process(self):
	"""Test for basic BLE process flow.

	Steps:
	1. Initiator discovers receiver.
	2. Initiator connects to receiver.
	3. Initiator discovers the BLE service receiver provided.
	4. Initiator reads a message from receiver.
	5. Initiator sends a message to receiver.
	6. Initiator disconnects from receiver.
	7. BLE scan and advertise stopped.

	Verifies:
	In each step, initiator and receiver get corresponding callbacks.
	"""
	Discover(self.initiator, self.receiver)
	Connect(self.initiator, self.receiver)
	DiscoverServices(self.initiator)
	ReadCharacteristic(self.initiator)
	WriteCharacteristic(self.initiator, self.receiver)
	Disconnect(self.initiator, self.receiver)
	StopDiscover(self.initiator, self.receiver)

	def test_cancel_open_ble_process(self):
	"""Test for BLE process flow involving cancel open.

	Steps:
	1. Initiator discovers receiver.
	2. Initiator initials connection to receiver, cancels connection and
	reconnects.
	3. Initiator discovers the BLE service receiver provided.
	6. Initiator disconnects from receiver.
	7. BLE scan and advertise stopped.

	Verifies:
	In each step, initiator and receiver get corresponding callbacks.
	"""
	Discover(self.initiator, self.receiver)
	CancelOpen(self.initiator, self.receiver)
	DiscoverServices(self.initiator)
	Disconnect(self.initiator, self.receiver)
	StopDiscover(self.initiator, self.receiver)

	def test_reliable_write_ble_process(self):
	"""Test for BLE process flow involving reliable write.

	Steps:
	1. Initiator discovers receiver.
	2. Initiator connects to receiver.
	3. Initiator discovers the BLE service receiver provided.
	4. Initiator starts reliable write to receiver and finishes successfully.
	5. Initiator disconnects from receiver.
	6. BLE scan and advertise stopped.

	Verifies:
	In each step, initiator and receiver get corresponding callbacks.
	"""
	Discover(self.initiator, self.receiver)
	Connect(self.initiator, self.receiver)
	DiscoverServices(self.initiator)
	ReliableWrite(self.initiator, self.receiver)
	Disconnect(self.initiator, self.receiver)
	StopDiscover(self.initiator, self.receiver)

	def teardown_test(self):
	# Turn Bluetooth off on both devices after test finishes.
	self.initiator.android.btDisable()
	self.receiver.android.btDisable()


	if __name__ == '__main__':
	test_runner.main()