Move to 2-space indent: AndroidDevice related test files. (#709)
Pure shortening of the indentation length, no actual code change.
diff --git a/tests/mobly/controllers/android_device_lib/adb_test.py b/tests/mobly/controllers/android_device_lib/adb_test.py
index e920bbd..36e207a 100755
--- a/tests/mobly/controllers/android_device_lib/adb_test.py
+++ b/tests/mobly/controllers/android_device_lib/adb_test.py
@@ -24,27 +24,27 @@
MOCK_INSTRUMENTATION_PACKAGE = 'com.my.instrumentation.tests'
MOCK_INSTRUMENTATION_RUNNER = 'com.my.instrumentation.runner'
MOCK_INSTRUMENTATION_OPTIONS = OrderedDict([
- ('option1', 'value1'),
- ('option2', 'value2'),
+ ('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')
+ '.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')
+ '.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')
+ ' -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'))
+ 'adb: unable to connect for root: closed'.encode('utf-8'))
# Mock Shell Command
MOCK_SHELL_COMMAND = 'ls'
@@ -56,781 +56,781 @@
class AdbTest(unittest.TestCase):
- """Unit tests for mobly.controllers.android_device_lib.adb.
+ """Unit tests for mobly.controllers.android_device_lib.adb.
+ """
+
+ def _mock_process(self, mock_psutil_process, mock_popen):
+ # the created proc object in adb._exec_cmd()
+ mock_proc = mock.Mock()
+ mock_popen.return_value = mock_proc
+
+ # the created process object in adb._exec_cmd()
+ mock_psutil_process.return_value = mock.Mock()
+
+ mock_proc.communicate = mock.Mock(
+ return_value=(MOCK_DEFAULT_STDOUT.encode('utf-8'),
+ MOCK_DEFAULT_STDERR.encode('utf-8')))
+ mock_proc.returncode = 0
+ return (mock_psutil_process, mock_popen)
+
+ 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_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 = adb.psutil.TimeoutExpired('Timed out')
+ 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 = adb.psutil.TimeoutExpired('Timed out')
+ with self.assertRaisesRegex(
+ adb.AdbTimeoutError, 'Timed out executing command "adb '
+ 'fake-cmd" after 0.01s.') as context:
+ 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'))
+
+ 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__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 = (
+ b'[selinux.restorecon_recursive][/data/misc_ce/10]\n' # Malformat
+ 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 = (
+ b'[selinux.restorecon_recursive][/data/misc_ce/10]\r\n' # Malformat
+ 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'
+ b'[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(MOCK_SHELL_COMMAND)
+
+ 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_process(self, mock_psutil_process, mock_popen):
- # the created proc object in adb._exec_cmd()
- mock_proc = mock.Mock()
- mock_popen.return_value = mock_proc
+ def mock_handler(raw_line):
+ pass
- # the created process object in adb._exec_cmd()
- mock_psutil_process.return_value = mock.Mock()
+ 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)
- mock_proc.communicate = mock.Mock(
- return_value=(MOCK_DEFAULT_STDOUT.encode('utf-8'),
- MOCK_DEFAULT_STDERR.encode('utf-8')))
- mock_proc.returncode = 0
- return (mock_psutil_process, mock_popen)
+ 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_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
+ def mock_handler(raw_line):
+ pass
- mock_popen.return_value.stdout.readline.side_effect = ['']
+ 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)
- mock_proc.communicate = mock.Mock(
- return_value=('', MOCK_DEFAULT_STDERR.encode('utf-8')))
- mock_proc.returncode = 0
- return mock_popen
+ def test_instrument_with_handler_with_options(self):
+ """Verifies the AndroidDevice object's instrument command is correct
+ with a handler passed in and options.
+ """
- @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)
+ def mock_handler(raw_line):
+ pass
- @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)
+ 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('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.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('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.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)
- 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('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, 2)
+ mock_sleep.assert_called_with(10)
- @mock.patch('mobly.utils.run_command')
- def test_exec_cmd_timed_out(self, mock_run_command):
- mock_run_command.side_effect = adb.psutil.TimeoutExpired('Timed out')
- 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)
+ 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)
- @mock.patch('mobly.utils.run_command')
- def test_exec_cmd_timed_out_without_serial(self, mock_run_command):
- mock_run_command.side_effect = adb.psutil.TimeoutExpired('Timed out')
- with self.assertRaisesRegex(
- adb.AdbTimeoutError, 'Timed out executing command "adb '
- 'fake-cmd" after 0.01s.') as context:
- adb.AdbProxy().fake_cmd(timeout=0.01)
+ 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_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)
+ 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))
- @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()
+ 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))
- 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.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('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'))
-
- 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__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 = (
- b'[selinux.restorecon_recursive][/data/misc_ce/10]\n' # Malformat
- 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 = (
- b'[selinux.restorecon_recursive][/data/misc_ce/10]\r\n' # Malformat
- 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'
- b'[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(MOCK_SHELL_COMMAND)
-
- 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, 2)
- mock_sleep.assert_called_with(10)
-
- 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)
+ @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()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/callback_handler_test.py b/tests/mobly/controllers/android_device_lib/callback_handler_test.py
index b00e127..9476056 100755
--- a/tests/mobly/controllers/android_device_lib/callback_handler_test.py
+++ b/tests/mobly/controllers/android_device_lib/callback_handler_test.py
@@ -20,129 +20,129 @@
MOCK_CALLBACK_ID = "1-0"
MOCK_RAW_EVENT = {
- 'callbackId': '2-1',
- 'name': 'AsyncTaskResult',
- 'time': 20460228696,
- 'data': {
- 'exampleData': "Here's a simple event.",
- 'successful': True,
- 'secretNumber': 12
- }
+ 'callbackId': '2-1',
+ 'name': 'AsyncTaskResult',
+ 'time': 20460228696,
+ 'data': {
+ 'exampleData': "Here's a simple event.",
+ 'successful': True,
+ 'secretNumber': 12
+ }
}
class CallbackHandlerTest(unittest.TestCase):
- """Unit tests for mobly.controllers.android_device_lib.callback_handler.
+ """Unit tests for mobly.controllers.android_device_lib.callback_handler.
+ """
+
+ def test_timeout_value(self):
+ self.assertGreaterEqual(jsonrpc_client_base._SOCKET_READ_TIMEOUT,
+ callback_handler.MAX_TIMEOUT)
+
+ def test_callback_id_property(self):
+ mock_event_client = mock.Mock()
+ handler = callback_handler.CallbackHandler(
+ callback_id=MOCK_CALLBACK_ID,
+ event_client=mock_event_client,
+ ret_value=None,
+ method_name=None,
+ ad=mock.Mock())
+ self.assertEqual(handler.callback_id, MOCK_CALLBACK_ID)
+ with self.assertRaisesRegex(AttributeError, "can't set attribute"):
+ handler.callback_id = 'ha'
+
+ def test_event_dict_to_snippet_event(self):
+ mock_event_client = mock.Mock()
+ mock_event_client.eventWaitAndGet = mock.Mock(
+ return_value=MOCK_RAW_EVENT)
+ handler = callback_handler.CallbackHandler(
+ callback_id=MOCK_CALLBACK_ID,
+ event_client=mock_event_client,
+ ret_value=None,
+ method_name=None,
+ ad=mock.Mock())
+ event = handler.waitAndGet('ha')
+ self.assertEqual(event.name, MOCK_RAW_EVENT['name'])
+ self.assertEqual(event.creation_time, MOCK_RAW_EVENT['time'])
+ self.assertEqual(event.data, MOCK_RAW_EVENT['data'])
+ self.assertEqual(event.callback_id, MOCK_RAW_EVENT['callbackId'])
+
+ def test_wait_and_get_timeout(self):
+ mock_event_client = mock.Mock()
+ java_timeout_msg = ('com.google.android.mobly.snippet.event.'
+ 'EventSnippet$EventSnippetException: timeout.')
+ mock_event_client.eventWaitAndGet = mock.Mock(
+ side_effect=jsonrpc_client_base.ApiError(mock.Mock(),
+ java_timeout_msg))
+ handler = callback_handler.CallbackHandler(
+ callback_id=MOCK_CALLBACK_ID,
+ event_client=mock_event_client,
+ ret_value=None,
+ method_name=None,
+ ad=mock.Mock())
+ expected_msg = 'Timed out after waiting .*s for event "ha" .*'
+ with self.assertRaisesRegex(callback_handler.TimeoutError,
+ expected_msg):
+ handler.waitAndGet('ha')
+
+ def test_wait_for_event(self):
+ mock_event_client = mock.Mock()
+ mock_event_client.eventWaitAndGet = mock.Mock(
+ return_value=MOCK_RAW_EVENT)
+ handler = callback_handler.CallbackHandler(
+ callback_id=MOCK_CALLBACK_ID,
+ event_client=mock_event_client,
+ ret_value=None,
+ method_name=None,
+ ad=mock.Mock())
+
+ def some_condition(event):
+ return event.data['successful']
+
+ event = handler.waitForEvent('AsyncTaskResult', some_condition, 0.01)
+
+ def test_wait_for_event_negative(self):
+ mock_event_client = mock.Mock()
+ mock_event_client.eventWaitAndGet = mock.Mock(
+ return_value=MOCK_RAW_EVENT)
+ handler = callback_handler.CallbackHandler(
+ callback_id=MOCK_CALLBACK_ID,
+ event_client=mock_event_client,
+ ret_value=None,
+ method_name=None,
+ ad=mock.Mock())
+ expected_msg = (
+ 'Timed out after 0.01s waiting for an "AsyncTaskResult" event that'
+ ' satisfies the predicate "some_condition".')
+
+ def some_condition(event):
+ return False
+
+ with self.assertRaisesRegex(callback_handler.TimeoutError,
+ expected_msg):
+ handler.waitForEvent('AsyncTaskResult', some_condition, 0.01)
+
+ def test_wait_for_event_max_timeout(self):
+ """waitForEvent should not raise the timeout exceed threshold error.
"""
+ mock_event_client = mock.Mock()
+ mock_event_client.eventWaitAndGet = mock.Mock(
+ return_value=MOCK_RAW_EVENT)
+ handler = callback_handler.CallbackHandler(
+ callback_id=MOCK_CALLBACK_ID,
+ event_client=mock_event_client,
+ ret_value=None,
+ method_name=None,
+ ad=mock.Mock())
- def test_timeout_value(self):
- self.assertGreaterEqual(jsonrpc_client_base._SOCKET_READ_TIMEOUT,
- callback_handler.MAX_TIMEOUT)
+ def some_condition(event):
+ return event.data['successful']
- def test_callback_id_property(self):
- mock_event_client = mock.Mock()
- handler = callback_handler.CallbackHandler(
- callback_id=MOCK_CALLBACK_ID,
- event_client=mock_event_client,
- ret_value=None,
- method_name=None,
- ad=mock.Mock())
- self.assertEqual(handler.callback_id, MOCK_CALLBACK_ID)
- with self.assertRaisesRegex(AttributeError, "can't set attribute"):
- handler.callback_id = 'ha'
-
- def test_event_dict_to_snippet_event(self):
- mock_event_client = mock.Mock()
- mock_event_client.eventWaitAndGet = mock.Mock(
- return_value=MOCK_RAW_EVENT)
- handler = callback_handler.CallbackHandler(
- callback_id=MOCK_CALLBACK_ID,
- event_client=mock_event_client,
- ret_value=None,
- method_name=None,
- ad=mock.Mock())
- event = handler.waitAndGet('ha')
- self.assertEqual(event.name, MOCK_RAW_EVENT['name'])
- self.assertEqual(event.creation_time, MOCK_RAW_EVENT['time'])
- self.assertEqual(event.data, MOCK_RAW_EVENT['data'])
- self.assertEqual(event.callback_id, MOCK_RAW_EVENT['callbackId'])
-
- def test_wait_and_get_timeout(self):
- mock_event_client = mock.Mock()
- java_timeout_msg = ('com.google.android.mobly.snippet.event.'
- 'EventSnippet$EventSnippetException: timeout.')
- mock_event_client.eventWaitAndGet = mock.Mock(
- side_effect=jsonrpc_client_base.ApiError(mock.Mock(),
- java_timeout_msg))
- handler = callback_handler.CallbackHandler(
- callback_id=MOCK_CALLBACK_ID,
- event_client=mock_event_client,
- ret_value=None,
- method_name=None,
- ad=mock.Mock())
- expected_msg = 'Timed out after waiting .*s for event "ha" .*'
- with self.assertRaisesRegex(callback_handler.TimeoutError,
- expected_msg):
- handler.waitAndGet('ha')
-
- def test_wait_for_event(self):
- mock_event_client = mock.Mock()
- mock_event_client.eventWaitAndGet = mock.Mock(
- return_value=MOCK_RAW_EVENT)
- handler = callback_handler.CallbackHandler(
- callback_id=MOCK_CALLBACK_ID,
- event_client=mock_event_client,
- ret_value=None,
- method_name=None,
- ad=mock.Mock())
-
- def some_condition(event):
- return event.data['successful']
-
- event = handler.waitForEvent('AsyncTaskResult', some_condition, 0.01)
-
- def test_wait_for_event_negative(self):
- mock_event_client = mock.Mock()
- mock_event_client.eventWaitAndGet = mock.Mock(
- return_value=MOCK_RAW_EVENT)
- handler = callback_handler.CallbackHandler(
- callback_id=MOCK_CALLBACK_ID,
- event_client=mock_event_client,
- ret_value=None,
- method_name=None,
- ad=mock.Mock())
- expected_msg = (
- 'Timed out after 0.01s waiting for an "AsyncTaskResult" event that'
- ' satisfies the predicate "some_condition".')
-
- def some_condition(event):
- return False
-
- with self.assertRaisesRegex(callback_handler.TimeoutError,
- expected_msg):
- handler.waitForEvent('AsyncTaskResult', some_condition, 0.01)
-
- def test_wait_for_event_max_timeout(self):
- """waitForEvent should not raise the timeout exceed threshold error.
- """
- mock_event_client = mock.Mock()
- mock_event_client.eventWaitAndGet = mock.Mock(
- return_value=MOCK_RAW_EVENT)
- handler = callback_handler.CallbackHandler(
- callback_id=MOCK_CALLBACK_ID,
- event_client=mock_event_client,
- ret_value=None,
- method_name=None,
- ad=mock.Mock())
-
- def some_condition(event):
- return event.data['successful']
-
- big_timeout = callback_handler.MAX_TIMEOUT * 2
- # This line should not raise.
- event = handler.waitForEvent(
- 'AsyncTaskResult', some_condition, timeout=big_timeout)
+ big_timeout = callback_handler.MAX_TIMEOUT * 2
+ # This line should not raise.
+ event = handler.waitForEvent(
+ 'AsyncTaskResult', some_condition, timeout=big_timeout)
if __name__ == "__main__":
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/errors_test.py b/tests/mobly/controllers/android_device_lib/errors_test.py
index d88c254..57b87b3 100755
--- a/tests/mobly/controllers/android_device_lib/errors_test.py
+++ b/tests/mobly/controllers/android_device_lib/errors_test.py
@@ -20,30 +20,30 @@
class ErrorsTest(unittest.TestCase):
- def test_device_error(self):
- device = mock.MagicMock()
- device.__repr__ = lambda _: '[MockDevice]'
- exception = errors.DeviceError(device, 'Some error message.')
- self.assertEqual(str(exception), '[MockDevice] Some error message.')
+ def test_device_error(self):
+ device = mock.MagicMock()
+ device.__repr__ = lambda _: '[MockDevice]'
+ exception = errors.DeviceError(device, 'Some error message.')
+ self.assertEqual(str(exception), '[MockDevice] Some error message.')
- def test_service_error(self):
- device = mock.MagicMock()
- device.__repr__ = lambda _: '[MockDevice]'
- exception = errors.ServiceError(device, 'Some error message.')
- self.assertEqual(
- str(exception), '[MockDevice]::Service<None> Some error message.')
+ def test_service_error(self):
+ device = mock.MagicMock()
+ device.__repr__ = lambda _: '[MockDevice]'
+ exception = errors.ServiceError(device, 'Some error message.')
+ self.assertEqual(
+ str(exception), '[MockDevice]::Service<None> Some error message.')
- def test_subclass_service_error(self):
- class Error(errors.ServiceError):
- SERVICE_TYPE = 'SomeType'
+ def test_subclass_service_error(self):
+ class Error(errors.ServiceError):
+ SERVICE_TYPE = 'SomeType'
- device = mock.MagicMock()
- device.__repr__ = lambda _: '[MockDevice]'
- exception = Error(device, 'Some error message.')
- self.assertEqual(
- str(exception),
- '[MockDevice]::Service<SomeType> Some error message.')
+ device = mock.MagicMock()
+ device.__repr__ = lambda _: '[MockDevice]'
+ exception = Error(device, 'Some error message.')
+ self.assertEqual(
+ str(exception),
+ '[MockDevice]::Service<SomeType> Some error message.')
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/jsonrpc_client_base_test.py b/tests/mobly/controllers/android_device_lib/jsonrpc_client_base_test.py
index 583c52c..aef3c73 100755
--- a/tests/mobly/controllers/android_device_lib/jsonrpc_client_base_test.py
+++ b/tests/mobly/controllers/android_device_lib/jsonrpc_client_base_test.py
@@ -24,288 +24,288 @@
class FakeRpcClient(jsonrpc_client_base.JsonRpcClientBase):
- def __init__(self):
- super(FakeRpcClient, self).__init__(
- app_name='FakeRpcClient', ad=mock.Mock())
+ def __init__(self):
+ super(FakeRpcClient, self).__init__(
+ app_name='FakeRpcClient', ad=mock.Mock())
class JsonRpcClientBaseTest(jsonrpc_client_test_base.JsonRpcClientTestBase):
- """Unit tests for mobly.controllers.android_device_lib.jsonrpc_client_base.
+ """Unit tests for mobly.controllers.android_device_lib.jsonrpc_client_base.
+ """
+
+ @mock.patch('socket.create_connection')
+ def test_open_timeout_io_error(self, mock_create_connection):
+ """Test socket timeout with io error
+
+ Test that if the net socket gives an io error, then the client
+ will eventually exit with an IOError.
"""
+ mock_create_connection.side_effect = IOError()
+ with self.assertRaises(IOError):
+ client = FakeRpcClient()
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_open_timeout_io_error(self, mock_create_connection):
- """Test socket timeout with io error
+ @mock.patch('socket.create_connection')
+ def test_connect_timeout(self, mock_create_connection):
+ """Test socket timeout
- Test that if the net socket gives an io error, then the client
- will eventually exit with an IOError.
- """
- mock_create_connection.side_effect = IOError()
- with self.assertRaises(IOError):
- client = FakeRpcClient()
- client.connect()
+ Test that a timeout exception will be raised if the socket gives a
+ timeout.
+ """
+ mock_create_connection.side_effect = socket.timeout
+ with self.assertRaises(socket.timeout):
+ client = FakeRpcClient()
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_connect_timeout(self, mock_create_connection):
- """Test socket timeout
+ @mock.patch('socket.create_connection')
+ def test_handshake_error(self, mock_create_connection):
+ """Test error in jsonrpc handshake
- Test that a timeout exception will be raised if the socket gives a
- timeout.
- """
- mock_create_connection.side_effect = socket.timeout
- with self.assertRaises(socket.timeout):
- client = FakeRpcClient()
- client.connect()
+ Test that if there is an error in the jsonrpc handshake then a protocol
+ error will be raised.
+ """
+ self.setup_mock_socket_file(mock_create_connection, resp=None)
+ client = FakeRpcClient()
+ with self.assertRaisesRegex(
+ jsonrpc_client_base.ProtocolError,
+ jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_HANDSHAKE):
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_handshake_error(self, mock_create_connection):
- """Test error in jsonrpc handshake
+ @mock.patch('socket.create_connection')
+ def test_connect_handshake(self, mock_create_connection):
+ """Test client handshake
- Test that if there is an error in the jsonrpc handshake then a protocol
- error will be raised.
- """
- self.setup_mock_socket_file(mock_create_connection, resp=None)
- client = FakeRpcClient()
- with self.assertRaisesRegex(
- jsonrpc_client_base.ProtocolError,
- jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_HANDSHAKE):
- client.connect()
+ Test that at the end of a handshake with no errors the client object
+ has the correct parameters.
+ """
+ self.setup_mock_socket_file(mock_create_connection)
+ client = FakeRpcClient()
+ client.connect()
+ self.assertEqual(client.uid, 1)
- @mock.patch('socket.create_connection')
- def test_connect_handshake(self, mock_create_connection):
- """Test client handshake
+ @mock.patch('socket.create_connection')
+ def test_connect_handshake_unknown_status(self, mock_create_connection):
+ """Test handshake with unknown status response
- Test that at the end of a handshake with no errors the client object
- has the correct parameters.
- """
- self.setup_mock_socket_file(mock_create_connection)
- client = FakeRpcClient()
- client.connect()
- self.assertEqual(client.uid, 1)
+ Test that when the handshake is given an unknown status then the client
+ will not be given a uid.
+ """
+ self.setup_mock_socket_file(
+ mock_create_connection, resp=self.MOCK_RESP_UNKNOWN_STATUS)
+ client = FakeRpcClient()
+ client.connect()
+ self.assertEqual(client.uid, jsonrpc_client_base.UNKNOWN_UID)
- @mock.patch('socket.create_connection')
- def test_connect_handshake_unknown_status(self, mock_create_connection):
- """Test handshake with unknown status response
+ @mock.patch('socket.create_connection')
+ def test_rpc_error_response(self, mock_create_connection):
+ """Test rpc that is given an error response
- Test that when the handshake is given an unknown status then the client
- will not be given a uid.
- """
- self.setup_mock_socket_file(
- mock_create_connection, resp=self.MOCK_RESP_UNKNOWN_STATUS)
- client = FakeRpcClient()
- client.connect()
- self.assertEqual(client.uid, jsonrpc_client_base.UNKNOWN_UID)
+ Test that when an rpc receives a response with an error will raised
+ an api error.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
- @mock.patch('socket.create_connection')
- def test_rpc_error_response(self, mock_create_connection):
- """Test rpc that is given an error response
+ client = FakeRpcClient()
+ client.connect()
- Test that when an rpc receives a response with an error will raised
- an api error.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
+ fake_file.resp = self.MOCK_RESP_WITH_ERROR
- client = FakeRpcClient()
- client.connect()
+ with self.assertRaisesRegex(jsonrpc_client_base.ApiError, '1'):
+ client.some_rpc(1, 2, 3)
- fake_file.resp = self.MOCK_RESP_WITH_ERROR
+ @mock.patch('socket.create_connection')
+ def test_rpc_callback_response(self, mock_create_connection):
+ """Test rpc that is given a callback response.
- with self.assertRaisesRegex(jsonrpc_client_base.ApiError, '1'):
- client.some_rpc(1, 2, 3)
+ Test that when an rpc receives a callback response, a callback object is
+ created correctly.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
- @mock.patch('socket.create_connection')
- def test_rpc_callback_response(self, mock_create_connection):
- """Test rpc that is given a callback response.
+ client = FakeRpcClient()
+ client.connect()
- Test that when an rpc receives a callback response, a callback object is
- created correctly.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
+ fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
+ client._event_client = mock.Mock()
- client = FakeRpcClient()
- client.connect()
+ callback = client.some_rpc(1, 2, 3)
+ self.assertEqual(callback.ret_value, 123)
+ self.assertEqual(callback._id, '1-0')
- fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
- client._event_client = mock.Mock()
+ @mock.patch('socket.create_connection')
+ def test_rpc_id_mismatch(self, mock_create_connection):
+ """Test rpc that returns a different id than expected
- callback = client.some_rpc(1, 2, 3)
- self.assertEqual(callback.ret_value, 123)
- self.assertEqual(callback._id, '1-0')
+ Test that if an rpc returns with an id that is different than what
+ is expected will give a protocl error.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
- @mock.patch('socket.create_connection')
- def test_rpc_id_mismatch(self, mock_create_connection):
- """Test rpc that returns a different id than expected
+ client = FakeRpcClient()
+ client.connect()
- Test that if an rpc returns with an id that is different than what
- is expected will give a protocl error.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
+ fake_file.resp = (self.MOCK_RESP_TEMPLATE % 52).encode('utf8')
- client = FakeRpcClient()
- client.connect()
+ with self.assertRaisesRegex(
+ jsonrpc_client_base.ProtocolError,
+ jsonrpc_client_base.ProtocolError.MISMATCHED_API_ID):
+ client.some_rpc(1, 2, 3)
- fake_file.resp = (self.MOCK_RESP_TEMPLATE % 52).encode('utf8')
+ @mock.patch('socket.create_connection')
+ def test_rpc_no_response(self, mock_create_connection):
+ """Test rpc that does not get a response
- with self.assertRaisesRegex(
- jsonrpc_client_base.ProtocolError,
- jsonrpc_client_base.ProtocolError.MISMATCHED_API_ID):
- client.some_rpc(1, 2, 3)
+ Test that when an rpc does not get a response it throws a protocol
+ error.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
- @mock.patch('socket.create_connection')
- def test_rpc_no_response(self, mock_create_connection):
- """Test rpc that does not get a response
+ client = FakeRpcClient()
+ client.connect()
- Test that when an rpc does not get a response it throws a protocol
- error.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
+ fake_file.resp = None
- client = FakeRpcClient()
- client.connect()
+ with self.assertRaisesRegex(
+ jsonrpc_client_base.ProtocolError,
+ jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_SERVER):
+ client.some_rpc(1, 2, 3)
- fake_file.resp = None
+ @mock.patch('socket.create_connection')
+ def test_rpc_send_to_socket(self, mock_create_connection):
+ """Test rpc sending and recieving
- with self.assertRaisesRegex(
- jsonrpc_client_base.ProtocolError,
- jsonrpc_client_base.ProtocolError.NO_RESPONSE_FROM_SERVER):
- client.some_rpc(1, 2, 3)
+ Tests that when an rpc is sent and received the corrent data
+ is used.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
- @mock.patch('socket.create_connection')
- def test_rpc_send_to_socket(self, mock_create_connection):
- """Test rpc sending and recieving
+ client = FakeRpcClient()
+ client.connect()
- Tests that when an rpc is sent and received the corrent data
- is used.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
+ result = client.some_rpc(1, 2, 3)
+ self.assertEqual(result, 123)
- client = FakeRpcClient()
- client.connect()
+ expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
+ actual = json.loads(fake_file.last_write.decode('utf-8'))
- result = client.some_rpc(1, 2, 3)
- self.assertEqual(result, 123)
+ self.assertEqual(expected, actual)
- expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
- actual = json.loads(fake_file.last_write.decode('utf-8'))
+ @mock.patch('socket.create_connection')
+ def test_rpc_send_to_socket_without_callback(self, mock_create_connection):
+ """Test rpc sending and recieving with Rpc protocol before callback was
+ added to the resp message.
- self.assertEqual(expected, actual)
+ Logic is the same as test_rpc_send_to_socket.
+ """
+ fake_file = self.setup_mock_socket_file(
+ mock_create_connection, resp=self.MOCK_RESP_WITHOUT_CALLBACK)
- @mock.patch('socket.create_connection')
- def test_rpc_send_to_socket_without_callback(self, mock_create_connection):
- """Test rpc sending and recieving with Rpc protocol before callback was
- added to the resp message.
+ client = FakeRpcClient()
+ client.connect()
- Logic is the same as test_rpc_send_to_socket.
- """
- fake_file = self.setup_mock_socket_file(
- mock_create_connection, resp=self.MOCK_RESP_WITHOUT_CALLBACK)
+ result = client.some_rpc(1, 2, 3)
+ self.assertEqual(result, 123)
- client = FakeRpcClient()
- client.connect()
+ expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
+ actual = json.loads(fake_file.last_write.decode('utf-8'))
- result = client.some_rpc(1, 2, 3)
- self.assertEqual(result, 123)
+ self.assertEqual(expected, actual)
- expected = {'id': 0, 'method': 'some_rpc', 'params': [1, 2, 3]}
- actual = json.loads(fake_file.last_write.decode('utf-8'))
+ @mock.patch('socket.create_connection')
+ def test_rpc_call_increment_counter(self, mock_create_connection):
+ """Test rpc counter
- self.assertEqual(expected, actual)
+ Test that with each rpc call the counter is incremented by 1.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
- @mock.patch('socket.create_connection')
- def test_rpc_call_increment_counter(self, mock_create_connection):
- """Test rpc counter
+ client = FakeRpcClient()
+ client.connect()
- Test that with each rpc call the counter is incremented by 1.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
+ for i in range(0, 10):
+ fake_file.resp = (self.MOCK_RESP_TEMPLATE % i).encode('utf-8')
+ client.some_rpc()
- client = FakeRpcClient()
- client.connect()
+ self.assertEqual(next(client._counter), 10)
- for i in range(0, 10):
- fake_file.resp = (self.MOCK_RESP_TEMPLATE % i).encode('utf-8')
- client.some_rpc()
+ @mock.patch('socket.create_connection')
+ def test_rpc_verbose_logging_with_long_string(self,
+ mock_create_connection):
+ """Test rpc response fully write into DEBUG level log."""
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
+ testing_rpc_response = self.generate_rpc_response(4000)
+ fake_file.resp = testing_rpc_response
- self.assertEqual(next(client._counter), 10)
+ client = FakeRpcClient()
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_rpc_verbose_logging_with_long_string(self,
- mock_create_connection):
- """Test rpc response fully write into DEBUG level log."""
- fake_file = self.setup_mock_socket_file(mock_create_connection)
- testing_rpc_response = self.generate_rpc_response(4000)
- fake_file.resp = testing_rpc_response
+ response = client._client_receive()
+ self.assertEqual(response, testing_rpc_response)
- client = FakeRpcClient()
- client.connect()
+ client.log.debug.assert_called_with('Snippet received: %s',
+ testing_rpc_response)
- response = client._client_receive()
- self.assertEqual(response, testing_rpc_response)
+ @mock.patch('socket.create_connection')
+ def test_rpc_truncated_logging_short_response(self,
+ mock_create_connection):
+ """Test rpc response will full logged when length is short."""
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
+ testing_rpc_response = self.generate_rpc_response(
+ int(jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH / 2))
+ fake_file.resp = testing_rpc_response
- client.log.debug.assert_called_with('Snippet received: %s',
- testing_rpc_response)
+ client = FakeRpcClient()
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_rpc_truncated_logging_short_response(self,
- mock_create_connection):
- """Test rpc response will full logged when length is short."""
- fake_file = self.setup_mock_socket_file(mock_create_connection)
- testing_rpc_response = self.generate_rpc_response(
- int(jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH / 2))
- fake_file.resp = testing_rpc_response
+ client.set_snippet_client_verbose_logging(False)
+ response = client._client_receive()
- client = FakeRpcClient()
- client.connect()
+ self.assertEqual(response, testing_rpc_response)
+ client.log.debug.assert_called_with('Snippet received: %s',
+ testing_rpc_response)
- client.set_snippet_client_verbose_logging(False)
- response = client._client_receive()
+ @mock.patch('socket.create_connection')
+ def test_rpc_truncated_logging_fit_size_response(self,
+ mock_create_connection):
+ """Test rpc response will full logged when length is equal to threshold.
+ """
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
+ testing_rpc_response = self.generate_rpc_response(
+ jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)
+ fake_file.resp = testing_rpc_response
- self.assertEqual(response, testing_rpc_response)
- client.log.debug.assert_called_with('Snippet received: %s',
- testing_rpc_response)
+ client = FakeRpcClient()
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_rpc_truncated_logging_fit_size_response(self,
- mock_create_connection):
- """Test rpc response will full logged when length is equal to threshold.
- """
- fake_file = self.setup_mock_socket_file(mock_create_connection)
- testing_rpc_response = self.generate_rpc_response(
- jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)
- fake_file.resp = testing_rpc_response
+ client.set_snippet_client_verbose_logging(False)
+ response = client._client_receive()
- client = FakeRpcClient()
- client.connect()
+ self.assertEqual(response, testing_rpc_response)
+ client.log.debug.assert_called_with('Snippet received: %s',
+ testing_rpc_response)
- client.set_snippet_client_verbose_logging(False)
- response = client._client_receive()
+ @mock.patch('socket.create_connection')
+ def test_rpc_truncated_logging_long_response(self, mock_create_connection):
+ """Test rpc response truncated with given length in DEBUG level log."""
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
+ resp_len = jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH * 40
+ testing_rpc_response = self.generate_rpc_response(resp_len)
+ fake_file.resp = testing_rpc_response
- self.assertEqual(response, testing_rpc_response)
- client.log.debug.assert_called_with('Snippet received: %s',
- testing_rpc_response)
+ client = FakeRpcClient()
+ client.connect()
- @mock.patch('socket.create_connection')
- def test_rpc_truncated_logging_long_response(self, mock_create_connection):
- """Test rpc response truncated with given length in DEBUG level log."""
- fake_file = self.setup_mock_socket_file(mock_create_connection)
- resp_len = jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH * 40
- testing_rpc_response = self.generate_rpc_response(resp_len)
- fake_file.resp = testing_rpc_response
+ client.set_snippet_client_verbose_logging(False)
+ response = client._client_receive()
- client = FakeRpcClient()
- client.connect()
-
- client.set_snippet_client_verbose_logging(False)
- response = client._client_receive()
-
- self.assertEqual(response, testing_rpc_response)
- # DEBUG level log should truncated by given length.
- client.log.debug.assert_called_with(
- 'Snippet received: %s... %d chars are truncated',
- testing_rpc_response
- [:jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH],
- resp_len - jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)
+ self.assertEqual(response, testing_rpc_response)
+ # DEBUG level log should truncated by given length.
+ client.log.debug.assert_called_with(
+ 'Snippet received: %s... %d chars are truncated',
+ testing_rpc_response
+ [:jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH],
+ resp_len - jsonrpc_client_base._MAX_RPC_RESP_LOGGING_LENGTH)
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/jsonrpc_shell_base_test.py b/tests/mobly/controllers/android_device_lib/jsonrpc_shell_base_test.py
index 677cb89..fb9db4f 100755
--- a/tests/mobly/controllers/android_device_lib/jsonrpc_shell_base_test.py
+++ b/tests/mobly/controllers/android_device_lib/jsonrpc_shell_base_test.py
@@ -22,70 +22,70 @@
class JsonRpcClientBaseTest(unittest.TestCase):
- """Unit tests for mobly.controllers.android_device_lib.jsonrpc_shell_base.
- """
+ """Unit tests for mobly.controllers.android_device_lib.jsonrpc_shell_base.
+ """
- @mock.patch.object(android_device, 'list_adb_devices')
- @mock.patch.object(android_device, 'get_instances')
- @mock.patch.object(os, 'environ', new={})
- def test_load_device(self, mock_get_instances, mock_list_adb_devices):
- mock_list_adb_devices.return_value = ['1234', '4312']
- mock_device = mock.MagicMock(spec=android_device.AndroidDevice)
- mock_get_instances.return_value = [mock_device]
- json_shell = jsonrpc_shell_base.JsonRpcShellBase()
- json_shell.load_device(serial='1234')
- self.assertEqual(json_shell._ad, mock_device)
+ @mock.patch.object(android_device, 'list_adb_devices')
+ @mock.patch.object(android_device, 'get_instances')
+ @mock.patch.object(os, 'environ', new={})
+ def test_load_device(self, mock_get_instances, mock_list_adb_devices):
+ mock_list_adb_devices.return_value = ['1234', '4312']
+ mock_device = mock.MagicMock(spec=android_device.AndroidDevice)
+ mock_get_instances.return_value = [mock_device]
+ json_shell = jsonrpc_shell_base.JsonRpcShellBase()
+ json_shell.load_device(serial='1234')
+ self.assertEqual(json_shell._ad, mock_device)
- @mock.patch.object(android_device, 'list_adb_devices')
- @mock.patch.object(android_device, 'get_instances')
- @mock.patch.object(os, 'environ', new={})
- def test_load_device_when_one_device(self, mock_get_instances,
- mock_list_adb_devices):
- mock_list_adb_devices.return_value = ['1234']
- mock_device = mock.MagicMock(spec=android_device.AndroidDevice)
- mock_get_instances.return_value = [mock_device]
- json_shell = jsonrpc_shell_base.JsonRpcShellBase()
- json_shell.load_device()
- self.assertEqual(json_shell._ad, mock_device)
+ @mock.patch.object(android_device, 'list_adb_devices')
+ @mock.patch.object(android_device, 'get_instances')
+ @mock.patch.object(os, 'environ', new={})
+ def test_load_device_when_one_device(self, mock_get_instances,
+ mock_list_adb_devices):
+ mock_list_adb_devices.return_value = ['1234']
+ mock_device = mock.MagicMock(spec=android_device.AndroidDevice)
+ mock_get_instances.return_value = [mock_device]
+ json_shell = jsonrpc_shell_base.JsonRpcShellBase()
+ json_shell.load_device()
+ self.assertEqual(json_shell._ad, mock_device)
- @mock.patch.object(android_device, 'list_adb_devices')
- @mock.patch.object(android_device, 'get_instances')
- @mock.patch.object(os, 'environ', new={'ANDROID_SERIAL': '1234'})
- def test_load_device_when_android_serial(self, mock_get_instances,
- mock_list_adb_devices):
- mock_list_adb_devices.return_value = ['1234', '4321']
- mock_device = mock.MagicMock(spec=android_device.AndroidDevice)
- mock_get_instances.return_value = [mock_device]
- json_shell = jsonrpc_shell_base.JsonRpcShellBase()
- json_shell.load_device()
- self.assertEqual(json_shell._ad, mock_device)
+ @mock.patch.object(android_device, 'list_adb_devices')
+ @mock.patch.object(android_device, 'get_instances')
+ @mock.patch.object(os, 'environ', new={'ANDROID_SERIAL': '1234'})
+ def test_load_device_when_android_serial(self, mock_get_instances,
+ mock_list_adb_devices):
+ mock_list_adb_devices.return_value = ['1234', '4321']
+ mock_device = mock.MagicMock(spec=android_device.AndroidDevice)
+ mock_get_instances.return_value = [mock_device]
+ json_shell = jsonrpc_shell_base.JsonRpcShellBase()
+ json_shell.load_device()
+ self.assertEqual(json_shell._ad, mock_device)
- @mock.patch.object(android_device, 'list_adb_devices')
- def test_load_device_when_no_devices(self, mock_list_adb_devices):
- mock_list_adb_devices.return_value = []
- json_shell = jsonrpc_shell_base.JsonRpcShellBase()
- with self.assertRaisesRegex(jsonrpc_shell_base.Error,
- 'No adb device found!'):
- json_shell.load_device()
+ @mock.patch.object(android_device, 'list_adb_devices')
+ def test_load_device_when_no_devices(self, mock_list_adb_devices):
+ mock_list_adb_devices.return_value = []
+ json_shell = jsonrpc_shell_base.JsonRpcShellBase()
+ with self.assertRaisesRegex(jsonrpc_shell_base.Error,
+ 'No adb device found!'):
+ json_shell.load_device()
- @mock.patch.object(android_device, 'list_adb_devices')
- @mock.patch.object(os, 'environ', new={})
- def test_load_device_when_unspecified_device(self, mock_list_adb_devices):
- mock_list_adb_devices.return_value = ['1234', '4321']
- json_shell = jsonrpc_shell_base.JsonRpcShellBase()
- with self.assertRaisesRegex(jsonrpc_shell_base.Error,
- 'Expected one phone.*'):
- json_shell.load_device()
+ @mock.patch.object(android_device, 'list_adb_devices')
+ @mock.patch.object(os, 'environ', new={})
+ def test_load_device_when_unspecified_device(self, mock_list_adb_devices):
+ mock_list_adb_devices.return_value = ['1234', '4321']
+ json_shell = jsonrpc_shell_base.JsonRpcShellBase()
+ with self.assertRaisesRegex(jsonrpc_shell_base.Error,
+ 'Expected one phone.*'):
+ json_shell.load_device()
- @mock.patch.object(android_device, 'list_adb_devices')
- @mock.patch.object(os, 'environ', new={})
- def test_load_device_when_device_not_found(self, mock_list_adb_devices):
- mock_list_adb_devices.return_value = ['4321']
- json_shell = jsonrpc_shell_base.JsonRpcShellBase()
- with self.assertRaisesRegex(jsonrpc_shell_base.Error,
- 'Device "1234" is not found by adb.'):
- json_shell.load_device(serial='1234')
+ @mock.patch.object(android_device, 'list_adb_devices')
+ @mock.patch.object(os, 'environ', new={})
+ def test_load_device_when_device_not_found(self, mock_list_adb_devices):
+ mock_list_adb_devices.return_value = ['4321']
+ json_shell = jsonrpc_shell_base.JsonRpcShellBase()
+ with self.assertRaisesRegex(jsonrpc_shell_base.Error,
+ 'Device "1234" is not found by adb.'):
+ json_shell.load_device(serial='1234')
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/service_manager_test.py b/tests/mobly/controllers/android_device_lib/service_manager_test.py
index c989e83..90efb1d 100755
--- a/tests/mobly/controllers/android_device_lib/service_manager_test.py
+++ b/tests/mobly/controllers/android_device_lib/service_manager_test.py
@@ -23,452 +23,452 @@
class MockService(base_service.BaseService):
- def __init__(self, device, configs=None):
- self._device = device
- self._configs = configs
- self._alive = False
- self.start_func = mock.MagicMock()
- self.stop_func = mock.MagicMock()
- self.pause_func = mock.MagicMock()
- self.resume_func = mock.MagicMock()
+ def __init__(self, device, configs=None):
+ self._device = device
+ self._configs = configs
+ self._alive = False
+ self.start_func = mock.MagicMock()
+ self.stop_func = mock.MagicMock()
+ self.pause_func = mock.MagicMock()
+ self.resume_func = mock.MagicMock()
- @property
- def is_alive(self):
- return self._alive
+ @property
+ def is_alive(self):
+ return self._alive
- def start(self, configs=None):
- self.start_func(configs)
- self._alive = True
+ def start(self, configs=None):
+ self.start_func(configs)
+ self._alive = True
- def stop(self):
- self.stop_func()
- self._alive = False
+ def stop(self):
+ self.stop_func()
+ self._alive = False
- def pause(self):
- self.pause_func()
- self._alive = False
+ def pause(self):
+ self.pause_func()
+ self._alive = False
- def resume(self):
- self.resume_func()
- self._alive = True
+ def resume(self):
+ self.resume_func()
+ self._alive = True
class ServiceManagerTest(unittest.TestCase):
- def setUp(self):
- # Reset hidden global `expects` state.
- if sys.version_info < (3, 0):
- reload(expects)
- else:
- import importlib
- importlib.reload(expects)
+ def setUp(self):
+ # Reset hidden global `expects` state.
+ if sys.version_info < (3, 0):
+ reload(expects)
+ else:
+ import importlib
+ importlib.reload(expects)
- def assert_recorded_one_error(self, message):
- self.assertEqual(expects.recorder.error_count, 1)
- for _, error in (
- expects.DEFAULT_TEST_RESULT_RECORD.extra_errors.items()):
- self.assertIn(message, error.details)
+ def assert_recorded_one_error(self, message):
+ self.assertEqual(expects.recorder.error_count, 1)
+ for _, error in (
+ expects.DEFAULT_TEST_RESULT_RECORD.extra_errors.items()):
+ self.assertIn(message, error.details)
- def test_service_manager_instantiation(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
+ def test_service_manager_instantiation(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
- def test_register(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService)
- service = manager.mock_service
- self.assertTrue(service)
- self.assertTrue(service.is_alive)
- self.assertTrue(manager.is_any_alive)
- self.assertEqual(service.alias, 'mock_service')
- self.assertEqual(service.start_func.call_count, 1)
+ def test_register(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService)
+ service = manager.mock_service
+ self.assertTrue(service)
+ self.assertTrue(service.is_alive)
+ self.assertTrue(manager.is_any_alive)
+ self.assertEqual(service.alias, 'mock_service')
+ self.assertEqual(service.start_func.call_count, 1)
- def test_register_with_configs(self):
- mock_configs = mock.MagicMock()
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService, configs=mock_configs)
- service = manager.mock_service
- self.assertTrue(service)
- self.assertEqual(service._configs, mock_configs)
- self.assertEqual(service.start_func.call_count, 1)
+ def test_register_with_configs(self):
+ mock_configs = mock.MagicMock()
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService, configs=mock_configs)
+ service = manager.mock_service
+ self.assertTrue(service)
+ self.assertEqual(service._configs, mock_configs)
+ self.assertEqual(service.start_func.call_count, 1)
- def test_register_do_not_start_service(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService, start_service=False)
- service = manager.mock_service
- self.assertTrue(service)
- self.assertFalse(service.is_alive)
- self.assertFalse(manager.is_any_alive)
- self.assertEqual(service.start_func.call_count, 0)
+ def test_register_do_not_start_service(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService, start_service=False)
+ service = manager.mock_service
+ self.assertTrue(service)
+ self.assertFalse(service.is_alive)
+ self.assertFalse(manager.is_any_alive)
+ self.assertEqual(service.start_func.call_count, 0)
- def test_register_not_a_class(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- with self.assertRaisesRegex(service_manager.Error,
- '.* is not a class!'):
- manager.register('mock_service', base_service)
+ def test_register_not_a_class(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ with self.assertRaisesRegex(service_manager.Error,
+ '.* is not a class!'):
+ manager.register('mock_service', base_service)
- def test_register_wrong_subclass_type(self):
- class MyClass(object):
- pass
+ def test_register_wrong_subclass_type(self):
+ class MyClass(object):
+ pass
- manager = service_manager.ServiceManager(mock.MagicMock())
- with self.assertRaisesRegex(service_manager.Error,
- '.* is not a subclass of BaseService!'):
- manager.register('mock_service', MyClass)
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ with self.assertRaisesRegex(service_manager.Error,
+ '.* is not a subclass of BaseService!'):
+ manager.register('mock_service', MyClass)
- def test_register_dup_alias(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService)
- msg = '.* A service is already registered with alias "mock_service"'
- with self.assertRaisesRegex(service_manager.Error, msg):
- manager.register('mock_service', MockService)
+ def test_register_dup_alias(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService)
+ msg = '.* A service is already registered with alias "mock_service"'
+ with self.assertRaisesRegex(service_manager.Error, msg):
+ manager.register('mock_service', MockService)
- def test_for_each(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- service1.ha = mock.MagicMock()
- service2.ha = mock.MagicMock()
- manager.for_each(lambda service: service.ha())
- service1.ha.assert_called_with()
- service2.ha.assert_called_with()
+ def test_for_each(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ service1.ha = mock.MagicMock()
+ service2.ha = mock.MagicMock()
+ manager.for_each(lambda service: service.ha())
+ service1.ha.assert_called_with()
+ service2.ha.assert_called_with()
- def test_for_each_modify_during_iteration(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- service1.ha = mock.MagicMock()
- service2.ha = mock.MagicMock()
- manager.for_each(lambda service: manager._service_objects.pop(service.
- alias))
- self.assertFalse(manager._service_objects)
+ def test_for_each_modify_during_iteration(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ service1.ha = mock.MagicMock()
+ service2.ha = mock.MagicMock()
+ manager.for_each(lambda service: manager._service_objects.pop(service.
+ alias))
+ self.assertFalse(manager._service_objects)
- def test_for_each_one_fail(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- service1.ha = mock.MagicMock()
- service1.ha.side_effect = Exception('Failure in service1.')
- service2.ha = mock.MagicMock()
- manager.for_each(lambda service: service.ha())
- service1.ha.assert_called_with()
- service2.ha.assert_called_with()
- self.assert_recorded_one_error('Failure in service1.')
+ def test_for_each_one_fail(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ service1.ha = mock.MagicMock()
+ service1.ha.side_effect = Exception('Failure in service1.')
+ service2.ha = mock.MagicMock()
+ manager.for_each(lambda service: service.ha())
+ service1.ha.assert_called_with()
+ service2.ha.assert_called_with()
+ self.assert_recorded_one_error('Failure in service1.')
- def test_create_output_excerpts_all(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- manager.register('mock_service3', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- service3 = manager.mock_service3
- service1.create_output_excerpts = mock.MagicMock()
- service2.create_output_excerpts = mock.MagicMock()
- service3.create_output_excerpts = mock.MagicMock()
- service1.create_output_excerpts.return_value = ['path/to/1.txt']
- service2.create_output_excerpts.return_value = [
- 'path/to/2-1.txt', 'path/to/2-2.txt'
- ]
- service3.create_output_excerpts.return_value = []
- mock_test_info = mock.MagicMock(output_path='path/to')
- result = manager.create_output_excerpts_all(mock_test_info)
- self.assertEqual(result['mock_service1'], ['path/to/1.txt'])
- self.assertEqual(result['mock_service2'],
- ['path/to/2-1.txt', 'path/to/2-2.txt'])
- self.assertEqual(result['mock_service3'], [])
+ def test_create_output_excerpts_all(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ manager.register('mock_service3', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ service3 = manager.mock_service3
+ service1.create_output_excerpts = mock.MagicMock()
+ service2.create_output_excerpts = mock.MagicMock()
+ service3.create_output_excerpts = mock.MagicMock()
+ service1.create_output_excerpts.return_value = ['path/to/1.txt']
+ service2.create_output_excerpts.return_value = [
+ 'path/to/2-1.txt', 'path/to/2-2.txt'
+ ]
+ service3.create_output_excerpts.return_value = []
+ mock_test_info = mock.MagicMock(output_path='path/to')
+ result = manager.create_output_excerpts_all(mock_test_info)
+ self.assertEqual(result['mock_service1'], ['path/to/1.txt'])
+ self.assertEqual(result['mock_service2'],
+ ['path/to/2-1.txt', 'path/to/2-2.txt'])
+ self.assertEqual(result['mock_service3'], [])
- def test_unregister(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService)
- service = manager.mock_service
- manager.unregister('mock_service')
- self.assertFalse(manager.is_any_alive)
- self.assertFalse(service.is_alive)
- self.assertEqual(service.stop_func.call_count, 1)
+ def test_unregister(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService)
+ service = manager.mock_service
+ manager.unregister('mock_service')
+ self.assertFalse(manager.is_any_alive)
+ self.assertFalse(service.is_alive)
+ self.assertEqual(service.stop_func.call_count, 1)
- def test_unregister_not_started_service(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService, start_service=False)
- service = manager.mock_service
- manager.unregister('mock_service')
- self.assertFalse(manager.is_any_alive)
- self.assertFalse(service.is_alive)
- self.assertEqual(service.stop_func.call_count, 0)
+ def test_unregister_not_started_service(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService, start_service=False)
+ service = manager.mock_service
+ manager.unregister('mock_service')
+ self.assertFalse(manager.is_any_alive)
+ self.assertFalse(service.is_alive)
+ self.assertEqual(service.stop_func.call_count, 0)
- def test_unregister_non_existent(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- with self.assertRaisesRegex(
- service_manager.Error,
- '.* No service is registered with alias "mock_service"'):
- manager.unregister('mock_service')
+ def test_unregister_non_existent(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ with self.assertRaisesRegex(
+ service_manager.Error,
+ '.* No service is registered with alias "mock_service"'):
+ manager.unregister('mock_service')
- def test_unregister_handle_error_from_stop(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service', MockService)
- service = manager.mock_service
- service.stop_func.side_effect = Exception('Something failed in stop.')
- manager.unregister('mock_service')
- self.assert_recorded_one_error(
- 'Failed to stop service instance "mock_service".')
+ def test_unregister_handle_error_from_stop(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service', MockService)
+ service = manager.mock_service
+ service.stop_func.side_effect = Exception('Something failed in stop.')
+ manager.unregister('mock_service')
+ self.assert_recorded_one_error(
+ 'Failed to stop service instance "mock_service".')
- def test_unregister_all(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- manager.unregister_all()
- self.assertFalse(manager.is_any_alive)
- self.assertFalse(service1.is_alive)
- self.assertFalse(service2.is_alive)
- self.assertEqual(service1.stop_func.call_count, 1)
- self.assertEqual(service2.stop_func.call_count, 1)
+ def test_unregister_all(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ manager.unregister_all()
+ self.assertFalse(manager.is_any_alive)
+ self.assertFalse(service1.is_alive)
+ self.assertFalse(service2.is_alive)
+ self.assertEqual(service1.stop_func.call_count, 1)
+ self.assertEqual(service2.stop_func.call_count, 1)
- def test_unregister_all_with_some_failed(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service1.stop_func.side_effect = Exception('Something failed in stop.')
- service2 = manager.mock_service2
- manager.unregister_all()
- self.assertFalse(manager.is_any_alive)
- self.assertTrue(service1.is_alive)
- self.assertFalse(service2.is_alive)
- self.assert_recorded_one_error(
- 'Failed to stop service instance "mock_service1".')
+ def test_unregister_all_with_some_failed(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service1.stop_func.side_effect = Exception('Something failed in stop.')
+ service2 = manager.mock_service2
+ manager.unregister_all()
+ self.assertFalse(manager.is_any_alive)
+ self.assertTrue(service1.is_alive)
+ self.assertFalse(service2.is_alive)
+ self.assert_recorded_one_error(
+ 'Failed to stop service instance "mock_service1".')
- def test_start_all(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService, start_service=False)
- manager.register('mock_service2', MockService, start_service=False)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- mock_call_tracker = mock.Mock()
- mock_call_tracker.start1 = service1.start_func
- mock_call_tracker.start2 = service2.start_func
- manager.start_all()
- self.assertTrue(service1.is_alive)
- self.assertTrue(service2.is_alive)
- self.assertEqual(service1.start_func.call_count, 1)
- self.assertEqual(service2.start_func.call_count, 1)
- self.assertEqual(mock_call_tracker.mock_calls,
- [mock.call.start1(None),
- mock.call.start2(None)])
+ def test_start_all(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService, start_service=False)
+ manager.register('mock_service2', MockService, start_service=False)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ mock_call_tracker = mock.Mock()
+ mock_call_tracker.start1 = service1.start_func
+ mock_call_tracker.start2 = service2.start_func
+ manager.start_all()
+ self.assertTrue(service1.is_alive)
+ self.assertTrue(service2.is_alive)
+ self.assertEqual(service1.start_func.call_count, 1)
+ self.assertEqual(service2.start_func.call_count, 1)
+ self.assertEqual(mock_call_tracker.mock_calls,
+ [mock.call.start1(None),
+ mock.call.start2(None)])
- def test_start_all_with_already_started_services(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService, start_service=False)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- manager.start_all()
- manager.start_all()
- self.assertTrue(service1.is_alive)
- self.assertTrue(service2.is_alive)
- self.assertEqual(service1.start_func.call_count, 1)
- self.assertEqual(service2.start_func.call_count, 1)
+ def test_start_all_with_already_started_services(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService, start_service=False)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ manager.start_all()
+ manager.start_all()
+ self.assertTrue(service1.is_alive)
+ self.assertTrue(service2.is_alive)
+ self.assertEqual(service1.start_func.call_count, 1)
+ self.assertEqual(service2.start_func.call_count, 1)
- def test_start_all_with_some_failed(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService, start_service=False)
- manager.register('mock_service2', MockService, start_service=False)
- service1 = manager.mock_service1
- service1.start_func.side_effect = Exception(
- 'Something failed in start.')
- service2 = manager.mock_service2
- manager.start_all()
- self.assertFalse(service1.is_alive)
- self.assertTrue(service2.is_alive)
- self.assert_recorded_one_error(
- 'Failed to start service "mock_service1"')
+ def test_start_all_with_some_failed(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService, start_service=False)
+ manager.register('mock_service2', MockService, start_service=False)
+ service1 = manager.mock_service1
+ service1.start_func.side_effect = Exception(
+ 'Something failed in start.')
+ service2 = manager.mock_service2
+ manager.start_all()
+ self.assertFalse(service1.is_alive)
+ self.assertTrue(service2.is_alive)
+ self.assert_recorded_one_error(
+ 'Failed to start service "mock_service1"')
- def test_stop_all(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- mock_call_tracker = mock.Mock()
- mock_call_tracker.stop1 = service1.stop_func
- mock_call_tracker.stop2 = service2.stop_func
- manager.stop_all()
- self.assertFalse(service1.is_alive)
- self.assertFalse(service2.is_alive)
- self.assertEqual(
- mock_call_tracker.mock_calls,
- [mock.call.stop2(), mock.call.stop1()])
- self.assertEqual(service1.start_func.call_count, 1)
- self.assertEqual(service2.start_func.call_count, 1)
- self.assertEqual(service1.stop_func.call_count, 1)
- self.assertEqual(service2.stop_func.call_count, 1)
+ def test_stop_all(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ mock_call_tracker = mock.Mock()
+ mock_call_tracker.stop1 = service1.stop_func
+ mock_call_tracker.stop2 = service2.stop_func
+ manager.stop_all()
+ self.assertFalse(service1.is_alive)
+ self.assertFalse(service2.is_alive)
+ self.assertEqual(
+ mock_call_tracker.mock_calls,
+ [mock.call.stop2(), mock.call.stop1()])
+ self.assertEqual(service1.start_func.call_count, 1)
+ self.assertEqual(service2.start_func.call_count, 1)
+ self.assertEqual(service1.stop_func.call_count, 1)
+ self.assertEqual(service2.stop_func.call_count, 1)
- def test_stop_all_with_already_stopped_services(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService, start_service=False)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- manager.stop_all()
- manager.stop_all()
- self.assertFalse(service1.is_alive)
- self.assertFalse(service2.is_alive)
- self.assertEqual(service1.start_func.call_count, 1)
- self.assertEqual(service2.start_func.call_count, 0)
- self.assertEqual(service1.stop_func.call_count, 1)
- self.assertEqual(service2.stop_func.call_count, 0)
+ def test_stop_all_with_already_stopped_services(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService, start_service=False)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ manager.stop_all()
+ manager.stop_all()
+ self.assertFalse(service1.is_alive)
+ self.assertFalse(service2.is_alive)
+ self.assertEqual(service1.start_func.call_count, 1)
+ self.assertEqual(service2.start_func.call_count, 0)
+ self.assertEqual(service1.stop_func.call_count, 1)
+ self.assertEqual(service2.stop_func.call_count, 0)
- def test_stop_all_with_some_failed(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service1.stop_func.side_effect = Exception(
- 'Something failed in start.')
- service2 = manager.mock_service2
- manager.stop_all()
- self.assertTrue(service1.is_alive)
- self.assertFalse(service2.is_alive)
- self.assert_recorded_one_error(
- 'Failed to stop service "mock_service1"')
+ def test_stop_all_with_some_failed(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service1.stop_func.side_effect = Exception(
+ 'Something failed in start.')
+ service2 = manager.mock_service2
+ manager.stop_all()
+ self.assertTrue(service1.is_alive)
+ self.assertFalse(service2.is_alive)
+ self.assert_recorded_one_error(
+ 'Failed to stop service "mock_service1"')
- def test_start_all_and_stop_all_serveral_times(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService, start_service=False)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- manager.stop_all()
- manager.start_all()
- manager.stop_all()
- manager.start_all()
- manager.stop_all()
- manager.start_all()
- self.assertTrue(service1.is_alive)
- self.assertTrue(service2.is_alive)
- self.assertEqual(service1.start_func.call_count, 4)
- self.assertEqual(service2.start_func.call_count, 3)
- self.assertEqual(service1.stop_func.call_count, 3)
- self.assertEqual(service2.stop_func.call_count, 2)
+ def test_start_all_and_stop_all_serveral_times(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService, start_service=False)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ manager.stop_all()
+ manager.start_all()
+ manager.stop_all()
+ manager.start_all()
+ manager.stop_all()
+ manager.start_all()
+ self.assertTrue(service1.is_alive)
+ self.assertTrue(service2.is_alive)
+ self.assertEqual(service1.start_func.call_count, 4)
+ self.assertEqual(service2.start_func.call_count, 3)
+ self.assertEqual(service1.stop_func.call_count, 3)
+ self.assertEqual(service2.stop_func.call_count, 2)
- def test_pause_all(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- mock_call_tracker = mock.Mock()
- mock_call_tracker.pause1 = service1.pause_func
- mock_call_tracker.pause2 = service2.pause_func
- manager.pause_all()
- self.assertEqual(
- mock_call_tracker.mock_calls,
- [mock.call.pause2(), mock.call.pause1()])
- self.assertEqual(service1.pause_func.call_count, 1)
- self.assertEqual(service2.pause_func.call_count, 1)
- self.assertEqual(service1.resume_func.call_count, 0)
- self.assertEqual(service2.resume_func.call_count, 0)
+ def test_pause_all(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ mock_call_tracker = mock.Mock()
+ mock_call_tracker.pause1 = service1.pause_func
+ mock_call_tracker.pause2 = service2.pause_func
+ manager.pause_all()
+ self.assertEqual(
+ mock_call_tracker.mock_calls,
+ [mock.call.pause2(), mock.call.pause1()])
+ self.assertEqual(service1.pause_func.call_count, 1)
+ self.assertEqual(service2.pause_func.call_count, 1)
+ self.assertEqual(service1.resume_func.call_count, 0)
+ self.assertEqual(service2.resume_func.call_count, 0)
- def test_pause_all_with_some_failed(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service1.pause_func.side_effect = Exception(
- 'Something failed in pause.')
- service2 = manager.mock_service2
- manager.pause_all()
- self.assertEqual(service1.pause_func.call_count, 1)
- self.assertEqual(service2.pause_func.call_count, 1)
- self.assertEqual(service1.resume_func.call_count, 0)
- self.assertEqual(service2.resume_func.call_count, 0)
- self.assert_recorded_one_error(
- 'Failed to pause service "mock_service1".')
+ def test_pause_all_with_some_failed(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service1.pause_func.side_effect = Exception(
+ 'Something failed in pause.')
+ service2 = manager.mock_service2
+ manager.pause_all()
+ self.assertEqual(service1.pause_func.call_count, 1)
+ self.assertEqual(service2.pause_func.call_count, 1)
+ self.assertEqual(service1.resume_func.call_count, 0)
+ self.assertEqual(service2.resume_func.call_count, 0)
+ self.assert_recorded_one_error(
+ 'Failed to pause service "mock_service1".')
- def test_resume_all(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service2 = manager.mock_service2
- mock_call_tracker = mock.Mock()
- mock_call_tracker.resume1 = service1.resume_func
- mock_call_tracker.resume2 = service2.resume_func
- manager.pause_all()
- manager.resume_all()
- self.assertEqual(
- mock_call_tracker.mock_calls,
- [mock.call.resume1(), mock.call.resume2()])
- self.assertEqual(service1.pause_func.call_count, 1)
- self.assertEqual(service2.pause_func.call_count, 1)
- self.assertEqual(service1.resume_func.call_count, 1)
- self.assertEqual(service2.resume_func.call_count, 1)
+ def test_resume_all(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service2 = manager.mock_service2
+ mock_call_tracker = mock.Mock()
+ mock_call_tracker.resume1 = service1.resume_func
+ mock_call_tracker.resume2 = service2.resume_func
+ manager.pause_all()
+ manager.resume_all()
+ self.assertEqual(
+ mock_call_tracker.mock_calls,
+ [mock.call.resume1(), mock.call.resume2()])
+ self.assertEqual(service1.pause_func.call_count, 1)
+ self.assertEqual(service2.pause_func.call_count, 1)
+ self.assertEqual(service1.resume_func.call_count, 1)
+ self.assertEqual(service2.resume_func.call_count, 1)
- def test_resume_all_with_some_failed(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- service1 = manager.mock_service1
- service1.resume_func.side_effect = Exception(
- 'Something failed in resume.')
- service2 = manager.mock_service2
- manager.pause_all()
- manager.resume_all()
- self.assertEqual(service1.pause_func.call_count, 1)
- self.assertEqual(service2.pause_func.call_count, 1)
- self.assertEqual(service1.resume_func.call_count, 1)
- self.assertEqual(service2.resume_func.call_count, 1)
- self.assert_recorded_one_error(
- 'Failed to resume service "mock_service1".')
+ def test_resume_all_with_some_failed(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ service1 = manager.mock_service1
+ service1.resume_func.side_effect = Exception(
+ 'Something failed in resume.')
+ service2 = manager.mock_service2
+ manager.pause_all()
+ manager.resume_all()
+ self.assertEqual(service1.pause_func.call_count, 1)
+ self.assertEqual(service2.pause_func.call_count, 1)
+ self.assertEqual(service1.resume_func.call_count, 1)
+ self.assertEqual(service2.resume_func.call_count, 1)
+ self.assert_recorded_one_error(
+ 'Failed to resume service "mock_service1".')
- def test_list_live_services(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService, start_service=False)
- manager.register('mock_service2', MockService)
- aliases = manager.list_live_services()
- self.assertEqual(aliases, ['mock_service2'])
- manager.stop_all()
- aliases = manager.list_live_services()
- self.assertEqual(aliases, [])
+ def test_list_live_services(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService, start_service=False)
+ manager.register('mock_service2', MockService)
+ aliases = manager.list_live_services()
+ self.assertEqual(aliases, ['mock_service2'])
+ manager.stop_all()
+ aliases = manager.list_live_services()
+ self.assertEqual(aliases, [])
- def test_start_services(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService, start_service=False)
- manager.register('mock_service2', MockService, start_service=False)
- manager.start_services(['mock_service2'])
- aliases = manager.list_live_services()
- self.assertEqual(aliases, ['mock_service2'])
+ def test_start_services(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService, start_service=False)
+ manager.register('mock_service2', MockService, start_service=False)
+ manager.start_services(['mock_service2'])
+ aliases = manager.list_live_services()
+ self.assertEqual(aliases, ['mock_service2'])
- def test_start_services_non_existent(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- msg = ('.* No service is registered under the name "mock_service", '
- 'cannot start.')
- with self.assertRaisesRegex(service_manager.Error, msg):
- manager.start_services(['mock_service'])
+ def test_start_services_non_existent(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ msg = ('.* No service is registered under the name "mock_service", '
+ 'cannot start.')
+ with self.assertRaisesRegex(service_manager.Error, msg):
+ manager.start_services(['mock_service'])
- def test_resume_services(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- manager.register('mock_service1', MockService)
- manager.register('mock_service2', MockService)
- manager.pause_all()
- aliases = manager.list_live_services()
- self.assertEqual(aliases, [])
- manager.resume_services(['mock_service2'])
- aliases = manager.list_live_services()
- self.assertEqual(aliases, ['mock_service2'])
+ def test_resume_services(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ manager.register('mock_service1', MockService)
+ manager.register('mock_service2', MockService)
+ manager.pause_all()
+ aliases = manager.list_live_services()
+ self.assertEqual(aliases, [])
+ manager.resume_services(['mock_service2'])
+ aliases = manager.list_live_services()
+ self.assertEqual(aliases, ['mock_service2'])
- def test_resume_services_non_existent(self):
- manager = service_manager.ServiceManager(mock.MagicMock())
- msg = ('.* No service is registered under the name "mock_service", '
- 'cannot resume.')
- with self.assertRaisesRegex(service_manager.Error, msg):
- manager.resume_services(['mock_service'])
+ def test_resume_services_non_existent(self):
+ manager = service_manager.ServiceManager(mock.MagicMock())
+ msg = ('.* No service is registered under the name "mock_service", '
+ 'cannot resume.')
+ with self.assertRaisesRegex(service_manager.Error, msg):
+ manager.resume_services(['mock_service'])
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/services/base_service_test.py b/tests/mobly/controllers/android_device_lib/services/base_service_test.py
index e97848d..b8474d2 100755
--- a/tests/mobly/controllers/android_device_lib/services/base_service_test.py
+++ b/tests/mobly/controllers/android_device_lib/services/base_service_test.py
@@ -18,14 +18,14 @@
class BaseServiceTest(unittest.TestCase):
- def setUp(self):
- self.mock_device = mock.MagicMock()
- self.service = base_service.BaseService(self.mock_device)
+ def setUp(self):
+ self.mock_device = mock.MagicMock()
+ self.service = base_service.BaseService(self.mock_device)
- def test_alias(self):
- self.service.alias = 'SomeService'
- self.assertEqual(self.service.alias, 'SomeService')
+ def test_alias(self):
+ self.service.alias = 'SomeService'
+ self.assertEqual(self.service.alias, 'SomeService')
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/services/logcat_test.py b/tests/mobly/controllers/android_device_lib/services/logcat_test.py
index 4401c9a..1a59b77 100755
--- a/tests/mobly/controllers/android_device_lib/services/logcat_test.py
+++ b/tests/mobly/controllers/android_device_lib/services/logcat_test.py
@@ -31,24 +31,24 @@
# The expected result of the cat adb operation.
MOCK_ADB_LOGCAT_CAT_RESULT = [
- '02-29 14:02:21.456 4454 Something\n',
- '02-29 14:02:21.789 4454 Something again\n'
+ '02-29 14:02:21.456 4454 Something\n',
+ '02-29 14:02:21.789 4454 Something again\n'
]
# A mocked piece of adb logcat output.
MOCK_ADB_LOGCAT = (u'02-29 14:02:19.123 4454 Nothing\n'
- u'%s'
- u'02-29 14:02:22.123 4454 Something again and again\n'
- ) % u''.join(MOCK_ADB_LOGCAT_CAT_RESULT)
+ u'%s'
+ u'02-29 14:02:22.123 4454 Something again and again\n'
+ ) % u''.join(MOCK_ADB_LOGCAT_CAT_RESULT)
# The expected result of the cat adb operation.
MOCK_ADB_UNICODE_LOGCAT_CAT_RESULT = [
- '02-29 14:02:21.456 4454 Something \u901a\n',
- '02-29 14:02:21.789 4454 Something again\n'
+ '02-29 14:02:21.456 4454 Something \u901a\n',
+ '02-29 14:02:21.789 4454 Something again\n'
]
# A mocked piece of adb logcat output.
MOCK_ADB_UNICODE_LOGCAT = (
- u'02-29 14:02:19.123 4454 Nothing\n'
- u'%s'
- u'02-29 14:02:22.123 4454 Something again and again\n'
+ u'02-29 14:02:19.123 4454 Nothing\n'
+ u'%s'
+ u'02-29 14:02:22.123 4454 Something again and again\n'
) % u''.join(MOCK_ADB_UNICODE_LOGCAT_CAT_RESULT)
# Mock start and end time of the adb cat.
@@ -57,456 +57,456 @@
# Mock AdbError for missing logpersist scripts
MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR = adb.AdbError(
- 'logpersist.stop --clear', b'',
- '/system/bin/sh: logpersist.stop: not found', 0)
+ 'logpersist.stop --clear', b'',
+ '/system/bin/sh: logpersist.stop: not found', 0)
MOCK_LOGPERSIST_START_MISSING_ADB_ERROR = adb.AdbError(
- 'logpersist.start --clear', b'',
- b'/system/bin/sh: logpersist.stop: not found', 0)
+ 'logpersist.start --clear', b'',
+ b'/system/bin/sh: logpersist.stop: not found', 0)
class LogcatTest(unittest.TestCase):
- """Tests for Logcat service and its integration with AndroidDevice."""
+ """Tests for Logcat service and its integration with AndroidDevice."""
- def setUp(self):
- # Set log_path to logging since mobly logger setup is not called.
- if not hasattr(logging, 'log_path'):
- setattr(logging, 'log_path', '/tmp/logs')
- # Creates a temp dir to be used by tests in this test class.
- self.tmp_dir = tempfile.mkdtemp()
+ def setUp(self):
+ # Set log_path to logging since mobly logger setup is not called.
+ if not hasattr(logging, 'log_path'):
+ setattr(logging, 'log_path', '/tmp/logs')
+ # Creates a temp dir to be used by tests in this test class.
+ self.tmp_dir = tempfile.mkdtemp()
- def tearDown(self):
- """Removes the temp dir.
- """
- shutil.rmtree(self.tmp_dir)
+ def tearDown(self):
+ """Removes the temp dir.
+ """
+ shutil.rmtree(self.tmp_dir)
- def AssertFileContains(self, content, file_path):
- with open(file_path, 'r') as f:
- output = f.read()
- self.assertIn(content, output)
+ def AssertFileContains(self, content, file_path):
+ with open(file_path, 'r') as f:
+ output = f.read()
+ self.assertIn(content, output)
- def AssertFileDoesNotContain(self, content, file_path):
- with open(file_path, 'r') as f:
- output = f.read()
- self.assertNotIn(content, output)
+ def AssertFileDoesNotContain(self, content, file_path):
+ with open(file_path, 'r') as f:
+ output = f.read()
+ self.assertNotIn(content, output)
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- @mock.patch('mobly.utils.create_dir')
- @mock.patch('mobly.utils.start_standing_subprocess',
- return_value='process')
- @mock.patch('mobly.utils.stop_standing_subprocess')
- @mock.patch.object(logcat.Logcat, '_open_logcat_file')
- @mock.patch('mobly.logger.get_log_file_timestamp')
- def test_start_and_stop(self, get_timestamp_mock, open_logcat_mock,
- stop_proc_mock, start_proc_mock, create_dir_mock,
- FastbootProxy, MockAdbProxy):
- """Verifies the steps of collecting adb logcat on an AndroidDevice
- object, including various function calls and the expected behaviors of
- the calls.
- """
- mock_serial = '1'
- get_timestamp_mock.return_value = '123'
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service.start()
- # Verify start did the correct operations.
- self.assertTrue(logcat_service._adb_logcat_process)
- expected_log_path = os.path.join(
- logging.log_path, 'AndroidDevice%s' % ad.serial,
- 'logcat,%s,fakemodel,123.txt' % ad.serial)
- create_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
- adb_cmd = ' "adb" -s %s logcat -v threadtime -T 1 >> %s'
- start_proc_mock.assert_called_with(
- adb_cmd % (ad.serial, '"%s" ' % expected_log_path), shell=True)
- self.assertEqual(logcat_service.adb_logcat_file_path,
- expected_log_path)
- expected_msg = (
- 'Logcat thread is already running, cannot start another'
- ' one.')
- # Expect error if start is called back to back.
- with self.assertRaisesRegex(logcat.Error, expected_msg):
- logcat_service.start()
- # Verify stop did the correct operations.
- logcat_service.stop()
- stop_proc_mock.assert_called_with('process')
- self.assertIsNone(logcat_service._adb_logcat_process)
- self.assertEqual(logcat_service.adb_logcat_file_path,
- expected_log_path)
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch('mobly.utils.create_dir')
+ @mock.patch('mobly.utils.start_standing_subprocess',
+ return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch.object(logcat.Logcat, '_open_logcat_file')
+ @mock.patch('mobly.logger.get_log_file_timestamp')
+ def test_start_and_stop(self, get_timestamp_mock, open_logcat_mock,
+ stop_proc_mock, start_proc_mock, create_dir_mock,
+ FastbootProxy, MockAdbProxy):
+ """Verifies the steps of collecting adb logcat on an AndroidDevice
+ object, including various function calls and the expected behaviors of
+ the calls.
+ """
+ mock_serial = '1'
+ get_timestamp_mock.return_value = '123'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service.start()
+ # Verify start did the correct operations.
+ self.assertTrue(logcat_service._adb_logcat_process)
+ expected_log_path = os.path.join(
+ logging.log_path, 'AndroidDevice%s' % ad.serial,
+ 'logcat,%s,fakemodel,123.txt' % ad.serial)
+ create_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
+ adb_cmd = ' "adb" -s %s logcat -v threadtime -T 1 >> %s'
+ start_proc_mock.assert_called_with(
+ adb_cmd % (ad.serial, '"%s" ' % expected_log_path), shell=True)
+ self.assertEqual(logcat_service.adb_logcat_file_path,
+ expected_log_path)
+ expected_msg = (
+ 'Logcat thread is already running, cannot start another'
+ ' one.')
+ # Expect error if start is called back to back.
+ with self.assertRaisesRegex(logcat.Error, expected_msg):
+ logcat_service.start()
+ # Verify stop did the correct operations.
+ logcat_service.stop()
+ stop_proc_mock.assert_called_with('process')
+ self.assertIsNone(logcat_service._adb_logcat_process)
+ self.assertEqual(logcat_service.adb_logcat_file_path,
+ expected_log_path)
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- @mock.patch('mobly.utils.create_dir')
- @mock.patch('mobly.utils.start_standing_subprocess',
- return_value='process')
- @mock.patch('mobly.utils.stop_standing_subprocess')
- @mock.patch.object(logcat.Logcat, '_open_logcat_file')
- def test_update_config(self, open_logcat_mock, stop_proc_mock,
- start_proc_mock, create_dir_mock, FastbootProxy,
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch('mobly.utils.create_dir')
+ @mock.patch('mobly.utils.start_standing_subprocess',
+ return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch.object(logcat.Logcat, '_open_logcat_file')
+ def test_update_config(self, open_logcat_mock, stop_proc_mock,
+ start_proc_mock, create_dir_mock, FastbootProxy,
+ MockAdbProxy):
+ mock_serial = '1'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service.start()
+ logcat_service.stop()
+ new_log_params = '-a -b -c'
+ new_file_path = 'some/path/log.txt'
+ new_config = logcat.Config(logcat_params=new_log_params,
+ output_file_path=new_file_path)
+ logcat_service.update_config(new_config)
+ logcat_service.start()
+ self.assertTrue(logcat_service._adb_logcat_process)
+ create_dir_mock.assert_has_calls([mock.call('some/path')])
+ expected_adb_cmd = (' "adb" -s 1 logcat -v threadtime -T 1 -a -b -c >> '
+ '"some/path/log.txt" ')
+ start_proc_mock.assert_called_with(expected_adb_cmd, shell=True)
+ self.assertEqual(logcat_service.adb_logcat_file_path,
+ 'some/path/log.txt')
+ logcat_service.stop()
+
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch('mobly.utils.create_dir')
+ @mock.patch('mobly.utils.start_standing_subprocess',
+ return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch.object(logcat.Logcat, '_open_logcat_file')
+ def test_update_config_while_running(self, open_logcat_mock, stop_proc_mock,
+ start_proc_mock, create_dir_mock,
+ FastbootProxy, MockAdbProxy):
+ mock_serial = '1'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service.start()
+ new_config = logcat.Config(logcat_params='-blah',
+ output_file_path='some/path/file.txt')
+ with self.assertRaisesRegex(
+ logcat.Error,
+ 'Logcat thread is already running, cannot start another one'):
+ logcat_service.update_config(new_config)
+ self.assertTrue(logcat_service.is_alive)
+ logcat_service.stop()
+
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch('mobly.utils.create_dir')
+ @mock.patch('mobly.utils.start_standing_subprocess',
+ return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch.object(logcat.Logcat, '_open_logcat_file')
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.services.logcat.Logcat.clear_adb_log',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ def test_pause_and_resume(self, clear_adb_mock, open_logcat_mock,
+ stop_proc_mock, start_proc_mock, create_dir_mock,
+ FastbootProxy, MockAdbProxy):
+ mock_serial = '1'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad, logcat.Config(clear_log=True))
+ logcat_service.start()
+ clear_adb_mock.assert_called_once_with()
+ self.assertTrue(logcat_service.is_alive)
+ logcat_service.pause()
+ self.assertFalse(logcat_service.is_alive)
+ stop_proc_mock.assert_called_with('process')
+ self.assertIsNone(logcat_service._adb_logcat_process)
+ clear_adb_mock.reset_mock()
+ logcat_service.resume()
+ self.assertTrue(logcat_service.is_alive)
+ clear_adb_mock.assert_not_called()
+ logcat_service.stop()
+
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch('mobly.utils.start_standing_subprocess',
+ return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.services.logcat.Logcat.clear_adb_log',
+ return_value=mock_android_device.MockAdbProxy('1'))
+ def test_logcat_service_create_output_excerpts(self, clear_adb_mock,
+ stop_proc_mock,
+ start_proc_mock,
+ FastbootProxy,
MockAdbProxy):
- mock_serial = '1'
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service.start()
- logcat_service.stop()
- new_log_params = '-a -b -c'
- new_file_path = 'some/path/log.txt'
- new_config = logcat.Config(logcat_params=new_log_params,
- output_file_path=new_file_path)
- logcat_service.update_config(new_config)
- logcat_service.start()
- self.assertTrue(logcat_service._adb_logcat_process)
- create_dir_mock.assert_has_calls([mock.call('some/path')])
- expected_adb_cmd = (' "adb" -s 1 logcat -v threadtime -T 1 -a -b -c >> '
- '"some/path/log.txt" ')
- start_proc_mock.assert_called_with(expected_adb_cmd, shell=True)
- self.assertEqual(logcat_service.adb_logcat_file_path,
- 'some/path/log.txt')
- logcat_service.stop()
+ mock_serial = '1'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service._start()
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- @mock.patch('mobly.utils.create_dir')
- @mock.patch('mobly.utils.start_standing_subprocess',
- return_value='process')
- @mock.patch('mobly.utils.stop_standing_subprocess')
- @mock.patch.object(logcat.Logcat, '_open_logcat_file')
- def test_update_config_while_running(self, open_logcat_mock, stop_proc_mock,
- start_proc_mock, create_dir_mock,
- FastbootProxy, MockAdbProxy):
- mock_serial = '1'
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service.start()
- new_config = logcat.Config(logcat_params='-blah',
- output_file_path='some/path/file.txt')
- with self.assertRaisesRegex(
- logcat.Error,
- 'Logcat thread is already running, cannot start another one'):
- logcat_service.update_config(new_config)
- self.assertTrue(logcat_service.is_alive)
- logcat_service.stop()
+ def _write_logcat_file_and_assert_excerpts_exists(logcat_file_content,
+ test_begin_time,
+ test_name):
+ with open(logcat_service.adb_logcat_file_path, 'a') as f:
+ f.write(logcat_file_content)
+ test_output_dir = os.path.join(self.tmp_dir, test_name)
+ mock_record = mock.MagicMock()
+ mock_record.begin_time = test_begin_time
+ test_run_info = runtime_test_info.RuntimeTestInfo(test_name,
+ test_output_dir,
+ mock_record)
+ actual_path = logcat_service.create_output_excerpts(test_run_info)[0]
+ expected_path = os.path.join(
+ test_output_dir, '{test_name}-{test_begin_time}'.format(
+ test_name=test_name, test_begin_time=test_begin_time),
+ 'logcat,{mock_serial},fakemodel,{test_name}-{test_begin_time}.txt'
+ .format(
+ mock_serial=mock_serial,
+ test_name=test_name,
+ test_begin_time=test_begin_time))
+ self.assertEqual(actual_path, expected_path)
+ self.assertTrue(os.path.exists(expected_path))
+ return expected_path
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- @mock.patch('mobly.utils.create_dir')
- @mock.patch('mobly.utils.start_standing_subprocess',
- return_value='process')
- @mock.patch('mobly.utils.stop_standing_subprocess')
- @mock.patch.object(logcat.Logcat, '_open_logcat_file')
- @mock.patch(
- 'mobly.controllers.android_device_lib.services.logcat.Logcat.clear_adb_log',
+ # Generate logs before the file pointer is created.
+ # This message will not be captured in the excerpt.
+ NOT_IN_EXCERPT = 'Not in excerpt.\n'
+ with open(logcat_service.adb_logcat_file_path, 'a') as f:
+ f.write(NOT_IN_EXCERPT)
+ # With the file pointer created, generate logs and make an excerpt.
+ logcat_service._open_logcat_file()
+ FILE_CONTENT = 'Some log.\n'
+ expected_path1 = _write_logcat_file_and_assert_excerpts_exists(
+ logcat_file_content=FILE_CONTENT,
+ test_begin_time=123,
+ test_name='test_foo',
+ )
+ self.AssertFileContains(FILE_CONTENT, expected_path1)
+ self.AssertFileDoesNotContain(NOT_IN_EXCERPT, expected_path1)
+ # Generate some new logs and do another excerpt.
+ FILE_CONTENT = 'Some more logs!!!\n'
+ expected_path2 = _write_logcat_file_and_assert_excerpts_exists(
+ logcat_file_content=FILE_CONTENT,
+ test_begin_time=456,
+ test_name='test_bar',
+ )
+ self.AssertFileContains(FILE_CONTENT, expected_path2)
+ self.AssertFileDoesNotContain(FILE_CONTENT, expected_path1)
+ # Simulate devices accidentally go offline, logcat service stopped.
+ logcat_service.stop()
+ FILE_CONTENT = 'Whatever logs\n'
+ expected_path3 = _write_logcat_file_and_assert_excerpts_exists(
+ logcat_file_content=FILE_CONTENT,
+ test_begin_time=789,
+ test_name='test_offline',
+ )
+ self.assertEqual(os.stat(expected_path3).st_size, 0)
+
+
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
return_value=mock_android_device.MockAdbProxy('1'))
- def test_pause_and_resume(self, clear_adb_mock, open_logcat_mock,
- stop_proc_mock, start_proc_mock, create_dir_mock,
- FastbootProxy, MockAdbProxy):
- mock_serial = '1'
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad, logcat.Config(clear_log=True))
- logcat_service.start()
- clear_adb_mock.assert_called_once_with()
- self.assertTrue(logcat_service.is_alive)
- logcat_service.pause()
- self.assertFalse(logcat_service.is_alive)
- stop_proc_mock.assert_called_with('process')
- self.assertIsNone(logcat_service._adb_logcat_process)
- clear_adb_mock.reset_mock()
- logcat_service.resume()
- self.assertTrue(logcat_service.is_alive)
- clear_adb_mock.assert_not_called()
- logcat_service.stop()
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ @mock.patch('mobly.utils.create_dir')
+ @mock.patch('mobly.utils.start_standing_subprocess',
+ return_value='process')
+ @mock.patch('mobly.utils.stop_standing_subprocess')
+ @mock.patch.object(logcat.Logcat, '_open_logcat_file')
+ @mock.patch('mobly.logger.get_log_file_timestamp')
+ def test_take_logcat_with_extra_params(self, get_timestamp_mock,
+ open_logcat_mock, stop_proc_mock,
+ start_proc_mock, create_dir_mock,
+ FastbootProxy, MockAdbProxy):
+ """Verifies the steps of collecting adb logcat on an AndroidDevice
+ object, including various function calls and the expected behaviors of
+ the calls.
+ """
+ mock_serial = '1'
+ get_timestamp_mock.return_value = '123'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ configs = logcat.Config()
+ configs.logcat_params = '-b radio'
+ logcat_service = logcat.Logcat(ad, configs)
+ logcat_service.start()
+ # Verify start did the correct operations.
+ self.assertTrue(logcat_service._adb_logcat_process)
+ expected_log_path = os.path.join(
+ logging.log_path, 'AndroidDevice%s' % ad.serial,
+ 'logcat,%s,fakemodel,123.txt' % ad.serial)
+ create_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
+ adb_cmd = ' "adb" -s %s logcat -v threadtime -T 1 -b radio >> %s'
+ start_proc_mock.assert_called_with(
+ adb_cmd % (ad.serial, '"%s" ' % expected_log_path), shell=True)
+ self.assertEqual(logcat_service.adb_logcat_file_path,
+ expected_log_path)
+ logcat_service.stop()
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- @mock.patch('mobly.utils.start_standing_subprocess',
- return_value='process')
- @mock.patch('mobly.utils.stop_standing_subprocess')
- @mock.patch(
- 'mobly.controllers.android_device_lib.services.logcat.Logcat.clear_adb_log',
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
return_value=mock_android_device.MockAdbProxy('1'))
- def test_logcat_service_create_output_excerpts(self, clear_adb_mock,
- stop_proc_mock,
- start_proc_mock,
- FastbootProxy,
- MockAdbProxy):
- mock_serial = '1'
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service._start()
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test_instantiation(self, MockFastboot, MockAdbProxy):
+ """Verifies the AndroidDevice object's basic attributes are correctly
+ set after instantiation.
+ """
+ mock_serial = 1
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ self.assertIsNone(logcat_service._adb_logcat_process)
+ self.assertIsNone(logcat_service.adb_logcat_file_path)
- def _write_logcat_file_and_assert_excerpts_exists(logcat_file_content,
- test_begin_time,
- test_name):
- with open(logcat_service.adb_logcat_file_path, 'a') as f:
- f.write(logcat_file_content)
- test_output_dir = os.path.join(self.tmp_dir, test_name)
- mock_record = mock.MagicMock()
- mock_record.begin_time = test_begin_time
- test_run_info = runtime_test_info.RuntimeTestInfo(test_name,
- test_output_dir,
- mock_record)
- actual_path = logcat_service.create_output_excerpts(test_run_info)[0]
- expected_path = os.path.join(
- test_output_dir, '{test_name}-{test_begin_time}'.format(
- test_name=test_name, test_begin_time=test_begin_time),
- 'logcat,{mock_serial},fakemodel,{test_name}-{test_begin_time}.txt'
- .format(
- mock_serial=mock_serial,
- test_name=test_name,
- test_begin_time=test_begin_time))
- self.assertEqual(actual_path, expected_path)
- self.assertTrue(os.path.exists(expected_path))
- return expected_path
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock.MagicMock())
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test__enable_logpersist_with_logpersist(self, MockFastboot,
+ MockAdbProxy):
+ mock_serial = '1'
+ mock_adb_proxy = MockAdbProxy.return_value
+ mock_adb_proxy.getprops.return_value = {
+ 'ro.build.id': 'AB42',
+ 'ro.build.type': 'userdebug',
+ 'ro.debuggable': '1',
+ }
+ mock_adb_proxy.has_shell_command.side_effect = lambda command: {
+ 'logpersist.start': True,
+ 'logpersist.stop': True,
+ }[command]
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service._enable_logpersist()
+ mock_adb_proxy.shell.assert_has_calls([
+ mock.call('logpersist.stop --clear'),
+ mock.call('logpersist.start'),
+ ])
- # Generate logs before the file pointer is created.
- # This message will not be captured in the excerpt.
- NOT_IN_EXCERPT = 'Not in excerpt.\n'
- with open(logcat_service.adb_logcat_file_path, 'a') as f:
- f.write(NOT_IN_EXCERPT)
- # With the file pointer created, generate logs and make an excerpt.
- logcat_service._open_logcat_file()
- FILE_CONTENT = 'Some log.\n'
- expected_path1 = _write_logcat_file_and_assert_excerpts_exists(
- logcat_file_content=FILE_CONTENT,
- test_begin_time=123,
- test_name='test_foo',
- )
- self.AssertFileContains(FILE_CONTENT, expected_path1)
- self.AssertFileDoesNotContain(NOT_IN_EXCERPT, expected_path1)
- # Generate some new logs and do another excerpt.
- FILE_CONTENT = 'Some more logs!!!\n'
- expected_path2 = _write_logcat_file_and_assert_excerpts_exists(
- logcat_file_content=FILE_CONTENT,
- test_begin_time=456,
- test_name='test_bar',
- )
- self.AssertFileContains(FILE_CONTENT, expected_path2)
- self.AssertFileDoesNotContain(FILE_CONTENT, expected_path1)
- # Simulate devices accidentally go offline, logcat service stopped.
- logcat_service.stop()
- FILE_CONTENT = 'Whatever logs\n'
- expected_path3 = _write_logcat_file_and_assert_excerpts_exists(
- logcat_file_content=FILE_CONTENT,
- test_begin_time=789,
- test_name='test_offline',
- )
- self.assertEqual(os.stat(expected_path3).st_size, 0)
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock.MagicMock())
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test__enable_logpersist_with_user_build_device(self, MockFastboot,
+ MockAdbProxy):
+ mock_serial = '1'
+ mock_adb_proxy = MockAdbProxy.return_value
+ mock_adb_proxy.getprops.return_value = {
+ 'ro.build.id': 'AB42',
+ 'ro.build.type': 'user',
+ 'ro.debuggable': '0',
+ }
+ mock_adb_proxy.has_shell_command.side_effect = lambda command: {
+ 'logpersist.start': True,
+ 'logpersist.stop': True,
+ }[command]
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service._enable_logpersist()
+ mock_adb_proxy.shell.assert_not_called()
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock.MagicMock())
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test__enable_logpersist_with_missing_all_logpersist(
+ self, MockFastboot, MockAdbProxy):
+ def adb_shell_helper(command):
+ if command == 'logpersist.start':
+ raise MOCK_LOGPERSIST_START_MISSING_ADB_ERROR
+ elif command == 'logpersist.stop --clear':
+ raise MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR
+ else:
+ return b''
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- @mock.patch('mobly.utils.create_dir')
- @mock.patch('mobly.utils.start_standing_subprocess',
- return_value='process')
- @mock.patch('mobly.utils.stop_standing_subprocess')
- @mock.patch.object(logcat.Logcat, '_open_logcat_file')
- @mock.patch('mobly.logger.get_log_file_timestamp')
- def test_take_logcat_with_extra_params(self, get_timestamp_mock,
- open_logcat_mock, stop_proc_mock,
- start_proc_mock, create_dir_mock,
- FastbootProxy, MockAdbProxy):
- """Verifies the steps of collecting adb logcat on an AndroidDevice
- object, including various function calls and the expected behaviors of
- the calls.
- """
- mock_serial = '1'
- get_timestamp_mock.return_value = '123'
- ad = android_device.AndroidDevice(serial=mock_serial)
- configs = logcat.Config()
- configs.logcat_params = '-b radio'
- logcat_service = logcat.Logcat(ad, configs)
- logcat_service.start()
- # Verify start did the correct operations.
- self.assertTrue(logcat_service._adb_logcat_process)
- expected_log_path = os.path.join(
- logging.log_path, 'AndroidDevice%s' % ad.serial,
- 'logcat,%s,fakemodel,123.txt' % ad.serial)
- create_dir_mock.assert_called_with(os.path.dirname(expected_log_path))
- adb_cmd = ' "adb" -s %s logcat -v threadtime -T 1 -b radio >> %s'
- start_proc_mock.assert_called_with(
- adb_cmd % (ad.serial, '"%s" ' % expected_log_path), shell=True)
- self.assertEqual(logcat_service.adb_logcat_file_path,
- expected_log_path)
- logcat_service.stop()
+ mock_serial = '1'
+ mock_adb_proxy = MockAdbProxy.return_value
+ mock_adb_proxy.getprops.return_value = {
+ 'ro.build.id': 'AB42',
+ 'ro.build.type': 'userdebug',
+ 'ro.debuggable': '1',
+ }
+ mock_adb_proxy.has_shell_command.side_effect = lambda command: {
+ 'logpersist.start': False,
+ 'logpersist.stop': False,
+ }[command]
+ mock_adb_proxy.shell.side_effect = adb_shell_helper
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service._enable_logpersist()
+ mock_adb_proxy.shell.assert_not_called()
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock_android_device.MockAdbProxy('1'))
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test_instantiation(self, MockFastboot, MockAdbProxy):
- """Verifies the AndroidDevice object's basic attributes are correctly
- set after instantiation.
- """
- mock_serial = 1
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- self.assertIsNone(logcat_service._adb_logcat_process)
- self.assertIsNone(logcat_service.adb_logcat_file_path)
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock.MagicMock())
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test__enable_logpersist_with_missing_logpersist_stop(
+ self, MockFastboot, MockAdbProxy):
+ def adb_shell_helper(command):
+ if command == 'logpersist.stop --clear':
+ raise MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR
+ else:
+ return b''
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock.MagicMock())
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test__enable_logpersist_with_logpersist(self, MockFastboot,
- MockAdbProxy):
- mock_serial = '1'
- mock_adb_proxy = MockAdbProxy.return_value
- mock_adb_proxy.getprops.return_value = {
- 'ro.build.id': 'AB42',
- 'ro.build.type': 'userdebug',
- 'ro.debuggable': '1',
- }
- mock_adb_proxy.has_shell_command.side_effect = lambda command: {
- 'logpersist.start': True,
- 'logpersist.stop': True,
- }[command]
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service._enable_logpersist()
- mock_adb_proxy.shell.assert_has_calls([
- mock.call('logpersist.stop --clear'),
- mock.call('logpersist.start'),
- ])
+ mock_serial = '1'
+ mock_adb_proxy = MockAdbProxy.return_value
+ mock_adb_proxy.getprops.return_value = {
+ 'ro.build.id': 'AB42',
+ 'ro.build.type': 'userdebug',
+ 'ro.debuggable': '1',
+ }
+ mock_adb_proxy.has_shell_command.side_effect = lambda command: {
+ 'logpersist.start': True,
+ 'logpersist.stop': False,
+ }[command]
+ mock_adb_proxy.shell.side_effect = adb_shell_helper
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service._enable_logpersist()
+ mock_adb_proxy.shell.assert_has_calls([
+ mock.call('logpersist.stop --clear'),
+ ])
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock.MagicMock())
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test__enable_logpersist_with_user_build_device(self, MockFastboot,
- MockAdbProxy):
- mock_serial = '1'
- mock_adb_proxy = MockAdbProxy.return_value
- mock_adb_proxy.getprops.return_value = {
- 'ro.build.id': 'AB42',
- 'ro.build.type': 'user',
- 'ro.debuggable': '0',
- }
- mock_adb_proxy.has_shell_command.side_effect = lambda command: {
- 'logpersist.start': True,
- 'logpersist.stop': True,
- }[command]
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service._enable_logpersist()
- mock_adb_proxy.shell.assert_not_called()
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
+ return_value=mock.MagicMock())
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test__enable_logpersist_with_missing_logpersist_start(
+ self, MockFastboot, MockAdbProxy):
+ def adb_shell_helper(command):
+ if command == 'logpersist.start':
+ raise MOCK_LOGPERSIST_START_MISSING_ADB_ERROR
+ else:
+ return b''
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock.MagicMock())
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test__enable_logpersist_with_missing_all_logpersist(
- self, MockFastboot, MockAdbProxy):
- def adb_shell_helper(command):
- if command == 'logpersist.start':
- raise MOCK_LOGPERSIST_START_MISSING_ADB_ERROR
- elif command == 'logpersist.stop --clear':
- raise MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR
- else:
- return b''
+ mock_serial = '1'
+ mock_adb_proxy = MockAdbProxy.return_value
+ mock_adb_proxy.getprops.return_value = {
+ 'ro.build.id': 'AB42',
+ 'ro.build.type': 'userdebug',
+ 'ro.debuggable': '1',
+ }
+ mock_adb_proxy.has_shell_command.side_effect = lambda command: {
+ 'logpersist.start': False,
+ 'logpersist.stop': True,
+ }[command]
+ mock_adb_proxy.shell.side_effect = adb_shell_helper
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service._enable_logpersist()
+ mock_adb_proxy.shell.assert_not_called()
- mock_serial = '1'
- mock_adb_proxy = MockAdbProxy.return_value
- mock_adb_proxy.getprops.return_value = {
- 'ro.build.id': 'AB42',
- 'ro.build.type': 'userdebug',
- 'ro.debuggable': '1',
- }
- mock_adb_proxy.has_shell_command.side_effect = lambda command: {
- 'logpersist.start': False,
- 'logpersist.stop': False,
- }[command]
- mock_adb_proxy.shell.side_effect = adb_shell_helper
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service._enable_logpersist()
- mock_adb_proxy.shell.assert_not_called()
-
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock.MagicMock())
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test__enable_logpersist_with_missing_logpersist_stop(
- self, MockFastboot, MockAdbProxy):
- def adb_shell_helper(command):
- if command == 'logpersist.stop --clear':
- raise MOCK_LOGPERSIST_STOP_MISSING_ADB_ERROR
- else:
- return b''
-
- mock_serial = '1'
- mock_adb_proxy = MockAdbProxy.return_value
- mock_adb_proxy.getprops.return_value = {
- 'ro.build.id': 'AB42',
- 'ro.build.type': 'userdebug',
- 'ro.debuggable': '1',
- }
- mock_adb_proxy.has_shell_command.side_effect = lambda command: {
- 'logpersist.start': True,
- 'logpersist.stop': False,
- }[command]
- mock_adb_proxy.shell.side_effect = adb_shell_helper
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service._enable_logpersist()
- mock_adb_proxy.shell.assert_has_calls([
- mock.call('logpersist.stop --clear'),
- ])
-
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy',
- return_value=mock.MagicMock())
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test__enable_logpersist_with_missing_logpersist_start(
- self, MockFastboot, MockAdbProxy):
- def adb_shell_helper(command):
- if command == 'logpersist.start':
- raise MOCK_LOGPERSIST_START_MISSING_ADB_ERROR
- else:
- return b''
-
- mock_serial = '1'
- mock_adb_proxy = MockAdbProxy.return_value
- mock_adb_proxy.getprops.return_value = {
- 'ro.build.id': 'AB42',
- 'ro.build.type': 'userdebug',
- 'ro.debuggable': '1',
- }
- mock_adb_proxy.has_shell_command.side_effect = lambda command: {
- 'logpersist.start': False,
- 'logpersist.stop': True,
- }[command]
- mock_adb_proxy.shell.side_effect = adb_shell_helper
- ad = android_device.AndroidDevice(serial=mock_serial)
- logcat_service = logcat.Logcat(ad)
- logcat_service._enable_logpersist()
- mock_adb_proxy.shell.assert_not_called()
-
- @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy')
- @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
- return_value=mock_android_device.MockFastbootProxy('1'))
- def test_clear_adb_log(self, MockFastboot, MockAdbProxy):
- mock_serial = '1'
- ad = android_device.AndroidDevice(serial=mock_serial)
- ad.adb.logcat = mock.MagicMock()
- ad.adb.logcat.side_effect = adb.AdbError(
- cmd='cmd',
- stdout=b'',
- stderr=b'failed to clear "main" log',
- ret_code=1)
- logcat_service = logcat.Logcat(ad)
- logcat_service.clear_adb_log()
+ @mock.patch('mobly.controllers.android_device_lib.adb.AdbProxy')
+ @mock.patch('mobly.controllers.android_device_lib.fastboot.FastbootProxy',
+ return_value=mock_android_device.MockFastbootProxy('1'))
+ def test_clear_adb_log(self, MockFastboot, MockAdbProxy):
+ mock_serial = '1'
+ ad = android_device.AndroidDevice(serial=mock_serial)
+ ad.adb.logcat = mock.MagicMock()
+ ad.adb.logcat.side_effect = adb.AdbError(
+ cmd='cmd',
+ stdout=b'',
+ stderr=b'failed to clear "main" log',
+ ret_code=1)
+ logcat_service = logcat.Logcat(ad)
+ logcat_service.clear_adb_log()
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/services/sl4a_service_test.py b/tests/mobly/controllers/android_device_lib/services/sl4a_service_test.py
index 9205918..42ee25c 100755
--- a/tests/mobly/controllers/android_device_lib/services/sl4a_service_test.py
+++ b/tests/mobly/controllers/android_device_lib/services/sl4a_service_test.py
@@ -20,49 +20,49 @@
@mock.patch('mobly.controllers.android_device_lib.sl4a_client.Sl4aClient')
class Sl4aServiceTest(unittest.TestCase):
- """Tests for the sl4a service."""
+ """Tests for the sl4a service."""
- def test_instantiation(self, _):
- service = sl4a_service.Sl4aService(mock.MagicMock())
- self.assertFalse(service.is_alive)
+ def test_instantiation(self, _):
+ service = sl4a_service.Sl4aService(mock.MagicMock())
+ self.assertFalse(service.is_alive)
- def test_start(self, mock_sl4a_client_class):
- mock_client = mock_sl4a_client_class.return_value
- service = sl4a_service.Sl4aService(mock.MagicMock())
- service.start()
- mock_client.start_app_and_connect.assert_called_once_with()
- self.assertTrue(service.is_alive)
+ def test_start(self, mock_sl4a_client_class):
+ mock_client = mock_sl4a_client_class.return_value
+ service = sl4a_service.Sl4aService(mock.MagicMock())
+ service.start()
+ mock_client.start_app_and_connect.assert_called_once_with()
+ self.assertTrue(service.is_alive)
- def test_stop(self, mock_sl4a_client_class):
- mock_client = mock_sl4a_client_class.return_value
- service = sl4a_service.Sl4aService(mock.MagicMock())
- service.start()
- service.stop()
- mock_client.stop_app.assert_called_once_with()
- self.assertFalse(service.is_alive)
+ def test_stop(self, mock_sl4a_client_class):
+ mock_client = mock_sl4a_client_class.return_value
+ service = sl4a_service.Sl4aService(mock.MagicMock())
+ service.start()
+ service.stop()
+ mock_client.stop_app.assert_called_once_with()
+ self.assertFalse(service.is_alive)
- def test_pause(self, mock_sl4a_client_class):
- mock_client = mock_sl4a_client_class.return_value
- service = sl4a_service.Sl4aService(mock.MagicMock())
- service.start()
- service.pause()
- mock_client.stop_event_dispatcher.assert_called_once_with()
- mock_client.clear_host_port.assert_called_once_with()
+ def test_pause(self, mock_sl4a_client_class):
+ mock_client = mock_sl4a_client_class.return_value
+ service = sl4a_service.Sl4aService(mock.MagicMock())
+ service.start()
+ service.pause()
+ mock_client.stop_event_dispatcher.assert_called_once_with()
+ mock_client.clear_host_port.assert_called_once_with()
- def test_resume(self, mock_sl4a_client_class):
- mock_client = mock_sl4a_client_class.return_value
- service = sl4a_service.Sl4aService(mock.MagicMock())
- service.start()
- service.pause()
- service.resume()
- mock_client.restore_app_connection.assert_called_once_with()
+ def test_resume(self, mock_sl4a_client_class):
+ mock_client = mock_sl4a_client_class.return_value
+ service = sl4a_service.Sl4aService(mock.MagicMock())
+ service.start()
+ service.pause()
+ service.resume()
+ mock_client.restore_app_connection.assert_called_once_with()
- def test_register_with_service_manager(self, _):
- mock_device = mock.MagicMock()
- manager = service_manager.ServiceManager(mock_device)
- manager.register('sl4a', sl4a_service.Sl4aService)
- self.assertTrue(manager.sl4a)
+ def test_register_with_service_manager(self, _):
+ mock_device = mock.MagicMock()
+ manager = service_manager.ServiceManager(mock_device)
+ manager.register('sl4a', sl4a_service.Sl4aService)
+ self.assertTrue(manager.sl4a)
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/services/snippet_management_service_test.py b/tests/mobly/controllers/android_device_lib/services/snippet_management_service_test.py
index e542382..974972a 100755
--- a/tests/mobly/controllers/android_device_lib/services/snippet_management_service_test.py
+++ b/tests/mobly/controllers/android_device_lib/services/snippet_management_service_test.py
@@ -21,158 +21,158 @@
class SnippetManagementServiceTest(unittest.TestCase):
- """Tests for the snippet management service."""
+ """Tests for the snippet management service."""
- def test_empty_manager_start_stop(self):
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.start()
- # When no client is registered, manager is never alive.
- self.assertFalse(manager.is_alive)
- manager.stop()
- self.assertFalse(manager.is_alive)
+ def test_empty_manager_start_stop(self):
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.start()
+ # When no client is registered, manager is never alive.
+ self.assertFalse(manager.is_alive)
+ manager.stop()
+ self.assertFalse(manager.is_alive)
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_get_snippet_client(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- self.assertEqual(manager.get_snippet_client('foo'), mock_client)
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_get_snippet_client(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ self.assertEqual(manager.get_snippet_client('foo'), mock_client)
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_get_snippet_client_fail(self, _):
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- self.assertIsNone(manager.get_snippet_client('foo'))
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_get_snippet_client_fail(self, _):
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ self.assertIsNone(manager.get_snippet_client('foo'))
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_stop_with_live_client(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- mock_client.start_app_and_connect.assert_called_once_with()
- manager.stop()
- mock_client.stop_app.assert_called_once_with()
- mock_client.stop_app.reset_mock()
- mock_client.is_alive = False
- self.assertFalse(manager.is_alive)
- manager.stop()
- mock_client.stop_app.assert_not_called()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_stop_with_live_client(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ mock_client.start_app_and_connect.assert_called_once_with()
+ manager.stop()
+ mock_client.stop_app.assert_called_once_with()
+ mock_client.stop_app.reset_mock()
+ mock_client.is_alive = False
+ self.assertFalse(manager.is_alive)
+ manager.stop()
+ mock_client.stop_app.assert_not_called()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_add_snippet_client_dup_name(self, _):
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- msg = ('.* Name "foo" is already registered with package ".*", it '
- 'cannot be used again.')
- with self.assertRaisesRegex(snippet_management_service.Error, msg):
- manager.add_snippet_client('foo', MOCK_PACKAGE + 'ha')
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_add_snippet_client_dup_name(self, _):
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ msg = ('.* Name "foo" is already registered with package ".*", it '
+ 'cannot be used again.')
+ with self.assertRaisesRegex(snippet_management_service.Error, msg):
+ manager.add_snippet_client('foo', MOCK_PACKAGE + 'ha')
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_add_snippet_client_dup_package(self, mock_class):
- mock_client = mock_class.return_value
- mock_client.package = MOCK_PACKAGE
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- msg = ('Snippet package "com.mock.package" has already been loaded '
- 'under name "foo".')
- with self.assertRaisesRegex(snippet_management_service.Error, msg):
- manager.add_snippet_client('bar', MOCK_PACKAGE)
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_add_snippet_client_dup_package(self, mock_class):
+ mock_client = mock_class.return_value
+ mock_client.package = MOCK_PACKAGE
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ msg = ('Snippet package "com.mock.package" has already been loaded '
+ 'under name "foo".')
+ with self.assertRaisesRegex(snippet_management_service.Error, msg):
+ manager.add_snippet_client('bar', MOCK_PACKAGE)
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_remove_snippet_client(self, mock_class):
- mock_client = mock.MagicMock()
- mock_class.return_value = mock_client
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- manager.remove_snippet_client('foo')
- msg = 'No snippet client is registered with name "foo".'
- with self.assertRaisesRegex(snippet_management_service.Error, msg):
- manager.foo.do_something()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_remove_snippet_client(self, mock_class):
+ mock_client = mock.MagicMock()
+ mock_class.return_value = mock_client
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ manager.remove_snippet_client('foo')
+ msg = 'No snippet client is registered with name "foo".'
+ with self.assertRaisesRegex(snippet_management_service.Error, msg):
+ manager.foo.do_something()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_remove_snippet_client(self, mock_class):
- mock_client = mock.MagicMock()
- mock_class.return_value = mock_client
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- with self.assertRaisesRegex(
- snippet_management_service.Error,
- 'No snippet client is registered with name "foo".'):
- manager.remove_snippet_client('foo')
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_remove_snippet_client(self, mock_class):
+ mock_client = mock.MagicMock()
+ mock_class.return_value = mock_client
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ with self.assertRaisesRegex(
+ snippet_management_service.Error,
+ 'No snippet client is registered with name "foo".'):
+ manager.remove_snippet_client('foo')
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_start_with_live_service(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- mock_client.start_app_and_connect.reset_mock()
- mock_client.is_alive = True
- manager.start()
- mock_client.start_app_and_connect.assert_not_called()
- self.assertTrue(manager.is_alive)
- mock_client.is_alive = False
- manager.start()
- mock_client.start_app_and_connect.assert_called_once_with()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_start_with_live_service(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ mock_client.start_app_and_connect.reset_mock()
+ mock_client.is_alive = True
+ manager.start()
+ mock_client.start_app_and_connect.assert_not_called()
+ self.assertTrue(manager.is_alive)
+ mock_client.is_alive = False
+ manager.start()
+ mock_client.start_app_and_connect.assert_called_once_with()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_pause(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- manager.pause()
- mock_client.clear_host_port.assert_called_once_with()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_pause(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ manager.pause()
+ mock_client.clear_host_port.assert_called_once_with()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_resume_positive_case(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- mock_client.is_alive = True
- mock_client.host_port = None
- manager.resume()
- mock_client.restore_app_connection.assert_called_once_with()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_resume_positive_case(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ mock_client.is_alive = True
+ mock_client.host_port = None
+ manager.resume()
+ mock_client.restore_app_connection.assert_called_once_with()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_resume_alive_with_host_port(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- mock_client.is_alive = True
- mock_client.host_port = 1
- manager.resume()
- mock_client.restore_app_connection.assert_not_called()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_resume_alive_with_host_port(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ mock_client.is_alive = True
+ mock_client.host_port = 1
+ manager.resume()
+ mock_client.restore_app_connection.assert_not_called()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_resume_not_alive_no_host_port(self, mock_class):
- mock_client = mock_class.return_value
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- mock_client.is_alive = False
- mock_client.host_port = None
- manager.resume()
- mock_client.restore_app_connection.assert_not_called()
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_resume_not_alive_no_host_port(self, mock_class):
+ mock_client = mock_class.return_value
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ mock_client.is_alive = False
+ mock_client.host_port = None
+ manager.resume()
+ mock_client.restore_app_connection.assert_not_called()
- @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
- def test_attribute_access(self, mock_class):
- mock_client = mock.MagicMock()
- mock_class.return_value = mock_client
- manager = snippet_management_service.SnippetManagementService(
- mock.MagicMock())
- manager.add_snippet_client('foo', MOCK_PACKAGE)
- manager.foo.ha('param')
- mock_client.ha.assert_called_once_with('param')
+ @mock.patch(SNIPPET_CLIENT_CLASS_PATH)
+ def test_attribute_access(self, mock_class):
+ mock_client = mock.MagicMock()
+ mock_class.return_value = mock_client
+ manager = snippet_management_service.SnippetManagementService(
+ mock.MagicMock())
+ manager.add_snippet_client('foo', MOCK_PACKAGE)
+ manager.foo.ha('param')
+ mock_client.ha.assert_called_once_with('param')
if __name__ == '__main__':
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/sl4a_client_test.py b/tests/mobly/controllers/android_device_lib/sl4a_client_test.py
index e786b03..feff163 100755
--- a/tests/mobly/controllers/android_device_lib/sl4a_client_test.py
+++ b/tests/mobly/controllers/android_device_lib/sl4a_client_test.py
@@ -26,58 +26,58 @@
class Sl4aClientTest(jsonrpc_client_test_base.JsonRpcClientTestBase):
- """Unit tests for mobly.controllers.android_device_lib.sl4a_client.
- """
+ """Unit tests for mobly.controllers.android_device_lib.sl4a_client.
+ """
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_start_app_and_connect(self, mock_get_port,
- mock_start_standing_subprocess,
- mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess, resp_lines=[b'\n'])
- client = self._make_client()
- client.start_app_and_connect()
- self.assertEqual(8080, client.device_port)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_start_app_and_connect(self, mock_get_port,
+ mock_start_standing_subprocess,
+ mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess, resp_lines=[b'\n'])
+ client = self._make_client()
+ client.start_app_and_connect()
+ self.assertEqual(8080, client.device_port)
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_app_not_installed(self, mock_get_port,
- mock_start_standing_subprocess,
- mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess, resp_lines=[b'\n'])
- client = self._make_client(
- adb_proxy=mock_android_device.MockAdbProxy())
- with self.assertRaisesRegex(jsonrpc_client_base.AppStartError,
- '.* SL4A is not installed on .*'):
- client.start_app_and_connect()
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_app_not_installed(self, mock_get_port,
+ mock_start_standing_subprocess,
+ mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess, resp_lines=[b'\n'])
+ client = self._make_client(
+ adb_proxy=mock_android_device.MockAdbProxy())
+ with self.assertRaisesRegex(jsonrpc_client_base.AppStartError,
+ '.* SL4A is not installed on .*'):
+ client.start_app_and_connect()
- def _make_client(self, adb_proxy=None):
- adb_proxy = adb_proxy or mock_android_device.MockAdbProxy(
- installed_packages=['com.googlecode.android_scripting'])
- ad = mock.Mock()
- ad.adb = adb_proxy
- ad.build_info = {
- 'build_version_codename':
- ad.adb.getprop('ro.build.version.codename'),
- 'build_version_sdk': ad.adb.getprop('ro.build.version.sdk'),
- }
- return sl4a_client.Sl4aClient(ad=ad)
+ def _make_client(self, adb_proxy=None):
+ adb_proxy = adb_proxy or mock_android_device.MockAdbProxy(
+ installed_packages=['com.googlecode.android_scripting'])
+ ad = mock.Mock()
+ ad.adb = adb_proxy
+ ad.build_info = {
+ 'build_version_codename':
+ ad.adb.getprop('ro.build.version.codename'),
+ 'build_version_sdk': ad.adb.getprop('ro.build.version.sdk'),
+ }
+ return sl4a_client.Sl4aClient(ad=ad)
- def _setup_mock_instrumentation_cmd(self, mock_start_standing_subprocess,
- resp_lines):
- mock_proc = mock_start_standing_subprocess()
- mock_proc.stdout.readline.side_effect = resp_lines
+ def _setup_mock_instrumentation_cmd(self, mock_start_standing_subprocess,
+ resp_lines):
+ mock_proc = mock_start_standing_subprocess()
+ mock_proc.stdout.readline.side_effect = resp_lines
if __name__ == "__main__":
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/snippet_client_test.py b/tests/mobly/controllers/android_device_lib/snippet_client_test.py
index cb62345..412edcb 100755
--- a/tests/mobly/controllers/android_device_lib/snippet_client_test.py
+++ b/tests/mobly/controllers/android_device_lib/snippet_client_test.py
@@ -32,485 +32,485 @@
def get_print_function_name():
- """Gets the name of the print function for mocking.
+ """Gets the name of the print function for mocking.
- Returns:
- A str representing the print function to mock.
- """
- if sys.version_info >= (3, 0):
- return 'builtins.print'
- else:
- return '__builtin__.print'
+ Returns:
+ A str representing the print function to mock.
+ """
+ if sys.version_info >= (3, 0):
+ return 'builtins.print'
+ else:
+ return '__builtin__.print'
class SnippetClientTest(jsonrpc_client_test_base.JsonRpcClientTestBase):
- """Unit tests for mobly.controllers.android_device_lib.snippet_client.
- """
+ """Unit tests for mobly.controllers.android_device_lib.snippet_client.
+ """
- def test_check_app_installed_normal(self):
- sc = self._make_client()
- sc._check_app_installed()
+ def test_check_app_installed_normal(self):
+ sc = self._make_client()
+ sc._check_app_installed()
- def test_check_app_installed_fail_app_not_installed(self):
- sc = self._make_client(mock_android_device.MockAdbProxy())
- expected_msg = '.* %s is not installed.' % MOCK_PACKAGE_NAME
- with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
- expected_msg):
- sc._check_app_installed()
+ def test_check_app_installed_fail_app_not_installed(self):
+ sc = self._make_client(mock_android_device.MockAdbProxy())
+ expected_msg = '.* %s is not installed.' % MOCK_PACKAGE_NAME
+ with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
+ expected_msg):
+ sc._check_app_installed()
- def test_check_app_installed_fail_not_instrumented(self):
- sc = self._make_client(
- mock_android_device.MockAdbProxy(
- installed_packages=[MOCK_PACKAGE_NAME]))
- expected_msg = ('.* %s is installed, but it is not instrumented.' %
- MOCK_PACKAGE_NAME)
- with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
- expected_msg):
- sc._check_app_installed()
+ def test_check_app_installed_fail_not_instrumented(self):
+ sc = self._make_client(
+ mock_android_device.MockAdbProxy(
+ installed_packages=[MOCK_PACKAGE_NAME]))
+ expected_msg = ('.* %s is installed, but it is not instrumented.' %
+ MOCK_PACKAGE_NAME)
+ with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
+ expected_msg):
+ sc._check_app_installed()
- def test_check_app_installed_fail_target_not_installed(self):
- sc = self._make_client(
- mock_android_device.MockAdbProxy(instrumented_packages=[(
- MOCK_PACKAGE_NAME,
- snippet_client._INSTRUMENTATION_RUNNER_PACKAGE,
- MOCK_MISSING_PACKAGE_NAME)]))
- expected_msg = ('.* Instrumentation target %s is not installed.' %
- MOCK_MISSING_PACKAGE_NAME)
- with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
- expected_msg):
- sc._check_app_installed()
+ def test_check_app_installed_fail_target_not_installed(self):
+ sc = self._make_client(
+ mock_android_device.MockAdbProxy(instrumented_packages=[(
+ MOCK_PACKAGE_NAME,
+ snippet_client._INSTRUMENTATION_RUNNER_PACKAGE,
+ MOCK_MISSING_PACKAGE_NAME)]))
+ expected_msg = ('.* Instrumentation target %s is not installed.' %
+ MOCK_MISSING_PACKAGE_NAME)
+ with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
+ expected_msg):
+ sc._check_app_installed()
- @mock.patch('socket.create_connection')
- def test_snippet_start(self, mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- client = self._make_client()
- client.connect()
- result = client.testSnippetCall()
- self.assertEqual(123, result)
+ @mock.patch('socket.create_connection')
+ def test_snippet_start(self, mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ client = self._make_client()
+ client.connect()
+ result = client.testSnippetCall()
+ self.assertEqual(123, result)
- @mock.patch('socket.create_connection')
- def test_snippet_start_event_client(self, mock_create_connection):
- fake_file = self.setup_mock_socket_file(mock_create_connection)
- client = self._make_client()
- client.host_port = 123 # normally picked by start_app_and_connect
- client.connect()
- fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
- callback = client.testSnippetCall()
- self.assertEqual(123, callback.ret_value)
- self.assertEqual('1-0', callback._id)
+ @mock.patch('socket.create_connection')
+ def test_snippet_start_event_client(self, mock_create_connection):
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
+ client = self._make_client()
+ client.host_port = 123 # normally picked by start_app_and_connect
+ client.connect()
+ fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
+ callback = client.testSnippetCall()
+ self.assertEqual(123, callback.ret_value)
+ self.assertEqual('1-0', callback._id)
- # Check to make sure the event client is using the same port as the
- # main client.
- self.assertEqual(123, callback._event_client.host_port)
+ # Check to make sure the event client is using the same port as the
+ # main client.
+ self.assertEqual(123, callback._event_client.host_port)
- fake_file.resp = self.MOCK_RESP_WITH_ERROR
- with self.assertRaisesRegex(jsonrpc_client_base.ApiError, '1'):
- callback.getAll('eventName')
+ fake_file.resp = self.MOCK_RESP_WITH_ERROR
+ with self.assertRaisesRegex(jsonrpc_client_base.ApiError, '1'):
+ callback.getAll('eventName')
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_restore_event_client(self, mock_get_port,
- mock_create_connection):
- mock_get_port.return_value = 789
- fake_file = self.setup_mock_socket_file(mock_create_connection)
- client = self._make_client()
- client.host_port = 123 # normally picked by start_app_and_connect
- client.device_port = 456
- client.connect()
- fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
- callback = client.testSnippetCall()
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_restore_event_client(self, mock_get_port,
+ mock_create_connection):
+ mock_get_port.return_value = 789
+ fake_file = self.setup_mock_socket_file(mock_create_connection)
+ client = self._make_client()
+ client.host_port = 123 # normally picked by start_app_and_connect
+ client.device_port = 456
+ client.connect()
+ fake_file.resp = self.MOCK_RESP_WITH_CALLBACK
+ callback = client.testSnippetCall()
- # before reconnect, clients use previously selected ports
- self.assertEqual(123, client.host_port)
- self.assertEqual(456, client.device_port)
- self.assertEqual(123, callback._event_client.host_port)
- self.assertEqual(456, callback._event_client.device_port)
+ # before reconnect, clients use previously selected ports
+ self.assertEqual(123, client.host_port)
+ self.assertEqual(456, client.device_port)
+ self.assertEqual(123, callback._event_client.host_port)
+ self.assertEqual(456, callback._event_client.device_port)
- # after reconnect, if host port specified, clients use specified port
- client.restore_app_connection(port=321)
- self.assertEqual(321, client.host_port)
- self.assertEqual(456, client.device_port)
- self.assertEqual(321, callback._event_client.host_port)
- self.assertEqual(456, callback._event_client.device_port)
+ # after reconnect, if host port specified, clients use specified port
+ client.restore_app_connection(port=321)
+ self.assertEqual(321, client.host_port)
+ self.assertEqual(456, client.device_port)
+ self.assertEqual(321, callback._event_client.host_port)
+ self.assertEqual(456, callback._event_client.device_port)
- # after reconnect, if host port not specified, clients use selected
- # available port
- client.restore_app_connection()
- self.assertEqual(789, client.host_port)
- self.assertEqual(456, client.device_port)
- self.assertEqual(789, callback._event_client.host_port)
- self.assertEqual(456, callback._event_client.device_port)
+ # after reconnect, if host port not specified, clients use selected
+ # available port
+ client.restore_app_connection()
+ self.assertEqual(789, client.host_port)
+ self.assertEqual(456, client.device_port)
+ self.assertEqual(789, callback._event_client.host_port)
+ self.assertEqual(456, callback._event_client.device_port)
- # if unable to reconnect for any reason, a
- # jsonrpc_client_base.AppRestoreConnectionError is raised.
- mock_create_connection.side_effect = IOError('socket timed out')
- with self.assertRaisesRegex(
- jsonrpc_client_base.AppRestoreConnectionError,
- ('Failed to restore app connection for %s at host port %s, '
- 'device port %s') % (MOCK_PACKAGE_NAME, 789, 456)):
- client.restore_app_connection()
+ # if unable to reconnect for any reason, a
+ # jsonrpc_client_base.AppRestoreConnectionError is raised.
+ mock_create_connection.side_effect = IOError('socket timed out')
+ with self.assertRaisesRegex(
+ jsonrpc_client_base.AppRestoreConnectionError,
+ ('Failed to restore app connection for %s at host port %s, '
+ 'device port %s') % (MOCK_PACKAGE_NAME, 789, 456)):
+ client.restore_app_connection()
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_app_and_connect(self, mock_get_port,
- mock_start_standing_subprocess,
- mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'SNIPPET START, PROTOCOL 1 0\n',
- b'SNIPPET SERVING, PORT 123\n',
- ])
- client = self._make_client()
- client.start_app_and_connect()
- self.assertEqual(123, client.device_port)
- self.assertTrue(client.is_alive)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_app_and_connect(self, mock_get_port,
+ mock_start_standing_subprocess,
+ mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'SNIPPET START, PROTOCOL 1 0\n',
+ b'SNIPPET SERVING, PORT 123\n',
+ ])
+ client = self._make_client()
+ client.start_app_and_connect()
+ self.assertEqual(123, client.device_port)
+ self.assertTrue(client.is_alive)
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.stop_standing_subprocess')
- def test_snippet_stop_app(self, mock_stop_standing_subprocess,
- mock_create_connection):
- adb_proxy = mock.MagicMock()
- adb_proxy.shell.return_value = b'OK (0 tests)'
- client = self._make_client(adb_proxy)
- client.stop_app()
- self.assertFalse(client.is_alive)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.stop_standing_subprocess')
+ def test_snippet_stop_app(self, mock_stop_standing_subprocess,
+ mock_create_connection):
+ adb_proxy = mock.MagicMock()
+ adb_proxy.shell.return_value = b'OK (0 tests)'
+ client = self._make_client(adb_proxy)
+ client.stop_app()
+ self.assertFalse(client.is_alive)
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient.disconnect')
- def test_snippet_stop_app_raises(self, mock_disconnect,
- mock_create_connection):
- mock_disconnect.side_effect = Exception('ha')
- adb_proxy = mock.MagicMock()
- adb_proxy.shell.return_value = b'OK (0 tests)'
- client = self._make_client(adb_proxy)
- client.host_port = 1
- with self.assertRaisesRegex(Exception, 'ha'):
- client.stop_app()
- adb_proxy.forward.assert_called_once_with(['--remove', 'tcp:1'])
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient.disconnect')
+ def test_snippet_stop_app_raises(self, mock_disconnect,
+ mock_create_connection):
+ mock_disconnect.side_effect = Exception('ha')
+ adb_proxy = mock.MagicMock()
+ adb_proxy.shell.return_value = b'OK (0 tests)'
+ client = self._make_client(adb_proxy)
+ client.host_port = 1
+ with self.assertRaisesRegex(Exception, 'ha'):
+ client.stop_app()
+ adb_proxy.forward.assert_called_once_with(['--remove', 'tcp:1'])
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- @mock.patch(
- 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.'
- 'disable_hidden_api_blacklist')
- @mock.patch(
- 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.'
- 'stop_app')
- def test_start_app_and_connect_precheck_fail(
- self, mock_stop, mock_precheck, mock_get_port,
- mock_start_standing_subprocess, mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'SNIPPET START, PROTOCOL 1 0\n',
- b'SNIPPET SERVING, PORT 123\n',
- ])
- client = self._make_client()
- mock_precheck.side_effect = snippet_client.AppStartPreCheckError(
- client.ad, 'ha')
- with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
- 'ha'):
- client.start_app_and_connect()
- mock_stop.assert_not_called()
- self.assertFalse(client.is_alive)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.'
+ 'disable_hidden_api_blacklist')
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.'
+ 'stop_app')
+ def test_start_app_and_connect_precheck_fail(
+ self, mock_stop, mock_precheck, mock_get_port,
+ mock_start_standing_subprocess, mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'SNIPPET START, PROTOCOL 1 0\n',
+ b'SNIPPET SERVING, PORT 123\n',
+ ])
+ client = self._make_client()
+ mock_precheck.side_effect = snippet_client.AppStartPreCheckError(
+ client.ad, 'ha')
+ with self.assertRaisesRegex(snippet_client.AppStartPreCheckError,
+ 'ha'):
+ client.start_app_and_connect()
+ mock_stop.assert_not_called()
+ self.assertFalse(client.is_alive)
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- @mock.patch(
- 'mobly.controllers.android_device_lib.snippet_client.SnippetClient._start_app_and_connect'
- )
- @mock.patch(
- 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.stop_app'
- )
- def test_start_app_and_connect_generic_error(
- self, mock_stop, mock_start, mock_get_port,
- mock_start_standing_subprocess, mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'SNIPPET START, PROTOCOL 1 0\n',
- b'SNIPPET SERVING, PORT 123\n',
- ])
- client = self._make_client()
- mock_start.side_effect = Exception('ha')
- with self.assertRaisesRegex(Exception, 'ha'):
- client.start_app_and_connect()
- mock_stop.assert_called_once_with()
- self.assertFalse(client.is_alive)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.snippet_client.SnippetClient._start_app_and_connect'
+ )
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.stop_app'
+ )
+ def test_start_app_and_connect_generic_error(
+ self, mock_stop, mock_start, mock_get_port,
+ mock_start_standing_subprocess, mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'SNIPPET START, PROTOCOL 1 0\n',
+ b'SNIPPET SERVING, PORT 123\n',
+ ])
+ client = self._make_client()
+ mock_start.side_effect = Exception('ha')
+ with self.assertRaisesRegex(Exception, 'ha'):
+ client.start_app_and_connect()
+ mock_stop.assert_called_once_with()
+ self.assertFalse(client.is_alive)
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- @mock.patch(
- 'mobly.controllers.android_device_lib.snippet_client.SnippetClient._start_app_and_connect'
- )
- @mock.patch(
- 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.stop_app'
- )
- def test_start_app_and_connect_fail_stop_also_fail(
- self, mock_stop, mock_start, mock_get_port,
- mock_start_standing_subprocess, mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'SNIPPET START, PROTOCOL 1 0\n',
- b'SNIPPET SERVING, PORT 123\n',
- ])
- client = self._make_client()
- mock_start.side_effect = Exception('Some error')
- mock_stop.side_effect = Exception('Another error')
- with self.assertRaisesRegex(Exception, 'Some error'):
- client.start_app_and_connect()
- mock_stop.assert_called_once_with()
- self.assertFalse(client.is_alive)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.snippet_client.SnippetClient._start_app_and_connect'
+ )
+ @mock.patch(
+ 'mobly.controllers.android_device_lib.snippet_client.SnippetClient.stop_app'
+ )
+ def test_start_app_and_connect_fail_stop_also_fail(
+ self, mock_stop, mock_start, mock_get_port,
+ mock_start_standing_subprocess, mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'SNIPPET START, PROTOCOL 1 0\n',
+ b'SNIPPET SERVING, PORT 123\n',
+ ])
+ client = self._make_client()
+ mock_start.side_effect = Exception('Some error')
+ mock_stop.side_effect = Exception('Another error')
+ with self.assertRaisesRegex(Exception, 'Some error'):
+ client.start_app_and_connect()
+ mock_stop.assert_called_once_with()
+ self.assertFalse(client.is_alive)
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient._do_start_app')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient._check_app_installed')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient._read_protocol_line')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient.connect')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_on_sdk_21(self, mock_get_port, mock_connect,
- mock_read_protocol_line,
- mock_check_app_installed,
- mock_do_start_app):
- """Check that `--user` is not added to start command on SDK < 24."""
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient._do_start_app')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient._check_app_installed')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient._read_protocol_line')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient.connect')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_on_sdk_21(self, mock_get_port, mock_connect,
+ mock_read_protocol_line,
+ mock_check_app_installed,
+ mock_do_start_app):
+ """Check that `--user` is not added to start command on SDK < 24."""
- def _mocked_shell(arg):
- if 'setsid' in arg:
- raise adb.AdbError('cmd', 'stdout', 'stderr', 'ret_code')
- else:
- return b'nohup'
+ def _mocked_shell(arg):
+ if 'setsid' in arg:
+ raise adb.AdbError('cmd', 'stdout', 'stderr', 'ret_code')
+ else:
+ return b'nohup'
- mock_get_port.return_value = 123
- mock_read_protocol_line.side_effect = [
- 'SNIPPET START, PROTOCOL 1 234',
- 'SNIPPET SERVING, PORT 1234',
- 'SNIPPET START, PROTOCOL 1 234',
- 'SNIPPET SERVING, PORT 1234',
- 'SNIPPET START, PROTOCOL 1 234',
- 'SNIPPET SERVING, PORT 1234',
- ]
+ mock_get_port.return_value = 123
+ mock_read_protocol_line.side_effect = [
+ 'SNIPPET START, PROTOCOL 1 234',
+ 'SNIPPET SERVING, PORT 1234',
+ 'SNIPPET START, PROTOCOL 1 234',
+ 'SNIPPET SERVING, PORT 1234',
+ 'SNIPPET START, PROTOCOL 1 234',
+ 'SNIPPET SERVING, PORT 1234',
+ ]
- # Test 'setsid' exists
- client = self._make_client()
- client._ad.build_info['build_version_sdk'] = 21
- client._adb.shell = mock.Mock(return_value=b'setsid')
- client.start_app_and_connect()
- cmd_setsid = '%s am instrument -w -e action start %s/%s' % (
- snippet_client._SETSID_COMMAND, MOCK_PACKAGE_NAME,
- snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
- mock_do_start_app.assert_has_calls([mock.call(cmd_setsid)])
+ # Test 'setsid' exists
+ client = self._make_client()
+ client._ad.build_info['build_version_sdk'] = 21
+ client._adb.shell = mock.Mock(return_value=b'setsid')
+ client.start_app_and_connect()
+ cmd_setsid = '%s am instrument -w -e action start %s/%s' % (
+ snippet_client._SETSID_COMMAND, MOCK_PACKAGE_NAME,
+ snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
+ mock_do_start_app.assert_has_calls([mock.call(cmd_setsid)])
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient._do_start_app')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient._check_app_installed')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient._read_protocol_line')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'SnippetClient.connect')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_app_and_connect_persistent_session(
- self, mock_get_port, mock_connect, mock_read_protocol_line,
- mock_check_app_installed, mock_do_start_app):
- def _mocked_shell(arg):
- if 'setsid' in arg:
- raise adb.AdbError('cmd', 'stdout', 'stderr', 'ret_code')
- else:
- return b'nohup'
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient._do_start_app')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient._check_app_installed')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient._read_protocol_line')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'SnippetClient.connect')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_app_and_connect_persistent_session(
+ self, mock_get_port, mock_connect, mock_read_protocol_line,
+ mock_check_app_installed, mock_do_start_app):
+ def _mocked_shell(arg):
+ if 'setsid' in arg:
+ raise adb.AdbError('cmd', 'stdout', 'stderr', 'ret_code')
+ else:
+ return b'nohup'
- mock_get_port.return_value = 123
- mock_read_protocol_line.side_effect = [
- 'SNIPPET START, PROTOCOL 1 234',
- 'SNIPPET SERVING, PORT 1234',
- 'SNIPPET START, PROTOCOL 1 234',
- 'SNIPPET SERVING, PORT 1234',
- 'SNIPPET START, PROTOCOL 1 234',
- 'SNIPPET SERVING, PORT 1234',
- ]
+ mock_get_port.return_value = 123
+ mock_read_protocol_line.side_effect = [
+ 'SNIPPET START, PROTOCOL 1 234',
+ 'SNIPPET SERVING, PORT 1234',
+ 'SNIPPET START, PROTOCOL 1 234',
+ 'SNIPPET SERVING, PORT 1234',
+ 'SNIPPET START, PROTOCOL 1 234',
+ 'SNIPPET SERVING, PORT 1234',
+ ]
- # Test 'setsid' exists
- client = self._make_client()
- client._adb.shell = mock.Mock(return_value=b'setsid')
- client.start_app_and_connect()
- cmd_setsid = '%s am instrument --user %s -w -e action start %s/%s' % (
- snippet_client._SETSID_COMMAND, MOCK_USER_ID, MOCK_PACKAGE_NAME,
- snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
- mock_do_start_app.assert_has_calls([mock.call(cmd_setsid)])
+ # Test 'setsid' exists
+ client = self._make_client()
+ client._adb.shell = mock.Mock(return_value=b'setsid')
+ client.start_app_and_connect()
+ cmd_setsid = '%s am instrument --user %s -w -e action start %s/%s' % (
+ snippet_client._SETSID_COMMAND, MOCK_USER_ID, MOCK_PACKAGE_NAME,
+ snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
+ mock_do_start_app.assert_has_calls([mock.call(cmd_setsid)])
- # Test 'setsid' does not exist, but 'nohup' exsits
- client = self._make_client()
- client._adb.shell = _mocked_shell
- client.start_app_and_connect()
- cmd_nohup = '%s am instrument --user %s -w -e action start %s/%s' % (
- snippet_client._NOHUP_COMMAND, MOCK_USER_ID, MOCK_PACKAGE_NAME,
- snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
- mock_do_start_app.assert_has_calls(
- [mock.call(cmd_setsid),
- mock.call(cmd_nohup)])
+ # Test 'setsid' does not exist, but 'nohup' exsits
+ client = self._make_client()
+ client._adb.shell = _mocked_shell
+ client.start_app_and_connect()
+ cmd_nohup = '%s am instrument --user %s -w -e action start %s/%s' % (
+ snippet_client._NOHUP_COMMAND, MOCK_USER_ID, MOCK_PACKAGE_NAME,
+ snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
+ mock_do_start_app.assert_has_calls(
+ [mock.call(cmd_setsid),
+ mock.call(cmd_nohup)])
- # Test both 'setsid' and 'nohup' do not exist
- client._adb.shell = mock.Mock(
- side_effect=adb.AdbError('cmd', 'stdout', 'stderr', 'ret_code'))
- client = self._make_client()
- client.start_app_and_connect()
- cmd_not_persist = ' am instrument --user %s -w -e action start %s/%s' % (
- MOCK_USER_ID, MOCK_PACKAGE_NAME,
- snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
- mock_do_start_app.assert_has_calls([
- mock.call(cmd_setsid),
- mock.call(cmd_nohup),
- mock.call(cmd_not_persist)
- ])
+ # Test both 'setsid' and 'nohup' do not exist
+ client._adb.shell = mock.Mock(
+ side_effect=adb.AdbError('cmd', 'stdout', 'stderr', 'ret_code'))
+ client = self._make_client()
+ client.start_app_and_connect()
+ cmd_not_persist = ' am instrument --user %s -w -e action start %s/%s' % (
+ MOCK_USER_ID, MOCK_PACKAGE_NAME,
+ snippet_client._INSTRUMENTATION_RUNNER_PACKAGE)
+ mock_do_start_app.assert_has_calls([
+ mock.call(cmd_setsid),
+ mock.call(cmd_nohup),
+ mock.call(cmd_not_persist)
+ ])
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_app_crash(self, mock_get_port,
- mock_start_standing_subprocess,
- mock_create_connection):
- mock_get_port.return_value = 456
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'INSTRUMENTATION_RESULT: shortMsg=Process crashed.\n'
- ])
- client = self._make_client()
- with self.assertRaisesRegex(
- snippet_client.ProtocolVersionError,
- 'INSTRUMENTATION_RESULT: shortMsg=Process crashed.'):
- client.start_app_and_connect()
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_app_crash(self, mock_get_port,
+ mock_start_standing_subprocess,
+ mock_create_connection):
+ mock_get_port.return_value = 456
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'INSTRUMENTATION_RESULT: shortMsg=Process crashed.\n'
+ ])
+ client = self._make_client()
+ with self.assertRaisesRegex(
+ snippet_client.ProtocolVersionError,
+ 'INSTRUMENTATION_RESULT: shortMsg=Process crashed.'):
+ client.start_app_and_connect()
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_app_and_connect_unknown_protocol(
- self, mock_get_port, mock_start_standing_subprocess):
- mock_get_port.return_value = 789
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[b'SNIPPET START, PROTOCOL 99 0\n'])
- client = self._make_client()
- with self.assertRaisesRegex(snippet_client.ProtocolVersionError,
- 'SNIPPET START, PROTOCOL 99 0'):
- client.start_app_and_connect()
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_app_and_connect_unknown_protocol(
+ self, mock_get_port, mock_start_standing_subprocess):
+ mock_get_port.return_value = 789
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[b'SNIPPET START, PROTOCOL 99 0\n'])
+ client = self._make_client()
+ with self.assertRaisesRegex(snippet_client.ProtocolVersionError,
+ 'SNIPPET START, PROTOCOL 99 0'):
+ client.start_app_and_connect()
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_app_and_connect_header_junk(
- self, mock_get_port, mock_start_standing_subprocess,
- mock_create_connection):
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'This is some header junk\n',
- b'Some phones print arbitrary output\n',
- b'SNIPPET START, PROTOCOL 1 0\n',
- b'Maybe in the middle too\n',
- b'SNIPPET SERVING, PORT 123\n',
- ])
- client = self._make_client()
- client.start_app_and_connect()
- self.assertEqual(123, client.device_port)
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_app_and_connect_header_junk(
+ self, mock_get_port, mock_start_standing_subprocess,
+ mock_create_connection):
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'This is some header junk\n',
+ b'Some phones print arbitrary output\n',
+ b'SNIPPET START, PROTOCOL 1 0\n',
+ b'Maybe in the middle too\n',
+ b'SNIPPET SERVING, PORT 123\n',
+ ])
+ client = self._make_client()
+ client.start_app_and_connect()
+ self.assertEqual(123, client.device_port)
- @mock.patch('socket.create_connection')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.start_standing_subprocess')
- @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
- 'utils.get_available_host_port')
- def test_snippet_start_app_and_connect_no_valid_line(
- self, mock_get_port, mock_start_standing_subprocess,
- mock_create_connection):
- mock_get_port.return_value = 456
- self.setup_mock_socket_file(mock_create_connection)
- self._setup_mock_instrumentation_cmd(
- mock_start_standing_subprocess,
- resp_lines=[
- b'This is some header junk\n',
- b'Some phones print arbitrary output\n',
- b'', # readline uses '' to mark EOF
- ])
- client = self._make_client()
- with self.assertRaisesRegex(jsonrpc_client_base.AppStartError,
- 'Unexpected EOF waiting for app to start'):
- client.start_app_and_connect()
+ @mock.patch('socket.create_connection')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.start_standing_subprocess')
+ @mock.patch('mobly.controllers.android_device_lib.snippet_client.'
+ 'utils.get_available_host_port')
+ def test_snippet_start_app_and_connect_no_valid_line(
+ self, mock_get_port, mock_start_standing_subprocess,
+ mock_create_connection):
+ mock_get_port.return_value = 456
+ self.setup_mock_socket_file(mock_create_connection)
+ self._setup_mock_instrumentation_cmd(
+ mock_start_standing_subprocess,
+ resp_lines=[
+ b'This is some header junk\n',
+ b'Some phones print arbitrary output\n',
+ b'', # readline uses '' to mark EOF
+ ])
+ client = self._make_client()
+ with self.assertRaisesRegex(jsonrpc_client_base.AppStartError,
+ 'Unexpected EOF waiting for app to start'):
+ client.start_app_and_connect()
- @mock.patch(get_print_function_name())
- def test_help_rpc_when_printing_by_default(self, mock_print):
- client = self._make_client()
- mock_rpc = mock.MagicMock()
- client._rpc = mock_rpc
+ @mock.patch(get_print_function_name())
+ def test_help_rpc_when_printing_by_default(self, mock_print):
+ client = self._make_client()
+ mock_rpc = mock.MagicMock()
+ client._rpc = mock_rpc
- result = client.help()
- mock_rpc.assert_called_once_with('help')
- self.assertEqual(None, result)
- mock_print.assert_called_once_with(mock_rpc.return_value)
+ result = client.help()
+ mock_rpc.assert_called_once_with('help')
+ self.assertEqual(None, result)
+ mock_print.assert_called_once_with(mock_rpc.return_value)
- @mock.patch(get_print_function_name())
- def test_help_rpc_when_not_printing(self, mock_print):
- client = self._make_client()
- mock_rpc = mock.MagicMock()
- client._rpc = mock_rpc
+ @mock.patch(get_print_function_name())
+ def test_help_rpc_when_not_printing(self, mock_print):
+ client = self._make_client()
+ mock_rpc = mock.MagicMock()
+ client._rpc = mock_rpc
- result = client.help(print_output=False)
- mock_rpc.assert_called_once_with('help')
- self.assertEqual(mock_rpc.return_value, result)
- mock_print.assert_not_called()
+ result = client.help(print_output=False)
+ mock_rpc.assert_called_once_with('help')
+ self.assertEqual(mock_rpc.return_value, result)
+ mock_print.assert_not_called()
- def _make_client(self, adb_proxy=None):
- adb_proxy = adb_proxy or mock_android_device.MockAdbProxy(
- instrumented_packages=[(
- MOCK_PACKAGE_NAME,
- snippet_client._INSTRUMENTATION_RUNNER_PACKAGE,
- MOCK_PACKAGE_NAME)])
- ad = mock.Mock()
- ad.adb = adb_proxy
- ad.adb.current_user_id = MOCK_USER_ID
- ad.build_info = {
- 'build_version_codename':
- ad.adb.getprop('ro.build.version.codename'),
- 'build_version_sdk': ad.adb.getprop('ro.build.version.sdk'),
- }
- return snippet_client.SnippetClient(package=MOCK_PACKAGE_NAME, ad=ad)
+ def _make_client(self, adb_proxy=None):
+ adb_proxy = adb_proxy or mock_android_device.MockAdbProxy(
+ instrumented_packages=[(
+ MOCK_PACKAGE_NAME,
+ snippet_client._INSTRUMENTATION_RUNNER_PACKAGE,
+ MOCK_PACKAGE_NAME)])
+ ad = mock.Mock()
+ ad.adb = adb_proxy
+ ad.adb.current_user_id = MOCK_USER_ID
+ ad.build_info = {
+ 'build_version_codename':
+ ad.adb.getprop('ro.build.version.codename'),
+ 'build_version_sdk': ad.adb.getprop('ro.build.version.sdk'),
+ }
+ return snippet_client.SnippetClient(package=MOCK_PACKAGE_NAME, ad=ad)
- def _setup_mock_instrumentation_cmd(self, mock_start_standing_subprocess,
- resp_lines):
- mock_proc = mock_start_standing_subprocess()
- mock_proc.stdout.readline.side_effect = resp_lines
+ def _setup_mock_instrumentation_cmd(self, mock_start_standing_subprocess,
+ resp_lines):
+ mock_proc = mock_start_standing_subprocess()
+ mock_proc.stdout.readline.side_effect = resp_lines
if __name__ == "__main__":
- unittest.main()
+ unittest.main()
diff --git a/tests/mobly/controllers/android_device_lib/snippet_event_test.py b/tests/mobly/controllers/android_device_lib/snippet_event_test.py
index 1b782fe..a995197 100755
--- a/tests/mobly/controllers/android_device_lib/snippet_event_test.py
+++ b/tests/mobly/controllers/android_device_lib/snippet_event_test.py
@@ -26,16 +26,16 @@
class SnippetEventTest(unittest.TestCase):
- def test_basic(self):
- """Verifies that an event object can be created and logged properly.
- """
- event = snippet_event.SnippetEvent(MOCK_CALLBACK_ID, MOCK_EVENT_NAME,
- MOCK_CREATION_TIME, MOCK_DATA)
- self.assertEqual(
- repr(event),
- "SnippetEvent(callback_id: myCallbackId, name: onXyzEvent, "
- "creation_time: 12345678, data: {'foo': 'bar'})")
+ def test_basic(self):
+ """Verifies that an event object can be created and logged properly.
+ """
+ event = snippet_event.SnippetEvent(MOCK_CALLBACK_ID, MOCK_EVENT_NAME,
+ MOCK_CREATION_TIME, MOCK_DATA)
+ self.assertEqual(
+ repr(event),
+ "SnippetEvent(callback_id: myCallbackId, name: onXyzEvent, "
+ "creation_time: 12345678, data: {'foo': 'bar'})")
if __name__ == "__main__":
- unittest.main()
+ unittest.main()
