| # Copyright 2017 Google Inc. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import collections |
| import io |
| import subprocess |
| import unittest |
| from unittest import mock |
| |
| from mobly.controllers.android_device_lib import adb |
| |
| # Mock parameters for instrumentation. |
| MOCK_INSTRUMENTATION_PACKAGE = 'com.my.instrumentation.tests' |
| MOCK_INSTRUMENTATION_RUNNER = 'com.my.instrumentation.runner' |
| MOCK_INSTRUMENTATION_OPTIONS = collections.OrderedDict( |
| [ |
| ('option1', 'value1'), |
| ('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 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 Shell Command |
| MOCK_SHELL_COMMAND = 'ls' |
| MOCK_COMMAND_OUTPUT = '/system/bin/ls'.encode('utf-8') |
| MOCK_DEFAULT_STDOUT = 'out' |
| MOCK_DEFAULT_STDERR = 'err' |
| MOCK_DEFAULT_COMMAND_OUTPUT = MOCK_DEFAULT_STDOUT.encode('utf-8') |
| MOCK_ADB_SHELL_COMMAND_CHECK = 'adb shell command -v ls' |
| |
| |
| class AdbTest(unittest.TestCase): |
| """Unit tests for mobly.controllers.android_device_lib.adb.""" |
| |
| def _mock_execute_and_process_stdout_process(self, mock_popen): |
| # the created proc object in adb._execute_and_process_stdout() |
| mock_proc = mock.Mock() |
| mock_popen.return_value = mock_proc |
| |
| mock_popen.return_value.stdout.readline.side_effect = [''] |
| |
| mock_proc.communicate = mock.Mock( |
| 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'), |
| ) |
| self.assertTrue(adb.is_adb_available()) |
| |
| @mock.patch('mobly.utils.run_command') |
| def test_is_adb_available_negative(self, mock_run_command): |
| mock_run_command.return_value = (0, ''.encode('utf-8'), ''.encode('utf-8')) |
| self.assertFalse(adb.is_adb_available()) |
| |
| @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 |
| ) |
| 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_serial = 'ABCD1234' |
| 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) |
| |
| @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 |
| ) |
| 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'), |
| ) |
| |
| 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_serial = '1234Abcd' |
| |
| with self.assertRaisesRegex( |
| 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) |
| self.assertIn(mock_serial, context.exception.cmd) |
| |
| @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 |
| ) |
| |
| with self.assertRaisesRegex( |
| adb.AdbTimeoutError, |
| '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 |
| ) |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| def test_execute_and_process_stdout_reads_stdout(self, mock_popen): |
| self._mock_execute_and_process_stdout_process(mock_popen) |
| 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 |
| ) |
| self.assertEqual(mock_handler.call_count, 2) |
| mock_handler.assert_any_call('1') |
| mock_handler.assert_any_call('2') |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| def test_execute_and_process_stdout_reads_unexpected_stdout(self, mock_popen): |
| unexpected_stdout = MOCK_DEFAULT_STDOUT.encode('utf-8') |
| |
| 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')) |
| ) |
| |
| 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 |
| ): |
| 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) |
| ) |
| |
| 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, |
| ) |
| |
| @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 |
| ): |
| 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) |
| ) |
| |
| 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, |
| ) |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| def test_execute_and_process_stdout_despite_cmd_exits(self, mock_popen): |
| self._mock_execute_and_process_stdout_process(mock_popen) |
| mock_popen.return_value.poll.side_effect = [None, 0] |
| 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 |
| ) |
| |
| self.assertEqual(mock_handler.call_count, 3) |
| mock_handler.assert_any_call('1') |
| mock_handler.assert_any_call('2') |
| mock_handler.assert_any_call('3') |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| def test_execute_and_process_stdout_when_cmd_eof(self, mock_popen): |
| self._mock_execute_and_process_stdout_process(mock_popen) |
| 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 |
| ) |
| |
| self.assertEqual(mock_handler.call_count, 3) |
| mock_handler.assert_any_call('1') |
| mock_handler.assert_any_call('2') |
| mock_handler.assert_any_call('3') |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| 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() |
| ) |
| self.assertEqual(MOCK_DEFAULT_STDERR, err.decode('utf-8')) |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| def test_execute_and_process_stdout_raises_adb_error(self, mock_popen): |
| self._mock_execute_and_process_stdout_process(mock_popen) |
| 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() |
| ) |
| |
| @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen') |
| def test_execute_and_process_stdout_when_handler_crash(self, mock_popen): |
| self._mock_execute_and_process_stdout_process(mock_popen) |
| mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', ''] |
| mock_handler = mock.MagicMock() |
| 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 |
| ) |
| |
| mock_popen.return_value.communicate.assert_called_once_with() |
| |
| def test_construct_adb_cmd(self): |
| adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', 'arg1', shell=False) |
| self.assertEqual(adb_cmd, ['adb', 'shell', 'arg1']) |
| |
| def test_construct_adb_cmd_with_one_command(self): |
| 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(self): |
| 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 |
| ) |
| 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'\/\/']) |
| |
| def test_construct_adb_cmd_with_serial(self): |
| 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 |
| ) |
| 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 |
| ) |
| 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 |
| ) |
| 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 |
| ) |
| 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 |
| ) |
| 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 |
| ) |
| 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\"' '\\/\\/'") |
| |
| def test_construct_adb_cmd_with_shell_true_with_serial(self): |
| 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 |
| ) |
| 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 |
| ) |
| 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, |
| ) |
| |
| 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 |
| ) |
| |
| 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: |
| 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 |
| ) |
| |
| 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: |
| 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 |
| ) |
| |
| 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: |
| 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 |
| ) |
| mock_execute_and_process_stdout.assert_called_once_with( |
| 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'), |
| ) |
| 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') |
| ) |
| |
| @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'), |
| ) |
| |
| out = adb.AdbProxy().connect(mock_address) |
| self.assertEqual('connected to localhost:1234', out.decode('utf-8')) |
| |
| @mock.patch('mobly.utils.run_command') |
| 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'), |
| ) |
| |
| out = adb.AdbProxy().connect(mock_address) |
| self.assertEqual('already connected to localhost:1234', out.decode('utf-8')) |
| |
| @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'), |
| ) |
| |
| with self.assertRaisesRegex( |
| adb.AdbError, 'Error executing adb cmd "connect localhost:1234".' |
| ): |
| out = adb.AdbProxy().connect(mock_address) |
| |
| def test_getprop(self): |
| 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'), 'blah') |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', 'getprop', 'haha'], |
| shell=False, |
| stderr=None, |
| 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, |
| ) |
| |
| def test__parse_getprop_output_special_values(self): |
| mock_adb_output = ( |
| b'[selinux.restorecon_recursive]: [/data/misc_ce/10]\n' |
| b'[selinux.abc]: [key: value]\n' # "key: value" as value |
| b'[persist.sys.boot.reason.history]: [reboot,adb,1558549857\n' |
| b'reboot,factory_reset,1558483886\n' # multi-line value |
| b'reboot,1558483823]\n' |
| b'[persist.something]: [haha\n' |
| b']\n' |
| b'[[wrapped.key]]: [[wrapped value]]\n' |
| b'[persist.byte]: [J\xaa\x8bb\xab\x9dP\x0f]\n' # non-decodable |
| ) |
| 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' |
| ), |
| 'selinux.abc': 'key: value', |
| 'persist.something': 'haha\n', |
| 'selinux.restorecon_recursive': '/data/misc_ce/10', |
| '[wrapped.key]': '[wrapped value]', |
| 'persist.byte': 'JbP\x0f', |
| } |
| self.assertEqual(parsed_props, expected_output) |
| |
| def test__parse_getprop_output_malformat_output(self): |
| 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' |
| ) |
| parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output) |
| expected_output = { |
| 'persist.sys.boot.reason': 'reboot,adb,1558549857', |
| 'persist.something': 'haha', |
| } |
| self.assertEqual(parsed_props, expected_output) |
| |
| def test__parse_getprop_output_special_line_separator(self): |
| 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' |
| ) |
| parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output) |
| expected_output = { |
| 'persist.sys.boot.reason': 'reboot,adb,1558549857', |
| 'persist.something': 'haha', |
| } |
| self.assertEqual(parsed_props, expected_output) |
| |
| @mock.patch('time.sleep', return_value=mock.MagicMock()) |
| def test_getprops(self, mock_sleep): |
| with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: |
| mock_exec_cmd.return_value = ( |
| b'\n[sendbug.preferred.domain]: [google.com]\n' |
| b'[sys.uidcpupower]: []\n' |
| b'[sys.wifitracing.started]: [1]\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', |
| ] |
| ) |
| self.assertEqual( |
| actual_output, |
| { |
| 'sys.wifitracing.started': '1', |
| 'sys.uidcpupower': '', |
| '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, |
| ) |
| 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[ro.build.type]: [userdebug]\n\n', |
| ] |
| actual_output = adb.AdbProxy().getprops(['ro.build.id']) |
| 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, |
| ) |
| self.assertEqual(mock_sleep.call_count, 1) |
| mock_sleep.assert_called_with(1) |
| |
| @mock.patch('time.sleep', return_value=mock.MagicMock()) |
| def test_getprops_when_empty_string_always_returned(self, mock_sleep): |
| with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: |
| mock_exec_cmd.return_value = b'' |
| 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, |
| ) |
| 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(['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 |
| the basic case. |
| """ |
| with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: |
| output = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE) |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND], |
| shell=False, |
| timeout=None, |
| stderr=None, |
| ) |
| self.assertEqual(output, mock_exec_cmd.return_value) |
| |
| def test_instrument_with_runner(self): |
| """Verifies the AndroidDevice object's instrument command is correct |
| 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 |
| ) |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND], |
| shell=False, |
| timeout=None, |
| stderr=None, |
| ) |
| self.assertEqual(stdout, mock_exec_cmd.return_value) |
| |
| def test_instrument_with_options(self): |
| """Verifies the AndroidDevice object's instrument command is correct |
| 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 |
| ) |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND], |
| shell=False, |
| timeout=None, |
| stderr=None, |
| ) |
| self.assertEqual(stdout, mock_exec_cmd.return_value) |
| |
| def test_instrument_with_handler(self): |
| """Verifies the AndroidDevice object's instrument command is correct |
| with a handler passed in. |
| """ |
| |
| def mock_handler(raw_line): |
| 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 |
| ) |
| mock_execute_and_process_stdout.assert_called_once_with( |
| ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND], |
| shell=False, |
| handler=mock_handler, |
| ) |
| self.assertEqual(stderr, mock_execute_and_process_stdout.return_value) |
| |
| def test_instrument_with_handler_with_runner(self): |
| """Verifies the AndroidDevice object's instrument command is correct |
| with a handler passed in and a runner specified. |
| """ |
| |
| def mock_handler(raw_line): |
| 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, |
| ) |
| mock_execute_and_process_stdout.assert_called_once_with( |
| ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND], |
| shell=False, |
| handler=mock_handler, |
| ) |
| self.assertEqual(stderr, mock_execute_and_process_stdout.return_value) |
| |
| def test_instrument_with_handler_with_options(self): |
| """Verifies the AndroidDevice object's instrument command is correct |
| with a handler passed in and options. |
| """ |
| |
| def mock_handler(raw_line): |
| 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, |
| ) |
| mock_execute_and_process_stdout.assert_called_once_with( |
| ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND], |
| shell=False, |
| 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 |
| ) |
| self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT) |
| |
| @mock.patch('time.sleep', return_value=mock.MagicMock()) |
| @mock.patch.object(adb.AdbProxy, '_exec_cmd') |
| 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, |
| ] |
| output = adb.AdbProxy().root() |
| 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) |
| |
| @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 |
| ) |
| with self.assertRaisesRegex(adb.AdbError, expected_msg): |
| adb.AdbProxy().root() |
| 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)]) |
| |
| def test_has_shell_command_called_correctly(self): |
| with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: |
| mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT |
| adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND) |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', 'command', '-v', MOCK_SHELL_COMMAND], |
| shell=False, |
| timeout=None, |
| stderr=None, |
| ) |
| |
| def test_has_shell_command_with_existing_command(self): |
| with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd: |
| mock_exec_cmd.return_value = MOCK_COMMAND_OUTPUT |
| self.assertTrue(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND)) |
| |
| 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 |
| ) |
| 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 |
| ) |
| self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND)) |
| |
| @mock.patch.object(adb.AdbProxy, 'getprop') |
| @mock.patch.object(adb.AdbProxy, '_exec_cmd') |
| def test_current_user_id_25_and_above(self, mock_exec_cmd, mock_getprop): |
| mock_getprop.return_value = b'25' |
| mock_exec_cmd.return_value = b'123' |
| user_id = adb.AdbProxy().current_user_id |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', 'am', 'get-current-user'], |
| shell=False, |
| stderr=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' |
| ) |
| user_id = adb.AdbProxy().current_user_id |
| mock_exec_cmd.assert_called_once_with( |
| ['adb', 'shell', 'dumpsys', 'user'], |
| shell=False, |
| stderr=None, |
| timeout=None, |
| ) |
| self.assertEqual(user_id, 123) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
