diff options
Diffstat (limited to 'tests/mobly/controllers/android_device_lib/adb_test.py')
| -rwxr-xr-x | tests/mobly/controllers/android_device_lib/adb_test.py | 669 |
1 files changed, 391 insertions, 278 deletions
diff --git a/tests/mobly/controllers/android_device_lib/adb_test.py b/tests/mobly/controllers/android_device_lib/adb_test.py index a013959..65c7eb5 100755 --- a/tests/mobly/controllers/android_device_lib/adb_test.py +++ b/tests/mobly/controllers/android_device_lib/adb_test.py @@ -28,23 +28,30 @@ MOCK_INSTRUMENTATION_OPTIONS = collections.OrderedDict([ ('option2', 'value2'), ]) # Mock android instrumentation commands. -MOCK_BASIC_INSTRUMENTATION_COMMAND = ('am instrument -r -w com.my' - '.instrumentation.tests/com.android' - '.common.support.test.runner' - '.AndroidJUnitRunner') -MOCK_RUNNER_INSTRUMENTATION_COMMAND = ('am instrument -r -w com.my' - '.instrumentation.tests/com.my' - '.instrumentation.runner') -MOCK_OPTIONS_INSTRUMENTATION_COMMAND = ('am instrument -r -w -e option1 value1' - ' -e option2 value2 com.my' - '.instrumentation.tests/com.android' - '.common.support.test.runner' - '.AndroidJUnitRunner') +MOCK_BASIC_INSTRUMENTATION_COMMAND = ( + 'am instrument -r -w com.my' + '.instrumentation.tests/com.android' + '.common.support.test.runner' + '.AndroidJUnitRunner' +) +MOCK_RUNNER_INSTRUMENTATION_COMMAND = ( + 'am instrument -r -w com.my' + '.instrumentation.tests/com.my' + '.instrumentation.runner' +) +MOCK_OPTIONS_INSTRUMENTATION_COMMAND = ( + 'am instrument -r -w -e option1 value1' + ' -e option2 value2 com.my' + '.instrumentation.tests/com.android' + '.common.support.test.runner' + '.AndroidJUnitRunner' +) # Mock root command outputs. MOCK_ROOT_SUCCESS_OUTPUT = 'adbd is already running as root' -MOCK_ROOT_ERROR_OUTPUT = ( - 'adb: unable to connect for root: closed'.encode('utf-8')) +MOCK_ROOT_ERROR_OUTPUT = 'adb: unable to connect for root: closed'.encode( + 'utf-8' +) # Mock Shell Command MOCK_SHELL_COMMAND = 'ls' @@ -66,14 +73,18 @@ class AdbTest(unittest.TestCase): mock_popen.return_value.stdout.readline.side_effect = [''] mock_proc.communicate = mock.Mock( - return_value=('', MOCK_DEFAULT_STDERR.encode('utf-8'))) + return_value=('', MOCK_DEFAULT_STDERR.encode('utf-8')) + ) mock_proc.returncode = 0 return mock_popen @mock.patch('mobly.utils.run_command') def test_is_adb_available(self, mock_run_command): - mock_run_command.return_value = (0, '/usr/local/bin/adb\n'.encode('utf-8'), - ''.encode('utf-8')) + mock_run_command.return_value = ( + 0, + '/usr/local/bin/adb\n'.encode('utf-8'), + ''.encode('utf-8'), + ) self.assertTrue(adb.is_adb_available()) @mock.patch('mobly.utils.run_command') @@ -83,23 +94,29 @@ class AdbTest(unittest.TestCase): @mock.patch('mobly.utils.run_command') def test_exec_cmd_no_timeout_success(self, mock_run_command): - mock_run_command.return_value = (0, MOCK_DEFAULT_STDOUT.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) - out = adb.AdbProxy()._exec_cmd(['fake_cmd'], - shell=False, - timeout=None, - stderr=None) + mock_run_command.return_value = ( + 0, + MOCK_DEFAULT_STDOUT.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) + out = adb.AdbProxy()._exec_cmd( + ['fake_cmd'], shell=False, timeout=None, stderr=None + ) self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8')) mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=None) @mock.patch('mobly.utils.run_command') def test_exec_cmd_error_with_serial(self, mock_run_command): # Return 1 for retcode for error. - mock_run_command.return_value = (1, MOCK_DEFAULT_STDOUT.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) + mock_run_command.return_value = ( + 1, + MOCK_DEFAULT_STDOUT.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) mock_serial = 'ABCD1234' - with self.assertRaisesRegex(adb.AdbError, - 'Error executing adb cmd .*') as context: + with self.assertRaisesRegex( + adb.AdbError, 'Error executing adb cmd .*' + ) as context: adb.AdbProxy(mock_serial).fake_cmd() self.assertEqual(context.exception.serial, mock_serial) self.assertIn(mock_serial, context.exception.cmd) @@ -107,38 +124,45 @@ class AdbTest(unittest.TestCase): @mock.patch('mobly.utils.run_command') def test_exec_cmd_error_without_serial(self, mock_run_command): # Return 1 for retcode for error. - mock_run_command.return_value = (1, MOCK_DEFAULT_STDOUT.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) - with self.assertRaisesRegex(adb.AdbError, - 'Error executing adb cmd .*') as context: - adb.AdbProxy()._exec_cmd(['fake_cmd'], - shell=False, - timeout=None, - stderr=None) + mock_run_command.return_value = ( + 1, + MOCK_DEFAULT_STDOUT.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) + with self.assertRaisesRegex( + adb.AdbError, 'Error executing adb cmd .*' + ) as context: + adb.AdbProxy()._exec_cmd( + ['fake_cmd'], shell=False, timeout=None, stderr=None + ) self.assertFalse(context.exception.serial) mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=None) @mock.patch('mobly.utils.run_command') def test_exec_cmd_with_timeout_success(self, mock_run_command): - mock_run_command.return_value = (0, MOCK_DEFAULT_STDOUT.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) + mock_run_command.return_value = ( + 0, + MOCK_DEFAULT_STDOUT.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) - out = adb.AdbProxy()._exec_cmd(['fake_cmd'], - shell=False, - timeout=1, - stderr=None) + out = adb.AdbProxy()._exec_cmd( + ['fake_cmd'], shell=False, timeout=1, stderr=None + ) self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8')) mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=1) @mock.patch('mobly.utils.run_command') def test_exec_cmd_timed_out(self, mock_run_command): - mock_run_command.side_effect = subprocess.TimeoutExpired(cmd='mock_command', - timeout=0.01) + mock_run_command.side_effect = subprocess.TimeoutExpired( + cmd='mock_command', timeout=0.01 + ) mock_serial = '1234Abcd' with self.assertRaisesRegex( - adb.AdbTimeoutError, 'Timed out executing command "adb -s ' - '1234Abcd fake-cmd" after 0.01s.') as context: + adb.AdbTimeoutError, + 'Timed out executing command "adb -s 1234Abcd fake-cmd" after 0.01s.', + ) as context: adb.AdbProxy(mock_serial).fake_cmd(timeout=0.01) self.assertEqual(context.exception.serial, mock_serial) @@ -146,21 +170,23 @@ class AdbTest(unittest.TestCase): @mock.patch('mobly.utils.run_command') def test_exec_cmd_timed_out_without_serial(self, mock_run_command): - mock_run_command.side_effect = subprocess.TimeoutExpired(cmd='mock_command', - timeout=0.01) + mock_run_command.side_effect = subprocess.TimeoutExpired( + cmd='mock_command', timeout=0.01 + ) with self.assertRaisesRegex( adb.AdbTimeoutError, - 'Timed out executing command "adb fake-cmd" after 0.01s.'): + 'Timed out executing command "adb fake-cmd" after 0.01s.', + ): adb.AdbProxy().fake_cmd(timeout=0.01) def test_exec_cmd_with_negative_timeout_value(self): - with self.assertRaisesRegex(ValueError, - 'Timeout is not a positive value: -1'): - adb.AdbProxy()._exec_cmd(['fake_cmd'], - shell=False, - timeout=-1, - stderr=None) + with self.assertRaisesRegex( + ValueError, 'Timeout is not a positive value: -1' + ): + adb.AdbProxy()._exec_cmd( + ['fake_cmd'], shell=False, timeout=-1, stderr=None + ) @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') def test_execute_and_process_stdout_reads_stdout(self, mock_popen): @@ -168,9 +194,9 @@ class AdbTest(unittest.TestCase): mock_popen.return_value.stdout.readline.side_effect = ['1', '2', ''] mock_handler = mock.MagicMock() - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock_handler) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock_handler + ) self.assertEqual(mock_handler.call_count, 2) mock_handler.assert_any_call('1') mock_handler.assert_any_call('2') @@ -182,50 +208,63 @@ class AdbTest(unittest.TestCase): self._mock_execute_and_process_stdout_process(mock_popen) mock_handler = mock.MagicMock() mock_popen.return_value.communicate = mock.Mock( - return_value=(unexpected_stdout, MOCK_DEFAULT_STDERR.encode('utf-8'))) + return_value=(unexpected_stdout, MOCK_DEFAULT_STDERR.encode('utf-8')) + ) - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock_handler) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock_handler + ) self.assertEqual(mock_handler.call_count, 1) mock_handler.assert_called_with(unexpected_stdout) @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') @mock.patch('logging.debug') - def test_execute_and_process_stdout_logs_cmd(self, mock_debug_logger, - mock_popen): + def test_execute_and_process_stdout_logs_cmd( + self, mock_debug_logger, mock_popen + ): raw_expected_stdout = '' expected_stdout = '[elided, processed via handler]' expected_stderr = MOCK_DEFAULT_STDERR.encode('utf-8') self._mock_execute_and_process_stdout_process(mock_popen) mock_popen.return_value.communicate = mock.Mock( - return_value=(raw_expected_stdout, expected_stderr)) + return_value=(raw_expected_stdout, expected_stderr) + ) - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock.MagicMock()) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock.MagicMock() + ) mock_debug_logger.assert_called_with( - 'cmd: %s, stdout: %s, stderr: %s, ret: %s', 'fake_cmd', expected_stdout, - expected_stderr, 0) + 'cmd: %s, stdout: %s, stderr: %s, ret: %s', + 'fake_cmd', + expected_stdout, + expected_stderr, + 0, + ) @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') @mock.patch('logging.debug') def test_execute_and_process_stdout_logs_cmd_with_unexpected_stdout( - self, mock_debug_logger, mock_popen): + self, mock_debug_logger, mock_popen + ): raw_expected_stdout = MOCK_DEFAULT_STDOUT.encode('utf-8') expected_stdout = '[unexpected stdout] %s' % raw_expected_stdout expected_stderr = MOCK_DEFAULT_STDERR.encode('utf-8') self._mock_execute_and_process_stdout_process(mock_popen) mock_popen.return_value.communicate = mock.Mock( - return_value=(raw_expected_stdout, expected_stderr)) + return_value=(raw_expected_stdout, expected_stderr) + ) - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock.MagicMock()) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock.MagicMock() + ) mock_debug_logger.assert_called_with( - 'cmd: %s, stdout: %s, stderr: %s, ret: %s', 'fake_cmd', expected_stdout, - expected_stderr, 0) + 'cmd: %s, stdout: %s, stderr: %s, ret: %s', + 'fake_cmd', + expected_stdout, + expected_stderr, + 0, + ) @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') def test_execute_and_process_stdout_despite_cmd_exits(self, mock_popen): @@ -234,9 +273,9 @@ class AdbTest(unittest.TestCase): mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', ''] mock_handler = mock.MagicMock() - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock_handler) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock_handler + ) self.assertEqual(mock_handler.call_count, 3) mock_handler.assert_any_call('1') @@ -249,9 +288,9 @@ class AdbTest(unittest.TestCase): mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', ''] mock_handler = mock.MagicMock() - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock_handler) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock_handler + ) self.assertEqual(mock_handler.call_count, 3) mock_handler.assert_any_call('1') @@ -262,9 +301,9 @@ class AdbTest(unittest.TestCase): def test_execute_and_process_stdout_returns_stderr(self, mock_popen): self._mock_execute_and_process_stdout_process(mock_popen) - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock.MagicMock()) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock.MagicMock() + ) self.assertEqual(MOCK_DEFAULT_STDERR, err.decode('utf-8')) @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') @@ -273,9 +312,9 @@ class AdbTest(unittest.TestCase): mock_popen.return_value.returncode = 1 with self.assertRaisesRegex(adb.AdbError, 'Error executing adb cmd .*'): - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock.MagicMock()) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock.MagicMock() + ) @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') def test_execute_and_process_stdout_when_handler_crash(self, mock_popen): @@ -285,9 +324,9 @@ class AdbTest(unittest.TestCase): mock_handler.side_effect = ['', TypeError('fake crash'), '', ''] with self.assertRaisesRegex(TypeError, 'fake crash'): - err = adb.AdbProxy()._execute_and_process_stdout(['fake_cmd'], - shell=False, - handler=mock_handler) + err = adb.AdbProxy()._execute_and_process_stdout( + ['fake_cmd'], shell=False, handler=mock_handler + ) mock_popen.return_value.communicate.assert_called_once_with() @@ -297,180 +336,192 @@ class AdbTest(unittest.TestCase): def test_construct_adb_cmd_with_one_command(self): adb_cmd = adb.AdbProxy()._construct_adb_cmd( - 'shell ls /asdafsfd/asdf-asfd/asa', [], shell=False) + 'shell ls /asdafsfd/asdf-asfd/asa', [], shell=False + ) self.assertEqual(adb_cmd, ['adb', 'shell ls /asdafsfd/asdf-asfd/asa']) def test_construct_adb_cmd_with_one_arg_command(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - 'ls /asdafsfd/asdf-asfd/asa', - shell=False) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', 'ls /asdafsfd/asdf-asfd/asa', shell=False + ) self.assertEqual(adb_cmd, ['adb', 'shell', 'ls /asdafsfd/asdf-asfd/asa']) def test_construct_adb_cmd_with_one_arg_command_list(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - ['ls /asdafsfd/asdf-asfd/asa'], - shell=False) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', ['ls /asdafsfd/asdf-asfd/asa'], shell=False + ) self.assertEqual(adb_cmd, ['adb', 'shell', 'ls /asdafsfd/asdf-asfd/asa']) def test_construct_adb_cmd_with_special_characters(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - ['a b', '"blah"', r'\/\/'], - shell=False) - self.assertEqual(adb_cmd, ['adb', 'shell', 'a b', '"blah"', r"\/\/"]) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', ['a b', '"blah"', r'\/\/'], shell=False + ) + self.assertEqual(adb_cmd, ['adb', 'shell', 'a b', '"blah"', r'\/\/']) def test_construct_adb_cmd_with_serial(self): - adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell', - 'arg1', - shell=False) + adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd( + 'shell', 'arg1', shell=False + ) self.assertEqual(adb_cmd, ['adb', '-s', '12345', 'shell', 'arg1']) def test_construct_adb_cmd_with_list(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', ['arg1', 'arg2'], - shell=False) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', ['arg1', 'arg2'], shell=False + ) self.assertEqual(adb_cmd, ['adb', 'shell', 'arg1', 'arg2']) def test_construct_adb_cmd_with_serial_with_list(self): - adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell', - ['arg1', 'arg2'], - shell=False) + adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd( + 'shell', ['arg1', 'arg2'], shell=False + ) self.assertEqual(adb_cmd, ['adb', '-s', '12345', 'shell', 'arg1', 'arg2']) def test_construct_adb_cmd_with_shell_true(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - 'arg1 arg2', - shell=True) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', 'arg1 arg2', shell=True + ) self.assertEqual(adb_cmd, '"adb" shell arg1 arg2') def test_construct_adb_cmd_with_shell_true_with_one_command(self): adb_cmd = adb.AdbProxy()._construct_adb_cmd( - 'shell ls /asdafsfd/asdf-asfd/asa', [], shell=True) + 'shell ls /asdafsfd/asdf-asfd/asa', [], shell=True + ) self.assertEqual(adb_cmd, '"adb" shell ls /asdafsfd/asdf-asfd/asa ') def test_construct_adb_cmd_with_shell_true_with_one_arg_command(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - 'ls /asdafsfd/asdf-asfd/asa', - shell=True) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', 'ls /asdafsfd/asdf-asfd/asa', shell=True + ) self.assertEqual(adb_cmd, '"adb" shell ls /asdafsfd/asdf-asfd/asa') def test_construct_adb_cmd_with_shell_true_with_one_arg_command_list(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - ['ls /asdafsfd/asdf-asfd/asa'], - shell=True) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', ['ls /asdafsfd/asdf-asfd/asa'], shell=True + ) self.assertEqual(adb_cmd, '"adb" shell \'ls /asdafsfd/asdf-asfd/asa\'') def test_construct_adb_cmd_with_shell_true_with_auto_quotes(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', - ['a b', '"blah"', r'\/\/'], - shell=True) - self.assertEqual(adb_cmd, '"adb" shell \'a b\' \'"blah"\' \'\\/\\/\'') + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', ['a b', '"blah"', r'\/\/'], shell=True + ) + self.assertEqual(adb_cmd, "\"adb\" shell 'a b' '\"blah\"' '\\/\\/'") def test_construct_adb_cmd_with_shell_true_with_serial(self): - adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell', - 'arg1 arg2', - shell=True) + adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd( + 'shell', 'arg1 arg2', shell=True + ) self.assertEqual(adb_cmd, '"adb" -s "12345" shell arg1 arg2') def test_construct_adb_cmd_with_shell_true_with_list(self): - adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', ['arg1', 'arg2'], - shell=True) + adb_cmd = adb.AdbProxy()._construct_adb_cmd( + 'shell', ['arg1', 'arg2'], shell=True + ) self.assertEqual(adb_cmd, '"adb" shell arg1 arg2') def test_construct_adb_cmd_with_shell_true_with_serial_with_list(self): - adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd('shell', - ['arg1', 'arg2'], - shell=True) + adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd( + 'shell', ['arg1', 'arg2'], shell=True + ) self.assertEqual(adb_cmd, '"adb" -s "12345" shell arg1 arg2') def test_exec_adb_cmd(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT adb.AdbProxy().shell(['arg1', 'arg2']) - mock_exec_cmd.assert_called_once_with(['adb', 'shell', 'arg1', 'arg2'], - shell=False, - timeout=None, - stderr=None) + mock_exec_cmd.assert_called_once_with( + ['adb', 'shell', 'arg1', 'arg2'], + shell=False, + timeout=None, + stderr=None, + ) def test_exec_adb_cmd_with_shell_true(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT adb.AdbProxy().shell('arg1 arg2', shell=True) - mock_exec_cmd.assert_called_once_with('"adb" shell arg1 arg2', - shell=True, - timeout=None, - stderr=None) + mock_exec_cmd.assert_called_once_with( + '"adb" shell arg1 arg2', shell=True, timeout=None, stderr=None + ) def test_exec_adb_cmd_formats_command(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: - with mock.patch.object(adb.AdbProxy, - '_construct_adb_cmd') as mock_construct_adb_cmd: + with mock.patch.object( + adb.AdbProxy, '_construct_adb_cmd' + ) as mock_construct_adb_cmd: mock_adb_cmd = mock.MagicMock() mock_adb_args = mock.MagicMock() mock_construct_adb_cmd.return_value = mock_adb_cmd mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT adb.AdbProxy().shell(mock_adb_args) - mock_construct_adb_cmd.assert_called_once_with('shell', - mock_adb_args, - shell=False) - mock_exec_cmd.assert_called_once_with(mock_adb_cmd, - shell=False, - timeout=None, - stderr=None) + mock_construct_adb_cmd.assert_called_once_with( + 'shell', mock_adb_args, shell=False + ) + mock_exec_cmd.assert_called_once_with( + mock_adb_cmd, shell=False, timeout=None, stderr=None + ) def test_exec_adb_cmd_formats_command_with_shell_true(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: - with mock.patch.object(adb.AdbProxy, - '_construct_adb_cmd') as mock_construct_adb_cmd: + with mock.patch.object( + adb.AdbProxy, '_construct_adb_cmd' + ) as mock_construct_adb_cmd: mock_adb_cmd = mock.MagicMock() mock_adb_args = mock.MagicMock() mock_construct_adb_cmd.return_value = mock_adb_cmd adb.AdbProxy().shell(mock_adb_args, shell=True) - mock_construct_adb_cmd.assert_called_once_with('shell', - mock_adb_args, - shell=True) - mock_exec_cmd.assert_called_once_with(mock_adb_cmd, - shell=True, - timeout=None, - stderr=None) + mock_construct_adb_cmd.assert_called_once_with( + 'shell', mock_adb_args, shell=True + ) + mock_exec_cmd.assert_called_once_with( + mock_adb_cmd, shell=True, timeout=None, stderr=None + ) def test_execute_adb_and_process_stdout_formats_command(self): with mock.patch.object( - adb.AdbProxy, - '_execute_and_process_stdout') as mock_execute_and_process_stdout: - with mock.patch.object(adb.AdbProxy, - '_construct_adb_cmd') as mock_construct_adb_cmd: + adb.AdbProxy, '_execute_and_process_stdout' + ) as mock_execute_and_process_stdout: + with mock.patch.object( + adb.AdbProxy, '_construct_adb_cmd' + ) as mock_construct_adb_cmd: mock_adb_cmd = mock.MagicMock() mock_adb_args = mock.MagicMock() mock_handler = mock.MagicMock() mock_construct_adb_cmd.return_value = mock_adb_cmd - adb.AdbProxy()._execute_adb_and_process_stdout('shell', - mock_adb_args, - shell=False, - handler=mock_handler) - mock_construct_adb_cmd.assert_called_once_with('shell', - mock_adb_args, - shell=False) + adb.AdbProxy()._execute_adb_and_process_stdout( + 'shell', mock_adb_args, shell=False, handler=mock_handler + ) + mock_construct_adb_cmd.assert_called_once_with( + 'shell', mock_adb_args, shell=False + ) mock_execute_and_process_stdout.assert_called_once_with( - mock_adb_cmd, shell=False, handler=mock_handler) + mock_adb_cmd, shell=False, handler=mock_handler + ) @mock.patch('mobly.utils.run_command') def test_exec_adb_cmd_with_stderr_pipe(self, mock_run_command): - mock_run_command.return_value = (0, MOCK_DEFAULT_STDOUT.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) + mock_run_command.return_value = ( + 0, + MOCK_DEFAULT_STDOUT.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) stderr_redirect = io.BytesIO() out = adb.AdbProxy().shell('arg1 arg2', shell=True, stderr=stderr_redirect) self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8')) - self.assertEqual(MOCK_DEFAULT_STDERR, - stderr_redirect.getvalue().decode('utf-8')) + self.assertEqual( + MOCK_DEFAULT_STDERR, stderr_redirect.getvalue().decode('utf-8') + ) @mock.patch('mobly.utils.run_command') def test_connect_success(self, mock_run_command): mock_address = 'localhost:1234' mock_run_command.return_value = ( - 0, f'connected to {mock_address}'.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) + 0, + f'connected to {mock_address}'.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) out = adb.AdbProxy().connect(mock_address) self.assertEqual('connected to localhost:1234', out.decode('utf-8')) @@ -479,8 +530,10 @@ class AdbTest(unittest.TestCase): def test_connect_already_connected(self, mock_run_command): mock_address = 'localhost:1234' mock_run_command.return_value = ( - 0, f'already connected to {mock_address}'.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) + 0, + f'already connected to {mock_address}'.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) out = adb.AdbProxy().connect(mock_address) self.assertEqual('already connected to localhost:1234', out.decode('utf-8')) @@ -488,11 +541,15 @@ class AdbTest(unittest.TestCase): @mock.patch('mobly.utils.run_command') def test_connect_fail(self, mock_run_command): mock_address = 'localhost:1234' - mock_run_command.return_value = (0, 'Connection refused\n'.encode('utf-8'), - MOCK_DEFAULT_STDERR.encode('utf-8')) + mock_run_command.return_value = ( + 0, + 'Connection refused\n'.encode('utf-8'), + MOCK_DEFAULT_STDERR.encode('utf-8'), + ) with self.assertRaisesRegex( - adb.AdbError, 'Error executing adb cmd "connect localhost:1234".'): + adb.AdbError, 'Error executing adb cmd "connect localhost:1234".' + ): out = adb.AdbProxy().connect(mock_address) def test_getprop(self): @@ -503,18 +560,22 @@ class AdbTest(unittest.TestCase): ['adb', 'shell', 'getprop', 'haha'], shell=False, stderr=None, - timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC) + timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC, + ) def test_getprop_custom_timeout(self): timeout_s = adb.DEFAULT_GETPROP_TIMEOUT_SEC * 2 with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: mock_exec_cmd.return_value = b'blah' - self.assertEqual(adb.AdbProxy().getprop('haha', timeout=timeout_s), - 'blah') - mock_exec_cmd.assert_called_once_with(['adb', 'shell', 'getprop', 'haha'], - shell=False, - stderr=None, - timeout=timeout_s) + self.assertEqual( + adb.AdbProxy().getprop('haha', timeout=timeout_s), 'blah' + ) + mock_exec_cmd.assert_called_once_with( + ['adb', 'shell', 'getprop', 'haha'], + shell=False, + stderr=None, + timeout=timeout_s, + ) def test__parse_getprop_output_special_values(self): mock_adb_output = ( @@ -530,9 +591,10 @@ class AdbTest(unittest.TestCase): ) parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output) expected_output = { - 'persist.sys.boot.reason.history': - ('reboot,adb,1558549857\nreboot,factory_reset,1558483886\n' - 'reboot,1558483823'), + 'persist.sys.boot.reason.history': ( + 'reboot,adb,1558549857\nreboot,factory_reset,1558483886\n' + 'reboot,1558483823' + ), 'selinux.abc': 'key: value', 'persist.something': 'haha\n', 'selinux.restorecon_recursive': '/data/misc_ce/10', @@ -542,26 +604,28 @@ class AdbTest(unittest.TestCase): self.assertEqual(parsed_props, expected_output) def test__parse_getprop_output_malformat_output(self): - mock_adb_output = ( - b'[selinux.restorecon_recursive][/data/misc_ce/10]\n' # Malformat + mock_adb_output = ( # Malformat + b'[selinux.restorecon_recursive][/data/misc_ce/10]\n' b'[persist.sys.boot.reason]: [reboot,adb,1558549857]\n' - b'[persist.something]: [haha]\n') + b'[persist.something]: [haha]\n' + ) parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output) expected_output = { 'persist.sys.boot.reason': 'reboot,adb,1558549857', - 'persist.something': 'haha' + 'persist.something': 'haha', } self.assertEqual(parsed_props, expected_output) def test__parse_getprop_output_special_line_separator(self): - mock_adb_output = ( - b'[selinux.restorecon_recursive][/data/misc_ce/10]\r\n' # Malformat + mock_adb_output = ( # Malformat + b'[selinux.restorecon_recursive][/data/misc_ce/10]\r\n' b'[persist.sys.boot.reason]: [reboot,adb,1558549857]\r\n' - b'[persist.something]: [haha]\r\n') + b'[persist.something]: [haha]\r\n' + ) parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output) expected_output = { 'persist.sys.boot.reason': 'reboot,adb,1558549857', - 'persist.something': 'haha' + 'persist.something': 'haha', } self.assertEqual(parsed_props, expected_output) @@ -572,42 +636,51 @@ class AdbTest(unittest.TestCase): b'\n[sendbug.preferred.domain]: [google.com]\n' b'[sys.uidcpupower]: []\n' b'[sys.wifitracing.started]: [1]\n' - b'[telephony.lteOnCdmaDevice]: [1]\n\n') + b'[telephony.lteOnCdmaDevice]: [1]\n\n' + ) actual_output = adb.AdbProxy().getprops([ 'sys.wifitracing.started', # "numeric" value 'sys.uidcpupower', # empty value 'sendbug.preferred.domain', # string value - 'nonExistentProp' + 'nonExistentProp', ]) self.assertEqual( - actual_output, { + actual_output, + { 'sys.wifitracing.started': '1', 'sys.uidcpupower': '', - 'sendbug.preferred.domain': 'google.com' - }) + 'sendbug.preferred.domain': 'google.com', + }, + ) mock_exec_cmd.assert_called_once_with( ['adb', 'shell', 'getprop'], shell=False, stderr=None, - timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC) + timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC, + ) mock_sleep.assert_not_called() @mock.patch('time.sleep', return_value=mock.MagicMock()) def test_getprops_when_empty_string_randomly_returned(self, mock_sleep): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: mock_exec_cmd.side_effect = [ - b'', (b'\n[ro.build.id]: [AB42]\n' - b'[ro.build.type]: [userdebug]\n\n') + b'', + b'\n[ro.build.id]: [AB42]\n[ro.build.type]: [userdebug]\n\n', ] actual_output = adb.AdbProxy().getprops(['ro.build.id']) - self.assertEqual(actual_output, { - 'ro.build.id': 'AB42', - }) + self.assertEqual( + actual_output, + { + 'ro.build.id': 'AB42', + }, + ) self.assertEqual(mock_exec_cmd.call_count, 2) - mock_exec_cmd.assert_called_with(['adb', 'shell', 'getprop'], - shell=False, - stderr=None, - timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC) + mock_exec_cmd.assert_called_with( + ['adb', 'shell', 'getprop'], + shell=False, + stderr=None, + timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC, + ) self.assertEqual(mock_sleep.call_count, 1) mock_sleep.assert_called_with(1) @@ -618,16 +691,34 @@ class AdbTest(unittest.TestCase): actual_output = adb.AdbProxy().getprops(['ro.build.id']) self.assertEqual(actual_output, {}) self.assertEqual(mock_exec_cmd.call_count, 3) - mock_exec_cmd.assert_called_with(['adb', 'shell', 'getprop'], - shell=False, - stderr=None, - timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC) + mock_exec_cmd.assert_called_with( + ['adb', 'shell', 'getprop'], + shell=False, + stderr=None, + timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC, + ) self.assertEqual(mock_sleep.call_count, 2) mock_sleep.assert_called_with(1) def test_forward(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: - adb.AdbProxy().forward(MOCK_SHELL_COMMAND) + adb.AdbProxy().forward(['tcp:12345', 'tcp:98765']) + mock_exec_cmd.assert_called_with( + ['adb', 'forward', 'tcp:12345', 'tcp:98765'], + shell=False, + timeout=None, + stderr=None, + ) + + def test_reverse(self): + with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: + adb.AdbProxy().reverse(['tcp:12345', 'tcp:98765']) + mock_exec_cmd.assert_called_with( + ['adb', 'reverse', 'tcp:12345', 'tcp:98765'], + shell=False, + timeout=None, + stderr=None, + ) def test_instrument_without_parameters(self): """Verifies the AndroidDevice object's instrument command is correct in @@ -639,7 +730,8 @@ class AdbTest(unittest.TestCase): ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND], shell=False, timeout=None, - stderr=None) + stderr=None, + ) self.assertEqual(output, mock_exec_cmd.return_value) def test_instrument_with_runner(self): @@ -647,13 +739,15 @@ class AdbTest(unittest.TestCase): with a runner specified. """ with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: - stdout = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE, - runner=MOCK_INSTRUMENTATION_RUNNER) + stdout = adb.AdbProxy().instrument( + MOCK_INSTRUMENTATION_PACKAGE, runner=MOCK_INSTRUMENTATION_RUNNER + ) mock_exec_cmd.assert_called_once_with( ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND], shell=False, timeout=None, - stderr=None) + stderr=None, + ) self.assertEqual(stdout, mock_exec_cmd.return_value) def test_instrument_with_options(self): @@ -661,13 +755,15 @@ class AdbTest(unittest.TestCase): with options. """ with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: - stdout = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE, - options=MOCK_INSTRUMENTATION_OPTIONS) + stdout = adb.AdbProxy().instrument( + MOCK_INSTRUMENTATION_PACKAGE, options=MOCK_INSTRUMENTATION_OPTIONS + ) mock_exec_cmd.assert_called_once_with( ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND], shell=False, timeout=None, - stderr=None) + stderr=None, + ) self.assertEqual(stdout, mock_exec_cmd.return_value) def test_instrument_with_handler(self): @@ -679,14 +775,16 @@ class AdbTest(unittest.TestCase): pass with mock.patch.object( - adb.AdbProxy, - '_execute_and_process_stdout') as mock_execute_and_process_stdout: - stderr = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE, - handler=mock_handler) + adb.AdbProxy, '_execute_and_process_stdout' + ) as mock_execute_and_process_stdout: + stderr = adb.AdbProxy().instrument( + MOCK_INSTRUMENTATION_PACKAGE, handler=mock_handler + ) mock_execute_and_process_stdout.assert_called_once_with( ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND], shell=False, - handler=mock_handler) + handler=mock_handler, + ) self.assertEqual(stderr, mock_execute_and_process_stdout.return_value) def test_instrument_with_handler_with_runner(self): @@ -698,15 +796,18 @@ class AdbTest(unittest.TestCase): pass with mock.patch.object( - adb.AdbProxy, - '_execute_and_process_stdout') as mock_execute_and_process_stdout: - stderr = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE, - runner=MOCK_INSTRUMENTATION_RUNNER, - handler=mock_handler) + adb.AdbProxy, '_execute_and_process_stdout' + ) as mock_execute_and_process_stdout: + stderr = adb.AdbProxy().instrument( + MOCK_INSTRUMENTATION_PACKAGE, + runner=MOCK_INSTRUMENTATION_RUNNER, + handler=mock_handler, + ) mock_execute_and_process_stdout.assert_called_once_with( ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND], shell=False, - handler=mock_handler) + handler=mock_handler, + ) self.assertEqual(stderr, mock_execute_and_process_stdout.return_value) def test_instrument_with_handler_with_options(self): @@ -718,25 +819,27 @@ class AdbTest(unittest.TestCase): pass with mock.patch.object( - adb.AdbProxy, - '_execute_and_process_stdout') as mock_execute_and_process_stdout: - stderr = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE, - options=MOCK_INSTRUMENTATION_OPTIONS, - handler=mock_handler) + adb.AdbProxy, '_execute_and_process_stdout' + ) as mock_execute_and_process_stdout: + stderr = adb.AdbProxy().instrument( + MOCK_INSTRUMENTATION_PACKAGE, + options=MOCK_INSTRUMENTATION_OPTIONS, + handler=mock_handler, + ) mock_execute_and_process_stdout.assert_called_once_with( ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND], shell=False, - handler=mock_handler) + handler=mock_handler, + ) self.assertEqual(stderr, mock_execute_and_process_stdout.return_value) @mock.patch.object(adb.AdbProxy, '_exec_cmd') def test_root_success(self, mock_exec_cmd): mock_exec_cmd.return_value = MOCK_ROOT_SUCCESS_OUTPUT output = adb.AdbProxy().root() - mock_exec_cmd.assert_called_once_with(['adb', 'root'], - shell=False, - timeout=None, - stderr=None) + mock_exec_cmd.assert_called_once_with( + ['adb', 'root'], shell=False, timeout=None, stderr=None + ) self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT) @mock.patch('time.sleep', return_value=mock.MagicMock()) @@ -744,13 +847,12 @@ class AdbTest(unittest.TestCase): def test_root_success_with_retry(self, mock_exec_cmd, mock_sleep): mock_exec_cmd.side_effect = [ adb.AdbError('adb root', '', MOCK_ROOT_ERROR_OUTPUT, 1), - MOCK_ROOT_SUCCESS_OUTPUT + MOCK_ROOT_SUCCESS_OUTPUT, ] output = adb.AdbProxy().root() - mock_exec_cmd.assert_called_with(['adb', 'root'], - shell=False, - timeout=None, - stderr=None) + mock_exec_cmd.assert_called_with( + ['adb', 'root'], shell=False, timeout=None, stderr=None + ) self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT) self.assertEqual(mock_sleep.call_count, 1) mock_sleep.assert_called_with(10) @@ -758,17 +860,20 @@ class AdbTest(unittest.TestCase): @mock.patch('time.sleep', return_value=mock.MagicMock()) @mock.patch.object(adb.AdbProxy, '_exec_cmd') def test_root_raises_adb_error_when_all_retries_failed( - self, mock_exec_cmd, mock_sleep): - mock_exec_cmd.side_effect = adb.AdbError('adb root', '', - MOCK_ROOT_ERROR_OUTPUT, 1) - expected_msg = ('Error executing adb cmd "adb root". ' - 'ret: 1, stdout: , stderr: %s' % MOCK_ROOT_ERROR_OUTPUT) + self, mock_exec_cmd, mock_sleep + ): + mock_exec_cmd.side_effect = adb.AdbError( + 'adb root', '', MOCK_ROOT_ERROR_OUTPUT, 1 + ) + expected_msg = ( + 'Error executing adb cmd "adb root". ret: 1, stdout: , stderr: %s' + % MOCK_ROOT_ERROR_OUTPUT + ) with self.assertRaisesRegex(adb.AdbError, expected_msg): adb.AdbProxy().root() - mock_exec_cmd.assert_called_with(['adb', 'root'], - shell=False, - timeout=None, - stderr=None) + mock_exec_cmd.assert_called_with( + ['adb', 'root'], shell=False, timeout=None, stderr=None + ) self.assertEqual(mock_sleep.call_count, adb.ADB_ROOT_RETRY_ATTEMPTS - 1) mock_sleep.assert_has_calls([mock.call(10), mock.call(20)]) @@ -780,7 +885,8 @@ class AdbTest(unittest.TestCase): ['adb', 'shell', 'command', '-v', MOCK_SHELL_COMMAND], shell=False, timeout=None, - stderr=None) + stderr=None, + ) def test_has_shell_command_with_existing_command(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: @@ -790,15 +896,17 @@ class AdbTest(unittest.TestCase): def test_has_shell_command_with_missing_command_on_older_devices(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT - mock_exec_cmd.side_effect = adb.AdbError(MOCK_ADB_SHELL_COMMAND_CHECK, '', - '', 0) + mock_exec_cmd.side_effect = adb.AdbError( + MOCK_ADB_SHELL_COMMAND_CHECK, '', '', 0 + ) self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND)) def test_has_shell_command_with_missing_command_on_newer_devices(self): with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT - mock_exec_cmd.side_effect = adb.AdbError(MOCK_ADB_SHELL_COMMAND_CHECK, '', - '', 1) + mock_exec_cmd.side_effect = adb.AdbError( + MOCK_ADB_SHELL_COMMAND_CHECK, '', '', 1 + ) self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND)) @mock.patch.object(adb.AdbProxy, 'getprop') @@ -811,25 +919,30 @@ class AdbTest(unittest.TestCase): ['adb', 'shell', 'am', 'get-current-user'], shell=False, stderr=None, - timeout=None) + timeout=None, + ) self.assertEqual(user_id, 123) @mock.patch.object(adb.AdbProxy, 'getprop') @mock.patch.object(adb.AdbProxy, '_exec_cmd') def test_current_user_id_between_21_and_24(self, mock_exec_cmd, mock_getprop): mock_getprop.return_value = b'23' - mock_exec_cmd.return_value = (b'Users:\n' - b'UserInfo{123:Owner:13} serialNo=0\n' - b'Created: <unknown>\n' - b'Last logged in: +1h22m12s497ms ago\n' - b'UserInfo{456:Owner:14} serialNo=0\n' - b'Created: <unknown>\n' - b'Last logged in: +1h01m12s497ms ago\n') + mock_exec_cmd.return_value = ( + b'Users:\n' + b'UserInfo{123:Owner:13} serialNo=0\n' + b'Created: <unknown>\n' + b'Last logged in: +1h22m12s497ms ago\n' + b'UserInfo{456:Owner:14} serialNo=0\n' + b'Created: <unknown>\n' + b'Last logged in: +1h01m12s497ms ago\n' + ) user_id = adb.AdbProxy().current_user_id - mock_exec_cmd.assert_called_once_with(['adb', 'shell', 'dumpsys', 'user'], - shell=False, - stderr=None, - timeout=None) + mock_exec_cmd.assert_called_once_with( + ['adb', 'shell', 'dumpsys', 'user'], + shell=False, + stderr=None, + timeout=None, + ) self.assertEqual(user_id, 123) |