| """ADB stress test for device push/pull commands. |
| |
| usage: adb_push_pull_stress.py [-h] [-d float] [-c int] |
| |
| optional arguments: |
| -h, --help Show this help message and exit. |
| -d float, --duration float Duration of time to run stress test (in hrs). |
| -c int, --count int Number of devices/emulators connected. |
| -p, --progress Print progress. |
| --log-dir Base directory under which logs will be placed. |
| """ |
| |
| from __future__ import print_function |
| import os |
| import subprocess |
| |
| import util |
| |
| TEMP_FILE = '__push_file.txt' |
| |
| # Number of lines in push/pull'd file. |
| FILE_SIZE = 100 |
| |
| |
| def create_temp_files(): |
| """Setup for push test. |
| |
| Creates all temporary files created used in push test.""" |
| file = open(TEMP_FILE, 'w') |
| for i in range(FILE_SIZE): |
| file.write('lorem ipsum\n') |
| file.close() |
| |
| |
| def delete_temp_files(): |
| """Teardown for push test. |
| |
| Deletes all temporary files created for push test. |
| """ |
| try: |
| os.remove(TEMP_FILE) |
| except OSError: |
| pass |
| |
| |
| def test_push(dut): |
| """Verify that pushing a file is successful. |
| |
| File size is determined by FILE_SIZE constant. |
| |
| Args: |
| dut: Device to test against. |
| |
| Returns: |
| True if successful, else False. |
| """ |
| arg = 'adb -s ' + str(dut) + ' push ' + TEMP_FILE + ' /sdcard/' |
| process = subprocess.Popen(arg.split(), stdout=subprocess.PIPE) |
| output, error = process.communicate() |
| success = True |
| for line in output.split('\n'): |
| if line.startswith('adb: error'): |
| success = False |
| print('\nERROR:\nEPush FAILED for: ' + str(dut)) |
| print(output) |
| |
| return success |
| |
| |
| def test_pull(dut): |
| """Verify that pulling a file is successful. |
| |
| File size is determined by FILE_SIZE constant. |
| |
| Returns: |
| True if successful, else False. |
| """ |
| arg = 'adb -s ' + str(dut) + ' pull /sdcard/' + TEMP_FILE |
| process = subprocess.Popen(arg.split(), stdout=subprocess.PIPE) |
| output, error = process.communicate() |
| success = True |
| for line in output.split('\n'): |
| if line.startswith('adb: error'): |
| print('\nERROR:\nEPush FAILED for: ' + str(dut)) |
| print(output) |
| success = False |
| |
| return success |
| |
| |
| def test_device(dut): |
| """Runs single push/pull against a single device. |
| |
| Args: |
| dut: device under test |
| """ |
| return test_push(dut) and test_pull(dut) |
| |
| |
| if __name__ == '__main__': |
| args = util.parse_args() |
| util.launcher(test_device, args.duration, args.count, |
| setup=create_temp_files, cleanup=delete_temp_files, is_print_progress=args.progress, |
| log_dir=os.path.join(args.log_dir, 'adb_push_pull_stress')) |