1 files changed, 106 insertions, 0 deletions
diff --git a/debian-bt-stress-test.py b/debian-bt-stress-test.py
new file mode 100755
index 0000000..da0873c
--- /dev/null
+++ b/debian-bt-stress-test.py
@@ -0,0 +1,106 @@
+#!/usr/bin/env python
+
+import os
+import sys
+import time
+import pexpect
+
+# Test client BT device address
+remote_device_address=sys.argv[1]
+test_case_list = ['power', 'discoverable', 'scan', 'pair']
+iteration = 1000
+output_dict = {
+    'power on': 'Changing power on succeeded',
+    'power off': 'Changing power off succeeded',
+    'discoverable on': 'Changing discoverable on succeeded',
+    'discoverable off': 'Changing discoverable off succeeded',
+    'scan on': 'Discovery started',
+    'scan off': 'Discovery stopped',
+    'pairable on': 'Changing pairable on succeeded',
+    'agent on': 'Agent registered',
+    'pair': 'Pairing successful',
+    'remove': 'Device has been removed'
+    }
+
+def print_test_result(test_case, test_result, count):
+        test_case = test_case + ' ' + str(count)
+        test_case = test_case.replace(' ', '_')
+        if test_result:
+            print('%s pass\n' % (test_case))
+        else:
+            print('%s fail\n' % (test_case))
+
+class Bluetoothctl:
+    def __init__(self):
+        self.child = pexpect.spawn('bluetoothctl')
+
+    def send_command(self, command, pause=5):
+        self.child.sendline(command)
+        time.sleep(pause)
+
+    def check_output(self, expected_output):
+        while True:
+            try:
+                line=self.child.readline()
+                if expected_output in line:
+                    print(line.strip())
+                    return True
+            except pexpect.TIMEOUT:
+                return False
+
+    def terminate(self):
+        self.child.terminate()
+
+# Run tests.
+print('Initializing bluetooth...')
+bl = Bluetoothctl()
+
+for test in test_case_list:
+    bl.send_command('power on')
+    if not bl.check_output('Changing power on succeeded'):
+        print('Failed to power on bluetooth adapter, exiting...')
+        sys.exit(1)
+
+    count = 0
+    print('\nRunning %s test for %s interations...' % (test, iteration))
+    print('--------')
+
+    if test in ['power', 'discoverable', 'scan']:
+        sub_tests = [test + ' on', test + ' off']
+        while count <= iteration:
+            for sub_test in sub_tests:
+                bl.send_command(sub_test)
+                expected_output = output_dict[sub_test]
+                test_result = bl.check_output(expected_output)
+                print_test_result(sub_test, test_result, count)
+            count = count + 1
+
+
+    if test is 'pair':
+        print('Test client need to be set to non-interactive pairing mode\n'
+              'setup-test-client.sh provided in this repo for this purpose')
+        raw_input('Press Enter to continue...')
+        for pre_set in ['pairable on', 'discoverable on', 'agent on']:
+            bl.send_command(pre_set)
+            if not bl.check_output(output_dict[pre_set]):
+                print('Failed to set %s, exiting...' % (pre_set))
+                sys.exit(1)
+            bl.send_command('remove ' + remote_device_address)
+
+        while count <= iteration:
+            bl.send_command('scan on')
+            if not bl.check_output('\r\x1b[K[\x1b[0;92mNEW\x1b[0m] Device ' + remote_device_address):
+                print('%s NOT found, retrying...' % (remote_device_address))
+                continue
+
+            bl.send_command('pair ' + remote_device_address)
+            test_result = bl.check_output(output_dict['pair'])
+            print_test_result('pair', test_result, count)
+
+            bl.send_command('remove ' + remote_device_address)
+            test_result = bl.check_output(output_dict['remove'])
+            print_test_result('unpair', test_result, count)
+
+            count = count + 1
+
+bl.terminate()