Upload curator android tests for vts
Test: manual inspection
Bug: 67869554
Change-Id: Ia6c0bb997bfa45f731632108fb7396571767d865
diff --git a/OWNERS b/OWNERS
new file mode 100644
index 0000000..56a58bc
--- /dev/null
+++ b/OWNERS
@@ -0,0 +1,3 @@
+angli@google.com
+wwtbuaa@google.com
+yim@google.com
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..145fc7c
--- /dev/null
+++ b/README.md
@@ -0,0 +1,99 @@
+# NBU Connectivity Test Suite
+
+This suite includes tests that exercises Android connectivity APIs and verify
+device behaviors.
+
+## Overview
+
+These tests simulate connectivity use cases critical to NBU market like
+peer-to-peer interactions.
+
+We created these tests using tools from the open source project Mobly, if you
+want to learn more, see its [Github page](https://github.com/google/mobly) and
+[tutorial](https://github.com/google/mobly/blob/master/docs/tutorial.md).
+
+The tests have two components, an agent apk running on the Android Device Under
+Test (DUT) and Python test scripts running on a computer to which the DUTs are
+connected via USB. The tests issue cmds to the agent on the device to trigger
+actions and read status, thus coordinating actions across multiple devices.
+
+The tests by default write output to `/tmp/logs`.
+
+## Test Environment Setup
+
+This section lists the components and steps required to create a setup to run
+these tests.
+
+### Test Zip File
+
+This is a zip file that you have downloaded.
+
+The zip file includes:
+
+* this README file
+* an apk called `android_snippet.apk`
+* several Python files.
+
+### Two Android Devices (DUT)
+
+The two devices should be of the same model and build (identical fingerprint).
+
+On each device:
+
+* Flash the build you want to test. The build's API level has to be `>=26`.
+* Enable USB debugging.
+* Enable BT snoop log.
+* Enable Wi-Fi verbose log.
+* Enable location service.
+* Install the `android_snippet.apk` with opition -g.
+
+#### Create a test config file
+
+Based on the two devices' serial numbers, we need to create a config file.
+
+Create a plain text file `config.yaml` in the following format, with the `<>`
+blocks replaced with the information of your actual devices:
+
+```
+TestBeds:
+ - Name: P2pTestBed
+ Controllers:
+ AndroidDevice:
+ - serial: <DUT1 serial>
+ - serial: <DUT2 serial>
+```
+
+### Linux Computer (Ubuntu 14.04+)
+
+We need the following packages:
+
+* adb
+* Python 2.7+
+
+To check your Python's version, use command `$ python --version`.
+
+In addition to those, we also need to install a few other tools:
+
+```
+$ sudo apt-get install python-setuptools python-pip
+$ pip install mobly
+```
+
+After all the tools are installed, connect the devices to the computer with USB
+cables and make sure they show up as "device" in the output of `$ adb devices`
+
+## Test Execution
+
+First, you need to put the file `config.yaml` in the same directory as the
+Python scripts. Then cd to that directory and run:
+
+```
+$ rm -rf /tmp/logs/ # Clear previous logs
+$ python xxx_test.py -c config.yaml
+```
+
+Execute all the `*_test.py` files the same way. Once they all finish executing,
+all the logs will be collected in `/tmp/logs/`. You can check the test results
+and debug info there.
+
+Zip up the content of `/tmp/logs/` and send it to your contact at Google.
diff --git a/__init__.py b/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/__init__.py
diff --git a/android_snippet.apk b/android_snippet.apk
new file mode 100755
index 0000000..562ee4b
--- /dev/null
+++ b/android_snippet.apk
Binary files differ
diff --git a/ble_test.py b/ble_test.py
new file mode 100644
index 0000000..58fe5f9
--- /dev/null
+++ b/ble_test.py
@@ -0,0 +1,545 @@
+"""BLE tests."""
+
+import time
+
+
+
+from mobly import asserts
+from mobly import mobly_g3
+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__':
+ mobly_g3.main()
diff --git a/bluetooth_test.py b/bluetooth_test.py
new file mode 100644
index 0000000..0fa96a5
--- /dev/null
+++ b/bluetooth_test.py
@@ -0,0 +1,105 @@
+"""Bluetooth tests."""
+
+
+
+from mobly import asserts
+from mobly import mobly_g3
+from mobly import utils
+from utils import android_base_test
+
+# Number of seconds for the target to stay discoverable on Bluetooth.
+DISCOVERABLE_TIME = 180
+# Random DATA to represent device bluetooth name.
+DEVICE_NAME = utils.rand_ascii_str(8)
+# UUID value for RfComm connection
+RFCOMM_UUID = 'fa87c0d0-afac-11de-8a39-0800200c9a66'
+
+
+def DiscoverBluetoothDeviceByName(device, name):
+ """Start a bluetooth scan and find the device that has the bluetooth name.
+
+ Args:
+ device: AndroidDevice. Device to start bluetooth scan.
+ name: the bluetooth name looking for.
+
+ Returns:
+ Dict represents a bluetooth device if found.
+ """
+ discovered_devices = device.android.btDiscoverAndGetResults()
+ for device in discovered_devices:
+ if name == device['Name']:
+ device_discovered = device
+ return device_discovered
+ asserts.fail('Failed to discover the target device %s over Bluetooth.' % name)
+
+
+class BluetoothTest(android_base_test.AndroidBaseTest):
+ """Bluetooth tests."""
+
+ def setup_class(self):
+ super(BluetoothTest, self).setup_class()
+ self.initiator = self.dut_a
+ # Sets the tag that represents this device in logs.
+ self.initiator.debug_tag = 'initiator'
+ # The device that is expected to be discovered and receive messages.
+ self.receiver = self.dut_b
+ self.receiver.debug_tag = 'receiver'
+
+ def setup_test(self):
+ # Make sure bluetooth is on.
+ self.initiator.android.btEnable()
+ self.receiver.android.btEnable()
+ # Set Bluetooth name on target device.
+ self.receiver.android.btSetName(DEVICE_NAME)
+
+ def test_bluetooth_process(self):
+ """Test for basic bluetooth rfcomm process flow.
+
+ Steps:
+ 1. Receiver becomes discoverable.
+ 2. Initiator discovers receiver via bluetooth.
+ 3. Initiator connects to receiver via rfcomm profile.
+ 4. Initiator sends a message to receiver and receiver receives the exact
+ message.
+
+ Verifies:
+ Receiver receives the correct message.
+ """
+ # Name value for RfComm connection
+ rfcomm_name = utils.rand_ascii_str(8)
+ self.receiver.connection_callback = (
+ self.receiver.android.btRfcommStartServer(rfcomm_name, RFCOMM_UUID))
+ self.receiver.log.info('Start Rfcomm server with name: %s uuid: %s' %
+ (rfcomm_name, RFCOMM_UUID))
+ target_name = self.receiver.android.btGetName()
+ self.receiver.log.info('Become discoverable with name "%s" for %ds.',
+ target_name, DISCOVERABLE_TIME)
+ self.receiver.android.btBecomeDiscoverable(DISCOVERABLE_TIME)
+ self.initiator.log.info('Looking for Bluetooth devices.')
+ discovered_device = DiscoverBluetoothDeviceByName(
+ self.initiator, target_name)
+ self.initiator.log.info('Target device is found. Device: %s'
+ % discovered_device)
+ remote_address = discovered_device['Address']
+ self.initiator.android.btRfcommConnect(remote_address, RFCOMM_UUID)
+ self.receiver.connection_callback.waitAndGet('onAccepted', 30)
+ # self.initiator.connection_callback.waitAndGet('onConnected', 30)
+ self.initiator.log.info('Connection established')
+ # Random data to be sent through bluetooth rfcomm.
+ data = utils.rand_ascii_str(8)
+ self.receiver.read_callback = self.receiver.android.btRfcommRead()
+ self.initiator.android.btRfcommWrite(data)
+ read_result = self.receiver.read_callback.waitAndGet('onDataAvailable', 30)
+ asserts.assert_equal(read_result.data['Data'], data)
+ self.receiver.log.info('Received correct message from the other side')
+ self.initiator.android.btRfcommDisconnect()
+ self.receiver.android.btRfcommStopServer()
+
+ def teardown_test(self):
+ # Turn Bluetooth off on both devices after test finishes.
+ self.receiver.android.btDisable()
+ self.initiator.android.btDisable()
+
+
+if __name__ == '__main__':
+ mobly_g3.main()
diff --git a/local_hotspot_test.py b/local_hotspot_test.py
new file mode 100644
index 0000000..774b9ed
--- /dev/null
+++ b/local_hotspot_test.py
@@ -0,0 +1,71 @@
+"""Local hotspot tests."""
+
+import time
+
+
+
+from mobly import asserts
+from mobly import mobly_g3
+from utils import android_base_test
+
+SSID = 'SSID'
+# Number of seconds for the receiver to restore wifi state.
+RESTORE_TIME = 20
+# The time period that SSID needs to be found.
+SCAN_TIMEOUT = 30
+
+
+class LocalHotspotTest(android_base_test.AndroidBaseTest):
+ """Local hotspot tests."""
+
+ def setup_class(self):
+ super(LocalHotspotTest, self).setup_class()
+ self.station = self.dut_a
+ self.ap = self.dut_b
+ # The device used to discover Bluetooth devices and send messages.
+ # Sets the tag that represents this device in logs.
+ self.station.debug_tag = 'station'
+ # The device that is expected to be discovered and receive messages.
+ self.ap.debug_tag = 'ap'
+
+ def setup_test(self):
+ # Make sure wifi is on.
+ self.station.android.wifiEnable()
+ self.ap.android.wifiEnable()
+
+ def test_local_hotspot_process(self):
+ """Test for basic local hotspot process flow.
+
+ Steps:
+ 1. Ap sets up a local hotspot and retrieves the credentials.
+ 2. Station connects to the hotspot.
+
+ Verifies:
+ Station can connect to the local hotspot created by ap.
+ """
+ wifi_on_before = self.ap.android.wifiIsEnabled()
+ start_localhotspot_callback = self.ap.android.startLocalHotspot()
+ start_result = start_localhotspot_callback.waitAndGet('onStarted', 30)
+ local_hotspot_info = start_result.data
+ self.ap.log.info('Local hotspot started')
+ network_found = self.station.android.wifiScanAndFindSsid(
+ local_hotspot_info[SSID], SCAN_TIMEOUT)
+ asserts.assert_true(network_found, 'Network is not found within 30 seconds')
+ self.station.android.wifiConnectByUpdate(local_hotspot_info[SSID],
+ local_hotspot_info['Password'])
+ self.station.log.info('Connected to the network %s.'
+ % local_hotspot_info[SSID])
+ self.ap.android.stopLocalHotspot()
+ time.sleep(RESTORE_TIME)
+ wifi_on_after = self.ap.android.wifiIsEnabled()
+ asserts.assert_equal(wifi_on_before, wifi_on_after)
+ self.ap.log.info('Wifi state restored')
+
+ def teardown_test(self):
+ # Turn wifi off on both devices after test finishes.
+ self.station.android.wifiDisable()
+ self.ap.android.wifiDisable()
+
+
+if __name__ == '__main__':
+ mobly_g3.main()
diff --git a/utils/__init__.py b/utils/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/utils/__init__.py
diff --git a/utils/android_base_test.py b/utils/android_base_test.py
new file mode 100644
index 0000000..d3ea0a6
--- /dev/null
+++ b/utils/android_base_test.py
@@ -0,0 +1,45 @@
+"""Base class for automated android tests."""
+
+
+
+from mobly import base_test
+from mobly import logger
+from mobly.controllers import android_device
+from mobly.controllers.android_device_lib import adb
+
+ANDROID_SNIPPET_PACKAGE = 'com.google.android.test'
+
+
+class AndroidBaseTest(base_test.BaseTestClass):
+
+ def log_bt_stack_conf(self, device):
+ try:
+ bt_stack_conf = device.adb.shell(['cat', '/etc/bluetooth/bt_stack.conf'])
+ except adb.AdbError:
+ device.log.debug('No /etc/bluetooth/bt_stack.conf file found on device.')
+ else:
+ device.log.debug('Content of /etc/bluetooth/bt_stack.conf: %s',
+ bt_stack_conf)
+
+ def setup_class(self):
+ # Registering android_device controller module, and declaring that the test
+ # requires at least two Android devices.
+ self.ads = self.register_controller(android_device, min_number=2)
+ self.dut_a = self.ads[0]
+ self.dut_b = self.ads[1]
+ self.dut_a.load_snippet('android', ANDROID_SNIPPET_PACKAGE)
+ self.dut_b.load_snippet('android', ANDROID_SNIPPET_PACKAGE)
+ self.log_bt_stack_conf(self.dut_a)
+ self.log_bt_stack_conf(self.dut_b)
+
+ def on_fail(self, record):
+ """Only executed after a test fails.
+
+ Collect debug info here.
+
+ Args:
+ record: a copy of the test result record of the test.
+ """
+ begin_time = logger.epoch_to_log_line_timestamp(record.begin_time)
+ android_device.take_bug_reports([self.dut_a, self.dut_b], record.test_name,
+ begin_time)
diff --git a/utils/assert_utils.py b/utils/assert_utils.py
new file mode 100644
index 0000000..bf0d62f
--- /dev/null
+++ b/utils/assert_utils.py
@@ -0,0 +1,32 @@
+"""Utility functions for handling async events.
+"""
+
+
+from mobly import asserts
+from mobly.controllers.android_device_lib import callback_handler
+
+
+def AssertAsyncSuccess(callback):
+ """Wrapping the result of an asynchronous Rpc for convenience.
+
+ This wrapper handles the event posted by the corresponding `FutureCallback`
+ object signalling whether the async operation was successful or not with
+ `onSuccess` and `onFailure` callbacks.
+
+ Args:
+ callback: CallbackHandler, the handler that's used to poll events triggered
+ by the async Rpc call.
+
+ Returns:
+ dictionary, the event that carries the onSuccess signal.
+ """
+ try:
+ status_event = callback.waitAndGet('onSuccess', 30)
+ except callback_handler.TimeoutError:
+ try:
+ failure_event = callback.waitAndGet('onFailure', 30)
+ asserts.fail('Operation failed: %s' % failure_event.data)
+ except callback_handler.TimeoutError:
+ asserts.fail('Timed out waiting for status event.')
+ return status_event
+
diff --git a/wifi_direct_test.py b/wifi_direct_test.py
new file mode 100644
index 0000000..6f4b083
--- /dev/null
+++ b/wifi_direct_test.py
@@ -0,0 +1,72 @@
+"""Wifi Direct tests."""
+
+
+from mobly import asserts
+from mobly import mobly_g3
+from utils import android_base_test
+from utils import assert_utils
+
+SSID = 'SSID'
+# The time period that SSID needs to be found.
+SCAN_TIMEOUT = 30
+# The listen channel and operation channel for wifi direct.
+CHANNEL = 1
+# The frequency of wifi direct to be expected.
+FREQUENCY = 2412
+
+
+class WifiDirectTest(android_base_test.AndroidBaseTest):
+ """Wifi Direct tests."""
+
+ def setup_class(self):
+ super(WifiDirectTest, self).setup_class()
+ self.station = self.dut_a
+ self.group_owner = self.dut_b
+ # Sets the tag that represents this device in logs.
+ self.station.debug_tag = 'station'
+ self.group_owner.debug_tag = 'group_owner'
+
+ def setup_test(self):
+ # Make sure wifi of group owner is off to disable wifi direct.
+ self.group_owner.android.wifiDisable()
+ # Make sure wifi is on.
+ self.station.android.wifiEnable()
+ self.group_owner.android.wifiEnable()
+
+ def test_wifi_direct_legacy_connection(self):
+ """Test for basic Wifi Direct process flow.
+
+ Steps:
+ 1. Group owner sets up a wifi p2p group.
+ 2. Station connects to the group as a regular hotspot.
+
+ Verifies:
+ Station can connect to the wifi p2p group created by group owner.
+ """
+ callback = self.group_owner.android.wifiP2pSetChannel(CHANNEL, CHANNEL)
+ assert_utils.AssertAsyncSuccess(callback)
+ self.group_owner.log.info('Wifi direct channel set.')
+ callback = self.group_owner.android.wifiP2pStartGroup()
+ group_info = assert_utils.AssertAsyncSuccess(callback)
+ self.group_owner.log.info('Wifi direct group started as a temporary group.')
+ network_found = self.station.android.wifiScanAndFindSsid(
+ group_info.data[SSID], SCAN_TIMEOUT)
+ asserts.assert_true(network_found, 'Network is not found within 30 seconds')
+ asserts.assert_equal(network_found['frequency'], FREQUENCY)
+ self.station.log.info('Network is found, connecting...')
+ connect_result = self.station.android.wifiConnectByUpdate(
+ group_info.data[SSID], group_info.data['Password'])
+ asserts.assert_true(connect_result, 'Failed to connect to the network')
+ self.station.log.info('Connected to the network')
+ callback = self.group_owner.android.wifiP2pRemoveGroup()
+ assert_utils.AssertAsyncSuccess(callback)
+ self.group_owner.log.info('Wifi direct group removed')
+
+ def teardown_test(self):
+ # Turn wifi off on both devices after test finishes.
+ self.station.android.wifiDisable()
+ self.group_owner.android.wifiDisable()
+
+
+if __name__ == '__main__':
+ mobly_g3.main()