KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use adb method in ATX
Best Python code snippet using ATX
adb_call_parser_test.py
Source:adb_call_parser_test.py 
1# coding=utf-82# Copyright 2022 DeepMind Technologies Limited.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Tests for adb_call_parser."""16import builtins17import os18import subprocess19import sys20from unittest import mock21from absl.testing import absltest22from absl.testing import parameterized23from android_env.components import adb_call_parser24from android_env.components import adb_controller25from android_env.proto import adb_pb226class AdbCallParserTest(parameterized.TestCase):27 def test_unknown_command(self):28 adb = mock.create_autospec(adb_controller.AdbController)29 parser = adb_call_parser.AdbCallParser(30 adb, tmp_dir=absltest.get_default_test_tmpdir())31 request = adb_pb2.AdbRequest()32 response = parser.parse(request)33 self.assertEqual(response.status,34 adb_pb2.AdbResponse.Status.UNKNOWN_COMMAND)35 def test_invalid_timeout(self):36 """AdbRequest.timeout_sec must be positive."""37 adb = mock.create_autospec(adb_controller.AdbController)38 parser = adb_call_parser.AdbCallParser(39 adb, tmp_dir=absltest.get_default_test_tmpdir())40 request = adb_pb2.AdbRequest()41 request.tap.x = 12342 request.timeout_sec = -543 response = parser.parse(request)44 self.assertEqual(response.status,45 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)46 @mock.patch.object(os.path, 'exists', autospec=True)47 def test_install_apk_file_not_found(self, mock_exists):48 adb = mock.create_autospec(adb_controller.AdbController)49 parser = adb_call_parser.AdbCallParser(50 adb, tmp_dir=absltest.get_default_test_tmpdir())51 request = adb_pb2.AdbRequest()52 request.install_apk.filesystem.path = '/my/home/game.apk'53 mock_exists.return_value = False54 response = parser.parse(request)55 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.INTERNAL_ERROR)56 self.assertNotEmpty(response.error_message)57 adb.execute_command.assert_not_called()58 @mock.patch.object(os.path, 'exists', autospec=True)59 def test_install_apk_successful(self, mock_exists):60 adb = mock.create_autospec(adb_controller.AdbController)61 parser = adb_call_parser.AdbCallParser(62 adb, tmp_dir=absltest.get_default_test_tmpdir())63 request = adb_pb2.AdbRequest()64 request.install_apk.filesystem.path = '/my/home/game.apk'65 mock_exists.return_value = True66 response = parser.parse(request)67 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)68 self.assertEmpty(response.error_message)69 adb.execute_command.assert_called_once_with(70 ['install', '-r', '-t', '-g', '/my/home/game.apk'], None)71 def test_start_activity_empty_full_activity(self):72 adb = mock.create_autospec(adb_controller.AdbController)73 parser = adb_call_parser.AdbCallParser(74 adb, tmp_dir=absltest.get_default_test_tmpdir())75 request = adb_pb2.AdbRequest()76 request.start_activity.extra_args.extend(['blah'])77 response = parser.parse(request)78 self.assertEqual(response.status,79 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)80 self.assertNotEmpty(response.error_message)81 def test_start_activity_successful(self):82 adb = mock.create_autospec(adb_controller.AdbController)83 command_output = (84 b'Stopping: my.project.SplashActivity\n'85 b'Starting: Intent { cmp=my.project.SplashActivity }\n')86 adb.execute_command.return_value = command_output87 parser = adb_call_parser.AdbCallParser(88 adb, tmp_dir=absltest.get_default_test_tmpdir())89 request = adb_pb2.AdbRequest()90 request.start_activity.full_activity = 'my.project.SplashActivity'91 request.start_activity.extra_args.extend(['blah'])92 request.start_activity.force_stop = True93 response = parser.parse(request)94 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)95 self.assertEmpty(response.error_message)96 adb.execute_command.assert_has_calls([97 mock.call([98 'shell', 'am', 'start', '-S', '-W', '-n',99 'my.project.SplashActivity', 'blah'100 ], timeout=None),101 ])102 def test_start_activity_successful_no_force_stop(self):103 adb = mock.create_autospec(adb_controller.AdbController)104 command_output = (105 b'Stopping: my.project.SplashActivity\n'106 b'Starting: Intent { cmp=my.project.SplashActivity }\n')107 adb.execute_command.return_value = command_output108 parser = adb_call_parser.AdbCallParser(109 adb, tmp_dir=absltest.get_default_test_tmpdir())110 request = adb_pb2.AdbRequest()111 request.start_activity.full_activity = 'my.project.SplashActivity'112 request.start_activity.extra_args.extend(['blah'])113 request.start_activity.force_stop = False114 response = parser.parse(request)115 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)116 self.assertEmpty(response.error_message)117 adb.execute_command.assert_has_calls([118 mock.call([119 'shell', 'am', 'start', '', '-W', '-n', 'my.project.SplashActivity',120 'blah'121 ],122 timeout=None),123 ])124 def test_start_activity_error(self):125 adb = mock.create_autospec(adb_controller.AdbController)126 command_output = (127 b'Stopping: my.project.SplashActivity\n'128 b'Starting: Intent { cmp=my.project.SplashActivity }\n'129 b'Error: Activity not started, unknown error code 101\n')130 adb.execute_command.return_value = command_output131 parser = adb_call_parser.AdbCallParser(132 adb, tmp_dir=absltest.get_default_test_tmpdir())133 request = adb_pb2.AdbRequest()134 request.start_activity.full_activity = 'my.project.SplashActivity'135 request.start_activity.extra_args.extend(['blah'])136 response = parser.parse(request)137 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.INTERNAL_ERROR)138 self.assertEqual(response.error_message,139 f'start_activity failed with error: {str(command_output)}')140 def test_force_stop(self):141 adb = mock.create_autospec(adb_controller.AdbController)142 parser = adb_call_parser.AdbCallParser(143 adb, tmp_dir=absltest.get_default_test_tmpdir())144 request = adb_pb2.AdbRequest()145 request.force_stop.package_name = 'my.project'146 response = parser.parse(request)147 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)148 self.assertEmpty(response.error_message)149 adb.execute_command.assert_called_once_with(150 ['shell', 'am', 'force-stop', 'my.project'], None)151 def test_grant_permissions_empty_package_name(self):152 adb = mock.create_autospec(adb_controller.AdbController)153 parser = adb_call_parser.AdbCallParser(154 adb, tmp_dir=absltest.get_default_test_tmpdir())155 request = adb_pb2.AdbRequest()156 request.package_manager.grant.permissions.extend(['perm1', 'perm2'])157 response = parser.parse(request)158 self.assertEqual(response.status,159 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)160 self.assertNotEmpty(response.error_message)161 def test_grant_permissions_empty_permissions(self):162 adb = mock.create_autospec(adb_controller.AdbController)163 parser = adb_call_parser.AdbCallParser(164 adb, tmp_dir=absltest.get_default_test_tmpdir())165 request = adb_pb2.AdbRequest()166 request.package_manager.grant.package_name = 'my.project'167 response = parser.parse(request)168 self.assertEqual(response.status,169 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)170 self.assertNotEmpty(response.error_message)171 def test_grant_permissions_successful(self):172 adb = mock.create_autospec(adb_controller.AdbController)173 adb.execute_command.return_value = b'whatever'174 parser = adb_call_parser.AdbCallParser(175 adb, tmp_dir=absltest.get_default_test_tmpdir())176 request = adb_pb2.AdbRequest()177 request.package_manager.grant.package_name = 'my.project'178 request.package_manager.grant.permissions.extend(['perm1', 'perm2'])179 response = parser.parse(request)180 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)181 self.assertEmpty(response.error_message)182 adb.execute_command.assert_has_calls([183 mock.call(['shell', 'pm', 'grant', 'my.project', 'perm1'], None),184 mock.call(['shell', 'pm', 'grant', 'my.project', 'perm2'], None),185 ])186 def test_press_button_invalid_button(self):187 adb = mock.create_autospec(adb_controller.AdbController)188 parser = adb_call_parser.AdbCallParser(189 adb, tmp_dir=absltest.get_default_test_tmpdir())190 request = adb_pb2.AdbRequest()191 request.press_button.button = 99999192 response = parser.parse(request)193 self.assertEqual(response.status,194 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)195 self.assertNotEmpty(response.error_message)196 def test_press_button_successful(self):197 adb = mock.create_autospec(adb_controller.AdbController)198 adb.execute_command.return_value = b''199 parser = adb_call_parser.AdbCallParser(200 adb, tmp_dir=absltest.get_default_test_tmpdir())201 # HOME.202 request = adb_pb2.AdbRequest()203 request.press_button.button = adb_pb2.AdbRequest.PressButton.Button.HOME204 response = parser.parse(request)205 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)206 self.assertEmpty(response.error_message)207 adb.execute_command.assert_called_with(208 ['shell', 'input', 'keyevent', 'KEYCODE_HOME'], None)209 # BACK.210 request = adb_pb2.AdbRequest()211 request.press_button.button = adb_pb2.AdbRequest.PressButton.Button.BACK212 response = parser.parse(request)213 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)214 self.assertEmpty(response.error_message)215 adb.execute_command.assert_called_with(216 ['shell', 'input', 'keyevent', 'KEYCODE_BACK'], None)217 # ENTER.218 request = adb_pb2.AdbRequest()219 request.press_button.button = adb_pb2.AdbRequest.PressButton.Button.ENTER220 response = parser.parse(request)221 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)222 self.assertEmpty(response.error_message)223 adb.execute_command.assert_called_with(224 ['shell', 'input', 'keyevent', 'KEYCODE_ENTER'], None)225 def test_start_screen_pinning_package_not_found(self):226 adb = mock.create_autospec(adb_controller.AdbController)227 adb.execute_command.return_value = (228 b' taskId=12345: my.project.AnotherActivity visible=true'229 b' topActivity=ComponentInfo{my.project.AnotherActivity}')230 parser = adb_call_parser.AdbCallParser(231 adb, tmp_dir=absltest.get_default_test_tmpdir())232 request = adb_pb2.AdbRequest()233 request.start_screen_pinning.full_activity = 'my.project.AmazingActivity'234 response = parser.parse(request)235 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.INTERNAL_ERROR)236 self.assertNotEmpty(response.error_message)237 adb.execute_command.assert_called_once_with(238 ['shell', 'am', 'stack', 'list'], None)239 def test_start_screen_pinning_successful(self):240 adb = mock.create_autospec(adb_controller.AdbController)241 adb.execute_command.return_value = (242 b' taskId=12345: my.project.AmazingActivity visible=true'243 b' topActivity=ComponentInfo{my.project.AmazingActivity}')244 parser = adb_call_parser.AdbCallParser(245 adb, tmp_dir=absltest.get_default_test_tmpdir())246 request = adb_pb2.AdbRequest()247 request.start_screen_pinning.full_activity = 'my.project.AmazingActivity'248 response = parser.parse(request)249 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)250 self.assertEmpty(response.error_message)251 adb.execute_command.assert_has_calls([252 mock.call(['shell', 'am', 'stack', 'list'], None),253 mock.call(['shell', 'am', 'task', 'lock', '12345'], None),254 ])255 def test_start_screen_pinning_base_activity(self):256 adb = mock.create_autospec(adb_controller.AdbController)257 adb.execute_command.return_value = (258 b' taskId=12345: my.project.MainActivity visible=true'259 b' topActivity=ComponentInfo{my.project.TopActivity}')260 parser = adb_call_parser.AdbCallParser(261 adb, tmp_dir=absltest.get_default_test_tmpdir())262 request = adb_pb2.AdbRequest()263 request.start_screen_pinning.full_activity = 'my.project.MainActivity'264 response = parser.parse(request)265 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)266 self.assertEmpty(response.error_message)267 adb.execute_command.assert_has_calls([268 mock.call(['shell', 'am', 'stack', 'list'], None),269 mock.call(['shell', 'am', 'task', 'lock', '12345'], None),270 ])271 def test_start_screen_pinning_top_activity(self):272 adb = mock.create_autospec(adb_controller.AdbController)273 adb.execute_command.return_value = (274 b' taskId=12345: my.project.MainActivity visible=true'275 b' topActivity=ComponentInfo{my.project.TopActivity}')276 parser = adb_call_parser.AdbCallParser(277 adb, tmp_dir=absltest.get_default_test_tmpdir())278 request = adb_pb2.AdbRequest()279 request.start_screen_pinning.full_activity = 'my.project.TopActivity'280 response = parser.parse(request)281 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)282 self.assertEmpty(response.error_message)283 adb.execute_command.assert_has_calls([284 mock.call(['shell', 'am', 'stack', 'list'], None),285 mock.call(['shell', 'am', 'task', 'lock', '12345'], None),286 ])287 def test_start_intent_empty_data_uri(self):288 adb = mock.create_autospec(adb_controller.AdbController)289 parser = adb_call_parser.AdbCallParser(290 adb, tmp_dir=absltest.get_default_test_tmpdir())291 request = adb_pb2.AdbRequest()292 request.start_intent.action = 'BEST_ACTION'293 request.start_intent.package_name = 'my.project'294 response = parser.parse(request)295 self.assertEqual(response.status,296 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)297 self.assertNotEmpty(response.error_message)298 def test_start_intent_empty_action(self):299 adb = mock.create_autospec(adb_controller.AdbController)300 parser = adb_call_parser.AdbCallParser(301 adb, tmp_dir=absltest.get_default_test_tmpdir())302 request = adb_pb2.AdbRequest()303 request.start_intent.data_uri = '{"some": 123}'304 request.start_intent.package_name = 'my.project'305 response = parser.parse(request)306 self.assertEqual(response.status,307 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)308 self.assertNotEmpty(response.error_message)309 def test_start_intent_empty_package_name(self):310 adb = mock.create_autospec(adb_controller.AdbController)311 parser = adb_call_parser.AdbCallParser(312 adb, tmp_dir=absltest.get_default_test_tmpdir())313 request = adb_pb2.AdbRequest()314 request.start_intent.data_uri = '{"some": 123}'315 request.start_intent.action = 'BEST_ACTION'316 response = parser.parse(request)317 self.assertEqual(response.status,318 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)319 self.assertNotEmpty(response.error_message)320 def test_start_intent_successful(self):321 adb = mock.create_autospec(adb_controller.AdbController)322 parser = adb_call_parser.AdbCallParser(323 adb, tmp_dir=absltest.get_default_test_tmpdir())324 request = adb_pb2.AdbRequest()325 request.start_intent.data_uri = '{"some": 123}'326 request.start_intent.action = 'BEST_ACTION'327 request.start_intent.package_name = 'my.project'328 response = parser.parse(request)329 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)330 self.assertEmpty(response.error_message)331 adb.execute_command.assert_called_once_with([332 'shell', 'am', 'start', '-a', 'BEST_ACTION', '-d', '{"some": 123}',333 'my.project'334 ],335 timeout=None)336 def test_uninstall_package_empty_package_name(self):337 adb = mock.create_autospec(adb_controller.AdbController)338 parser = adb_call_parser.AdbCallParser(339 adb, tmp_dir=absltest.get_default_test_tmpdir())340 request = adb_pb2.AdbRequest()341 request.uninstall_package.package_name = ''342 response = parser.parse(request)343 self.assertEqual(response.status,344 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)345 self.assertNotEmpty(response.error_message)346 def test_uninstall_package_successful(self):347 adb = mock.create_autospec(adb_controller.AdbController)348 adb.execute_command.return_value = b'package:my.package'349 parser = adb_call_parser.AdbCallParser(350 adb, tmp_dir=absltest.get_default_test_tmpdir())351 request = adb_pb2.AdbRequest()352 request.uninstall_package.package_name = 'my.package'353 response = parser.parse(request)354 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)355 self.assertEmpty(response.error_message)356 def test_get_current_activity_no_visible_task(self):357 adb = mock.create_autospec(adb_controller.AdbController)358 adb.execute_command.return_value = None359 parser = adb_call_parser.AdbCallParser(360 adb, tmp_dir=absltest.get_default_test_tmpdir())361 request = adb_pb2.AdbRequest(362 get_current_activity=adb_pb2.AdbRequest.GetCurrentActivity())363 response = parser.parse(request)364 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.INTERNAL_ERROR)365 self.assertNotEmpty(response.error_message)366 adb.execute_command.assert_has_calls([367 mock.call(368 ['shell', 'am', 'stack', 'list', '|', 'grep', '-E', 'visible=true'],369 None),370 mock.call(['shell', 'am', 'stack', 'list'], None),371 ])372 def test_get_orientation_empty_dumpsys(self):373 adb = mock.create_autospec(adb_controller.AdbController)374 adb.execute_command.return_value = b''375 parser = adb_call_parser.AdbCallParser(376 adb, tmp_dir=absltest.get_default_test_tmpdir())377 request = adb_pb2.AdbRequest(378 get_orientation=adb_pb2.AdbRequest.GetOrientationRequest())379 response = parser.parse(request)380 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.INTERNAL_ERROR)381 self.assertNotEmpty(response.error_message)382 adb.execute_command.assert_called_once_with(['shell', 'dumpsys', 'input'],383 None)384 def test_get_orientation_invalid_device_no_surface_orientation(self):385 adb = mock.create_autospec(adb_controller.AdbController)386 adb.execute_command.return_value = b' PhysicalWidth: -123px'387 parser = adb_call_parser.AdbCallParser(388 adb, tmp_dir=absltest.get_default_test_tmpdir())389 request = adb_pb2.AdbRequest(390 get_orientation=adb_pb2.AdbRequest.GetOrientationRequest())391 response = parser.parse(request)392 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.INTERNAL_ERROR)393 self.assertNotEmpty(response.error_message)394 adb.execute_command.assert_called_once_with(['shell', 'dumpsys', 'input'],395 None)396 @parameterized.named_parameters(397 ('rotation_0', b'0', 0),398 ('rotation_90', b'1', 1),399 ('rotation_180', b'2', 2),400 ('rotation_270', b'3', 3),401 )402 def test_get_orientation_success(self, orientation, expected_orientation):403 adb = mock.create_autospec(adb_controller.AdbController)404 adb.execute_command.return_value = b"""405 SomeRandomKey: 12345406 SurfaceOrientation: """ + orientation + b"""407 MoreRandomStuff: awesome_value408"""409 parser = adb_call_parser.AdbCallParser(410 adb, tmp_dir=absltest.get_default_test_tmpdir())411 request = adb_pb2.AdbRequest(412 get_orientation=adb_pb2.AdbRequest.GetOrientationRequest())413 response = parser.parse(request)414 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)415 self.assertEmpty(response.error_message)416 self.assertEqual(response.get_orientation.orientation, expected_orientation)417 adb.execute_command.assert_called_once_with(['shell', 'dumpsys', 'input'],418 None)419 def test_get_current_activity_no_matches(self):420 adb = mock.create_autospec(adb_controller.AdbController)421 adb.execute_command.return_value = b'whatever'422 parser = adb_call_parser.AdbCallParser(423 adb, tmp_dir=absltest.get_default_test_tmpdir())424 request = adb_pb2.AdbRequest(425 get_current_activity=adb_pb2.AdbRequest.GetCurrentActivity())426 for platform in ['win32', 'linux']:427 with mock.patch.object(428 sys, 'platform', autospec=True, return_value=platform):429 response = parser.parse(request)430 self.assertEqual(response.status,431 adb_pb2.AdbResponse.Status.INTERNAL_ERROR)432 self.assertNotEmpty(response.error_message)433 adb.execute_command.assert_has_calls([434 mock.call([435 'shell', 'am', 'stack', 'list', '|', 'grep', '-E',436 'visible=true'437 ], None),438 mock.call(['shell', 'am', 'stack', 'list'], None),439 ])440 def test_get_current_activity_successful(self):441 adb = mock.create_autospec(adb_controller.AdbController)442 adb.execute_command.return_value = b'{MyAwesomeActivity}'443 parser = adb_call_parser.AdbCallParser(444 adb, tmp_dir=absltest.get_default_test_tmpdir())445 request = adb_pb2.AdbRequest(446 get_current_activity=adb_pb2.AdbRequest.GetCurrentActivity())447 for platform in ['win32', 'linux']:448 with mock.patch.object(449 sys, 'platform', autospec=True, return_value=platform):450 response = parser.parse(request)451 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)452 self.assertEmpty(response.error_message)453 # `execute_command` will be called once for each platform.454 adb.execute_command.assert_called_with(455 ['shell', 'am', 'stack', 'list', '|', 'grep', '-E', 'visible=true'],456 None)457 self.assertEqual(response.get_current_activity.full_activity,458 'MyAwesomeActivity')459 def test_push_no_path(self):460 adb = mock.create_autospec(adb_controller.AdbController)461 adb.execute_command.return_value = b'whatever'462 parser = adb_call_parser.AdbCallParser(463 adb, tmp_dir=absltest.get_default_test_tmpdir())464 request = adb_pb2.AdbRequest(465 push=adb_pb2.AdbRequest.Push(content=b'Has content but no path'))466 response = parser.parse(request)467 self.assertEqual(response.status,468 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)469 self.assertNotEmpty(response.error_message)470 adb.execute_command.assert_not_called()471 def test_push_successful(self):472 adb = mock.create_autospec(adb_controller.AdbController)473 adb.execute_command.return_value = b'whatever'474 parser = adb_call_parser.AdbCallParser(475 adb, tmp_dir=absltest.get_default_test_tmpdir())476 request = adb_pb2.AdbRequest(477 push=adb_pb2.AdbRequest.Push(478 content=b'My text.', path='/sdcard/my_file.txt'))479 response = parser.parse(request)480 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)481 self.assertEmpty(response.error_message)482 adb.execute_command.assert_called_once()483 args, kwargs = adb.execute_command.call_args484 self.assertLen(args, 1)485 cmd_args = args[0]486 self.assertLen(cmd_args, 3)487 self.assertEqual(cmd_args[0], 'push')488 self.assertEqual(cmd_args[2], '/sdcard/my_file.txt')489 self.assertIn('timeout', kwargs)490 self.assertIsNone(kwargs['timeout'])491 def test_pull_no_path(self):492 adb = mock.create_autospec(adb_controller.AdbController)493 adb.execute_command.return_value = b'whatever'494 parser = adb_call_parser.AdbCallParser(495 adb, tmp_dir=absltest.get_default_test_tmpdir())496 request = adb_pb2.AdbRequest(pull=adb_pb2.AdbRequest.Pull())497 response = parser.parse(request)498 self.assertEqual(response.status,499 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)500 self.assertNotEmpty(response.error_message)501 adb.execute_command.assert_not_called()502 @mock.patch.object(builtins, 'open', autospec=True)503 def test_pull_successful(self, mock_open):504 adb = mock.create_autospec(adb_controller.AdbController)505 adb.execute_command.return_value = b'whatever'506 mock_open.return_value.__enter__ = mock_open507 mock_open.return_value.read.return_value = b'S3cR3t. dO nOt TeLl ANYONE'508 parser = adb_call_parser.AdbCallParser(509 adb, tmp_dir=absltest.get_default_test_tmpdir())510 request = adb_pb2.AdbRequest(511 pull=adb_pb2.AdbRequest.Pull(path='/sdcard/my_file.txt'))512 response = parser.parse(request)513 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)514 self.assertEmpty(response.error_message)515 self.assertEqual(response.pull.content, b'S3cR3t. dO nOt TeLl ANYONE')516 adb.execute_command.assert_called_once()517 args, kwargs = adb.execute_command.call_args518 self.assertLen(args, 1)519 cmd_args = args[0]520 self.assertLen(cmd_args, 3)521 self.assertEqual(cmd_args[0], 'pull')522 self.assertEqual(cmd_args[1], '/sdcard/my_file.txt')523 self.assertIn('timeout', kwargs)524 self.assertIsNone(kwargs['timeout'])525 def test_input_text_no_text(self):526 adb = mock.create_autospec(adb_controller.AdbController)527 adb.execute_command.return_value = b'whatever'528 parser = adb_call_parser.AdbCallParser(529 adb, tmp_dir=absltest.get_default_test_tmpdir())530 request = adb_pb2.AdbRequest(input_text=adb_pb2.AdbRequest.InputText())531 response = parser.parse(request)532 self.assertEqual(response.status,533 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)534 self.assertNotEmpty(response.error_message)535 adb.execute_command.assert_not_called()536 def test_input_text_successful(self):537 adb = mock.create_autospec(adb_controller.AdbController)538 adb.execute_command.return_value = b'whatever'539 parser = adb_call_parser.AdbCallParser(540 adb, tmp_dir=absltest.get_default_test_tmpdir())541 request = adb_pb2.AdbRequest(542 input_text=adb_pb2.AdbRequest.InputText(543 text='The Greatest Text of All Time'))544 response = parser.parse(request)545 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)546 self.assertEmpty(response.error_message)547 adb.execute_command.assert_called_once_with(548 ['shell', 'input', 'text', 'The Greatest Text of All Time'], None)549 @parameterized.named_parameters(550 ('negative_x_and_negative_y',551 adb_pb2.AdbRequest(tap=adb_pb2.AdbRequest.Tap(x=-1, y=-1))),552 ('negative_x',553 adb_pb2.AdbRequest(tap=adb_pb2.AdbRequest.Tap(x=-1, y=123))),554 ('negative_y',555 adb_pb2.AdbRequest(tap=adb_pb2.AdbRequest.Tap(x=456, y=-1))),556 )557 def test_tap_failed(self, request: adb_pb2.AdbRequest):558 adb = mock.create_autospec(adb_controller.AdbController)559 adb.execute_command.return_value = b'whatever'560 parser = adb_call_parser.AdbCallParser(561 adb, tmp_dir=absltest.get_default_test_tmpdir())562 response = parser.parse(request)563 self.assertEqual(response.status,564 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)565 self.assertNotEmpty(response.error_message)566 adb.execute_command.assert_not_called()567 def test_tap_successful(self):568 adb = mock.create_autospec(adb_controller.AdbController)569 adb.execute_command.return_value = b'whatever'570 parser = adb_call_parser.AdbCallParser(571 adb, tmp_dir=absltest.get_default_test_tmpdir())572 request = adb_pb2.AdbRequest(tap=adb_pb2.AdbRequest.Tap(x=135, y=246))573 response = parser.parse(request)574 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)575 self.assertEmpty(response.error_message)576 adb.execute_command.assert_called_once_with(577 ['shell', 'input', 'tap', '135', '246'], None)578 @parameterized.named_parameters(579 ('empty_request', adb_pb2.AdbRequest.SettingsRequest()),580 ('no_namespace',581 adb_pb2.AdbRequest.SettingsRequest(582 get=adb_pb2.AdbRequest.SettingsRequest.Get(key='my_key'))),583 ('get_no_key',584 adb_pb2.AdbRequest.SettingsRequest(585 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,586 get=adb_pb2.AdbRequest.SettingsRequest.Get())),587 ('put_no_key',588 adb_pb2.AdbRequest.SettingsRequest(589 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,590 put=adb_pb2.AdbRequest.SettingsRequest.Put())),591 ('put_no_value',592 adb_pb2.AdbRequest.SettingsRequest(593 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,594 put=adb_pb2.AdbRequest.SettingsRequest.Put(key='another_key'))),595 ('delete_no_key',596 adb_pb2.AdbRequest.SettingsRequest(597 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,598 delete_key=adb_pb2.AdbRequest.SettingsRequest.Delete())),599 ('reset_no_package_name_and_no_mode',600 adb_pb2.AdbRequest.SettingsRequest(601 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,602 reset=adb_pb2.AdbRequest.SettingsRequest.Reset())),603 )604 def test_settings_failures(self, request):605 adb = mock.create_autospec(adb_controller.AdbController)606 adb.execute_command.return_value = b'whatever'607 parser = adb_call_parser.AdbCallParser(608 adb, tmp_dir=absltest.get_default_test_tmpdir())609 request = adb_pb2.AdbRequest(settings=request)610 response = parser.parse(request)611 self.assertEqual(response.status,612 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)613 self.assertNotEmpty(response.error_message)614 adb.execute_command.assert_not_called()615 def test_settings_success_get(self):616 adb = mock.create_autospec(adb_controller.AdbController)617 adb.execute_command.return_value = b'here it is!'618 parser = adb_call_parser.AdbCallParser(619 adb, tmp_dir=absltest.get_default_test_tmpdir())620 request = adb_pb2.AdbRequest.SettingsRequest(621 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,622 get=adb_pb2.AdbRequest.SettingsRequest.Get(key='some_key'))623 request = adb_pb2.AdbRequest(settings=request)624 response = parser.parse(request)625 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)626 self.assertEmpty(response.error_message)627 self.assertEqual(response.settings.output, b'here it is!')628 adb.execute_command.assert_called_once_with(629 ['shell', 'settings', 'get', 'system', 'some_key'], None)630 def test_settings_success_put(self):631 adb = mock.create_autospec(adb_controller.AdbController)632 adb.execute_command.return_value = b'Done for ya!'633 parser = adb_call_parser.AdbCallParser(634 adb, tmp_dir=absltest.get_default_test_tmpdir())635 request = adb_pb2.AdbRequest.SettingsRequest(636 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SECURE,637 put=adb_pb2.AdbRequest.SettingsRequest.Put(key='key1', value='val2'))638 request = adb_pb2.AdbRequest(settings=request)639 response = parser.parse(request)640 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)641 self.assertEmpty(response.error_message)642 self.assertEqual(response.settings.output, b'Done for ya!')643 adb.execute_command.assert_called_once_with(644 ['shell', 'settings', 'put', 'secure', 'key1', 'val2'], None)645 def test_settings_success_delete(self):646 adb = mock.create_autospec(adb_controller.AdbController)647 adb.execute_command.return_value = b'Key deleted.'648 parser = adb_call_parser.AdbCallParser(649 adb, tmp_dir=absltest.get_default_test_tmpdir())650 request = adb_pb2.AdbRequest.SettingsRequest(651 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.GLOBAL,652 delete_key=adb_pb2.AdbRequest.SettingsRequest.Delete(key='useless_key'))653 request = adb_pb2.AdbRequest(settings=request)654 response = parser.parse(request)655 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)656 self.assertEmpty(response.error_message)657 self.assertEqual(response.settings.output, b'Key deleted.')658 adb.execute_command.assert_called_once_with(659 ['shell', 'settings', 'delete', 'global', 'useless_key'], None)660 @parameterized.named_parameters(661 ('mode_untrusted_defaults',662 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.UNTRUSTED_DEFAULTS, '',663 'untrusted_defaults'),664 ('mode_untrusted_clear',665 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.UNTRUSTED_CLEAR, '',666 'untrusted_clear'),667 ('mode_trusted_defaults',668 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.TRUSTED_DEFAULTS, '',669 'trusted_defaults'),670 # If `package_name` is given, it takes precedence over `mode`.671 ('mode_unknown_package_given',672 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.UNKNOWN, 'great.package',673 'great.package'),674 ('mode_untrusted_defaults_package_given',675 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.UNTRUSTED_DEFAULTS,676 'great.package', 'great.package'),677 ('mode_untrusted_clear_package_given',678 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.UNTRUSTED_CLEAR,679 'great.package', 'great.package'),680 ('mode_trusted_defaults_package_given',681 adb_pb2.AdbRequest.SettingsRequest.Reset.Mode.TRUSTED_DEFAULTS,682 'great.package', 'great.package'),683 )684 def test_settings_success_reset(self, mode, package_name, expected_arg):685 adb = mock.create_autospec(adb_controller.AdbController)686 adb.execute_command.return_value = b'Pkg reset.'687 parser = adb_call_parser.AdbCallParser(688 adb, tmp_dir=absltest.get_default_test_tmpdir())689 request = adb_pb2.AdbRequest.SettingsRequest(690 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.GLOBAL,691 reset=adb_pb2.AdbRequest.SettingsRequest.Reset(692 package_name=package_name, mode=mode))693 request = adb_pb2.AdbRequest(settings=request)694 response = parser.parse(request)695 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)696 self.assertEmpty(response.error_message)697 self.assertEqual(response.settings.output, b'Pkg reset.')698 adb.execute_command.assert_called_once_with(699 ['shell', 'settings', 'reset', 'global', expected_arg], None)700 def test_settings_success_list(self):701 adb = mock.create_autospec(adb_controller.AdbController)702 adb.execute_command.return_value = b'volume_ring=5\nvolume_system=7'703 parser = adb_call_parser.AdbCallParser(704 adb, tmp_dir=absltest.get_default_test_tmpdir())705 request = adb_pb2.AdbRequest.SettingsRequest(706 name_space=adb_pb2.AdbRequest.SettingsRequest.Namespace.SYSTEM,707 list=adb_pb2.AdbRequest.SettingsRequest.List())708 request = adb_pb2.AdbRequest(settings=request)709 response = parser.parse(request)710 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)711 self.assertEmpty(response.error_message)712 self.assertEqual(response.settings.output,713 b'volume_ring=5\nvolume_system=7')714 adb.execute_command.assert_called_once_with(715 ['shell', 'settings', 'list', 'system'], None)716 def test_generic_command(self):717 adb = mock.create_autospec(adb_controller.AdbController)718 expected_output = b'generic_output'719 args = ['shell', 'am', 'broadcast', '-n', 'receiver', '-a', 'action']720 adb.execute_command.return_value = expected_output721 parser = adb_call_parser.AdbCallParser(722 adb, tmp_dir=absltest.get_default_test_tmpdir())723 generic_request = adb_pb2.AdbRequest.GenericRequest(args=args)724 request = adb_pb2.AdbRequest(generic=generic_request)725 response = parser.parse(request)726 self.assertEqual(adb_pb2.AdbResponse.Status.OK, response.status)727 self.assertEmpty(response.error_message)728 self.assertEqual(response.generic.output, expected_output)729 adb.execute_command.assert_called_once_with(args, None)730 def test_generic_command_adb_error(self):731 adb = mock.create_autospec(adb_controller.AdbController)732 args = ['shell', 'am', 'broadcast', '-n', 'receiver', '-a', 'action']733 adb.execute_command.side_effect = subprocess.CalledProcessError(734 cmd='cmd', output='adb_error', returncode=-1)735 parser = adb_call_parser.AdbCallParser(736 adb, tmp_dir=absltest.get_default_test_tmpdir())737 generic_request = adb_pb2.AdbRequest.GenericRequest(args=args)738 request = adb_pb2.AdbRequest(generic=generic_request)739 response = parser.parse(request)740 self.assertEqual(adb_pb2.AdbResponse.Status.ADB_ERROR, response.status)741 self.assertEqual('adb_error', response.error_message)742 self.assertEmpty(response.generic.output)743 adb.execute_command.assert_called_once_with(args, None)744 def test_generic_command_timeout(self):745 adb = mock.create_autospec(adb_controller.AdbController)746 args = ['shell', 'am', 'broadcast', '-n', 'receiver', '-a', 'action']747 adb.execute_command.side_effect = subprocess.TimeoutExpired(748 cmd='cmd', timeout=10)749 parser = adb_call_parser.AdbCallParser(750 adb, tmp_dir=absltest.get_default_test_tmpdir())751 generic_request = adb_pb2.AdbRequest.GenericRequest(args=args)752 request = adb_pb2.AdbRequest(generic=generic_request)753 response = parser.parse(request)754 self.assertEqual(adb_pb2.AdbResponse.Status.TIMEOUT, response.status)755 self.assertEqual('Timeout', response.error_message)756 self.assertEmpty(response.generic.output)757 adb.execute_command.assert_called_once_with(args, None)758 @parameterized.named_parameters(759 ('features',760 adb_pb2.AdbRequest(761 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(762 list=adb_pb2.AdbRequest.PackageManagerRequest.List(763 features=adb_pb2.AdbRequest.PackageManagerRequest.List764 .Features())))),765 ('libraries',766 adb_pb2.AdbRequest(767 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(768 list=adb_pb2.AdbRequest.PackageManagerRequest.List(769 libraries=adb_pb2.AdbRequest.PackageManagerRequest.List770 .Libraries())))),771 ('packages',772 adb_pb2.AdbRequest(773 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(774 list=adb_pb2.AdbRequest.PackageManagerRequest.List(775 packages=adb_pb2.AdbRequest.PackageManagerRequest.List776 .Packages())))),777 )778 def test_package_manager_list_bad_output(self, request):779 adb = mock.create_autospec(adb_controller.AdbController)780 adb.execute_command.return_value = b"""Something irrelevant."""781 parser = adb_call_parser.AdbCallParser(782 adb, tmp_dir=absltest.get_default_test_tmpdir())783 response = parser.parse(request)784 response.package_manager.output = b"""Something irrelevant."""785 self.assertEmpty(response.package_manager.list.items)786 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)787 self.assertEmpty(response.error_message)788 adb.execute_command.assert_called_once()789 def test_package_manager_list_features(self):790 adb = mock.create_autospec(adb_controller.AdbController)791 output = b"""792feature:android.hardware.audio.output793feature:android.hardware.bluetooth794feature:android.hardware.camera795feature:android.hardware.fingerprint796feature:android.software.autofill797feature:android.software.backup798feature:android.software.webview799"""800 adb.execute_command.return_value = output801 parser = adb_call_parser.AdbCallParser(802 adb, tmp_dir=absltest.get_default_test_tmpdir())803 request = adb_pb2.AdbRequest(804 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(805 list=adb_pb2.AdbRequest.PackageManagerRequest.List(806 features=adb_pb2.AdbRequest.PackageManagerRequest.List.Features(807 ))))808 response = parser.parse(request)809 self.assertEqual(response.package_manager.output, output)810 self.assertEqual(response.package_manager.list.items, [811 'android.hardware.audio.output',812 'android.hardware.bluetooth',813 'android.hardware.camera',814 'android.hardware.fingerprint',815 'android.software.autofill',816 'android.software.backup',817 'android.software.webview',818 ])819 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)820 self.assertEmpty(response.error_message)821 adb.execute_command.assert_called_once_with(822 ['shell', 'pm', 'list', 'features'], None)823 def test_package_manager_list_libraries(self):824 adb = mock.create_autospec(adb_controller.AdbController)825 output = b"""826library:android.ext.shared827library:android.hidl.base-V1.0-java828library:android.hidl.manager-V1.0-java829library:android.net.ipsec.ike830library:android.test.base831library:android.test.mock832library:android.test.runner833library:androidx.window.sidecar834library:com.android.future.usb.accessory835library:com.android.location.provider836library:com.android.media.remotedisplay837library:com.android.mediadrm.signer838library:com.android.nfc_extras839library:com.google.android.gms840library:com.google.android.trichromelibrary841library:javax.obex842library:org.apache.http.legacy843"""844 adb.execute_command.return_value = output845 parser = adb_call_parser.AdbCallParser(846 adb, tmp_dir=absltest.get_default_test_tmpdir())847 request = adb_pb2.AdbRequest(848 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(849 list=adb_pb2.AdbRequest.PackageManagerRequest.List(850 libraries=adb_pb2.AdbRequest.PackageManagerRequest.List851 .Libraries())))852 response = parser.parse(request)853 self.assertEqual(response.package_manager.output, output)854 self.assertEqual(response.package_manager.list.items, [855 'android.ext.shared',856 'android.hidl.base-V1.0-java',857 'android.hidl.manager-V1.0-java',858 'android.net.ipsec.ike',859 'android.test.base',860 'android.test.mock',861 'android.test.runner',862 'androidx.window.sidecar',863 'com.android.future.usb.accessory',864 'com.android.location.provider',865 'com.android.media.remotedisplay',866 'com.android.mediadrm.signer',867 'com.android.nfc_extras',868 'com.google.android.gms',869 'com.google.android.trichromelibrary',870 'javax.obex',871 'org.apache.http.legacy',872 ])873 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)874 self.assertEmpty(response.error_message)875 adb.execute_command.assert_called_once_with(876 ['shell', 'pm', 'list', 'libraries'], None)877 def test_package_manager_list_packages(self):878 adb = mock.create_autospec(adb_controller.AdbController)879 output = b"""880package:com.android.phone881package:com.awesome.company882package:com.another.great.thingie883"""884 adb.execute_command.return_value = output885 parser = adb_call_parser.AdbCallParser(886 adb, tmp_dir=absltest.get_default_test_tmpdir())887 request = adb_pb2.AdbRequest(888 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(889 list=adb_pb2.AdbRequest.PackageManagerRequest.List(890 packages=adb_pb2.AdbRequest.PackageManagerRequest.List.Packages(891 ))))892 response = parser.parse(request)893 self.assertEqual(response.package_manager.output, output)894 self.assertEqual(response.package_manager.list.items, [895 'com.android.phone',896 'com.awesome.company',897 'com.another.great.thingie',898 ])899 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)900 self.assertEmpty(response.error_message)901 adb.execute_command.assert_called_once_with(902 ['shell', 'pm', 'list', 'packages'], None)903 def test_package_manager_clear_no_package_name(self):904 adb = mock.create_autospec(adb_controller.AdbController)905 adb.execute_command.return_value = b"""Something irrelevant."""906 parser = adb_call_parser.AdbCallParser(907 adb, tmp_dir=absltest.get_default_test_tmpdir())908 request = adb_pb2.AdbRequest(909 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(910 clear=adb_pb2.AdbRequest.PackageManagerRequest.Clear(911 package_name='')))912 response = parser.parse(request)913 self.assertEmpty(response.package_manager.output)914 self.assertEqual(response.status,915 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)916 self.assertNotEmpty(response.error_message)917 adb.execute_command.assert_not_called()918 def test_package_manager_clear_successful_no_user_id(self):919 adb = mock.create_autospec(adb_controller.AdbController)920 adb.execute_command.return_value = b"""Some successful message."""921 parser = adb_call_parser.AdbCallParser(922 adb, tmp_dir=absltest.get_default_test_tmpdir())923 request = adb_pb2.AdbRequest(924 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(925 clear=adb_pb2.AdbRequest.PackageManagerRequest.Clear(926 package_name='my.package')))927 response = parser.parse(request)928 self.assertEqual(response.package_manager.output,929 b"""Some successful message.""")930 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)931 self.assertEmpty(response.error_message)932 adb.execute_command.assert_called_once_with(933 ['shell', 'pm', 'clear', 'my.package'], None)934 def test_package_manager_clear_successful_with_user_id(self):935 adb = mock.create_autospec(adb_controller.AdbController)936 adb.execute_command.return_value = b"""Some successful message."""937 parser = adb_call_parser.AdbCallParser(938 adb, tmp_dir=absltest.get_default_test_tmpdir())939 request = adb_pb2.AdbRequest(940 package_manager=adb_pb2.AdbRequest.PackageManagerRequest(941 clear=adb_pb2.AdbRequest.PackageManagerRequest.Clear(942 package_name='my.package', user_id='mrawesome')))943 response = parser.parse(request)944 self.assertEqual(response.package_manager.output,945 b"""Some successful message.""")946 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)947 self.assertEmpty(response.error_message)948 adb.execute_command.assert_called_once_with(949 ['shell', 'pm', 'clear', '-f', 'mrawesome', 'my.package'], None)950 def test_dumpsys_empty_request(self):951 """An empty `DumpsysRequest` is a valid request."""952 adb = mock.create_autospec(adb_controller.AdbController)953 adb.execute_command.return_value = b'whatever'954 parser = adb_call_parser.AdbCallParser(955 adb, tmp_dir=absltest.get_default_test_tmpdir())956 request = adb_pb2.AdbRequest(dumpsys=adb_pb2.AdbRequest.DumpsysRequest())957 response = parser.parse(request)958 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)959 self.assertEmpty(response.error_message)960 adb.execute_command.assert_called_once_with(['shell', 'dumpsys'],961 timeout=None)962 @parameterized.named_parameters(963 ('negative_timeout_sec',964 adb_pb2.AdbRequest(965 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(timeout_sec=-1))),966 ('negative_timeout_ms',967 adb_pb2.AdbRequest(968 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(timeout_ms=-2))),969 )970 def test_dumpsys_negative_timeouts(self, request):971 """`DumpsysRequest.timeout_{sec, ms}` if passed, should be positive."""972 adb = mock.create_autospec(adb_controller.AdbController)973 parser = adb_call_parser.AdbCallParser(974 adb, tmp_dir=absltest.get_default_test_tmpdir())975 response = parser.parse(request)976 self.assertEqual(response.status,977 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)978 self.assertNotEmpty(response.error_message)979 adb.execute_command.assert_not_called()980 @parameterized.named_parameters(981 ('both_timeouts_zero', 0, 0, ['shell', 'dumpsys']),982 ('sec_takes_precedence_zero', 123, 0, ['shell', 'dumpsys', '-t', '123']),983 ('sec_takes_precedence', 123, 456, ['shell', 'dumpsys', '-t', '123']),984 ('ms_if_no_sec', 0, 456, ['shell', 'dumpsys', '-T', '456']),985 )986 def test_dumpsys_timeout_successful(self, timeout_sec, timeout_ms, expected):987 adb = mock.create_autospec(adb_controller.AdbController)988 adb.execute_command.return_value = b'whatever'989 parser = adb_call_parser.AdbCallParser(990 adb, tmp_dir=absltest.get_default_test_tmpdir())991 request = adb_pb2.AdbRequest(992 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(993 timeout_sec=timeout_sec, timeout_ms=timeout_ms))994 response = parser.parse(request)995 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)996 self.assertEmpty(response.error_message)997 adb.execute_command.assert_called_once_with(expected, timeout=None)998 @parameterized.named_parameters(999 ('priority_undefined',1000 adb_pb2.AdbRequest.DumpsysRequest.PriorityLevel.UNSET,1001 ['shell', 'dumpsys']),1002 ('priority_normal',1003 adb_pb2.AdbRequest.DumpsysRequest.PriorityLevel.NORMAL,1004 ['shell', 'dumpsys', '--priority', 'NORMAL']),1005 ('priority_high', adb_pb2.AdbRequest.DumpsysRequest.PriorityLevel.HIGH,1006 ['shell', 'dumpsys', '--priority', 'HIGH']),1007 ('priority_critical',1008 adb_pb2.AdbRequest.DumpsysRequest.PriorityLevel.CRITICAL,1009 ['shell', 'dumpsys', '--priority', 'CRITICAL']),1010 )1011 def test_dumpsys_priority_timeout_successful(self, priority, expected):1012 adb = mock.create_autospec(adb_controller.AdbController)1013 adb.execute_command.return_value = b'whatever'1014 parser = adb_call_parser.AdbCallParser(1015 adb, tmp_dir=absltest.get_default_test_tmpdir())1016 request = adb_pb2.AdbRequest(1017 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(priority=priority))1018 response = parser.parse(request)1019 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)1020 self.assertEmpty(response.error_message)1021 adb.execute_command.assert_called_once_with(expected, timeout=None)1022 def test_dumpsys_list_only_cannot_be_combined(self):1023 """When passing `-l`, the request cannot contain a few fields."""1024 adb = mock.create_autospec(adb_controller.AdbController)1025 adb.execute_command.return_value = b'whatever'1026 parser = adb_call_parser.AdbCallParser(1027 adb, tmp_dir=absltest.get_default_test_tmpdir())1028 for d in [1029 {1030 'service': 'window'1031 },1032 {1033 'args': ['myoption', 'anotheroption']1034 },1035 {1036 'skip_services': 'usb'1037 },1038 ]:1039 request = adb_pb2.AdbRequest(1040 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(list_only=True, **d))1041 response = parser.parse(request)1042 self.assertEqual(response.status,1043 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)1044 self.assertNotEmpty(response.error_message)1045 adb.execute_command.assert_not_called()1046 def test_dumpsys_list_only_success(self):1047 adb = mock.create_autospec(adb_controller.AdbController)1048 adb.execute_command.return_value = b'whatever'1049 parser = adb_call_parser.AdbCallParser(1050 adb, tmp_dir=absltest.get_default_test_tmpdir())1051 request = adb_pb2.AdbRequest(1052 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(list_only=True))1053 response = parser.parse(request)1054 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)1055 self.assertEmpty(response.error_message)1056 adb.execute_command.assert_called_once_with(['shell', 'dumpsys', '-l'],1057 timeout=None)1058 def test_dumpsys_skip_services_cannot_combine_with_service(self):1059 """When using `DumpsysRequest.skip_service`, it cannot contain `.service`."""1060 adb = mock.create_autospec(adb_controller.AdbController)1061 adb.execute_command.return_value = b'whatever'1062 parser = adb_call_parser.AdbCallParser(1063 adb, tmp_dir=absltest.get_default_test_tmpdir())1064 request = adb_pb2.AdbRequest(1065 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(1066 service='wifi', skip_services=['window', 'usb']))1067 response = parser.parse(request)1068 self.assertEqual(response.status,1069 adb_pb2.AdbResponse.Status.FAILED_PRECONDITION)1070 self.assertNotEmpty(response.error_message)1071 adb.execute_command.assert_not_called()1072 def test_dumpsys_skip_services(self):1073 adb = mock.create_autospec(adb_controller.AdbController)1074 adb.execute_command.return_value = b'whatever'1075 parser = adb_call_parser.AdbCallParser(1076 adb, tmp_dir=absltest.get_default_test_tmpdir())1077 request = adb_pb2.AdbRequest(1078 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(1079 skip_services=['window', 'usb']))1080 response = parser.parse(request)1081 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)1082 self.assertEmpty(response.error_message)1083 adb.execute_command.assert_called_once_with(1084 ['shell', 'dumpsys', '--skip', 'window,usb'], timeout=None)1085 def test_dumpsys_single_service(self):1086 adb = mock.create_autospec(adb_controller.AdbController)1087 adb.execute_command.return_value = b'whatever'1088 parser = adb_call_parser.AdbCallParser(1089 adb, tmp_dir=absltest.get_default_test_tmpdir())1090 request = adb_pb2.AdbRequest(1091 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(service='window'))1092 response = parser.parse(request)1093 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)1094 self.assertEmpty(response.error_message)1095 adb.execute_command.assert_called_once_with(['shell', 'dumpsys', 'window'],1096 timeout=None)1097 def test_dumpsys_single_service_with_args(self):1098 adb = mock.create_autospec(adb_controller.AdbController)1099 adb.execute_command.return_value = b'whatever'1100 parser = adb_call_parser.AdbCallParser(1101 adb, tmp_dir=absltest.get_default_test_tmpdir())1102 request = adb_pb2.AdbRequest(1103 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(1104 service='window', args=['arg1', 'arg2']))1105 response = parser.parse(request)1106 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)1107 self.assertEmpty(response.error_message)1108 adb.execute_command.assert_called_once_with(1109 ['shell', 'dumpsys', 'window', 'arg1', 'arg2'], timeout=None)1110 def test_dumpsys_single_service_with_proto(self):1111 adb = mock.create_autospec(adb_controller.AdbController)1112 adb.execute_command.return_value = b'some binary output'1113 parser = adb_call_parser.AdbCallParser(1114 adb, tmp_dir=absltest.get_default_test_tmpdir())1115 request = adb_pb2.AdbRequest(1116 dumpsys=adb_pb2.AdbRequest.DumpsysRequest(service='window', proto=True))1117 response = parser.parse(request)1118 self.assertEqual(response.status, adb_pb2.AdbResponse.Status.OK)1119 self.assertEmpty(response.error_message)1120 adb.execute_command.assert_called_once_with(1121 ['shell', 'dumpsys', 'window', '--proto'], timeout=None)1122if __name__ == '__main__':...
test_adb.py
Source:test_adb.py
1#!/usr/bin/env python22"""Simple conformance test for adb.3This script will use the available adb in path and run simple4tests that attempt to touch all accessible attached devices.5"""6import hashlib7import os8import pipes9import random10import re11import shlex12import subprocess13import sys14import tempfile15import unittest16def trace(cmd):17 """Print debug message if tracing enabled."""18 if False:19 print >> sys.stderr, cmd20def call(cmd_str):21 """Run process and return output tuple (stdout, stderr, ret code)."""22 trace(cmd_str)23 process = subprocess.Popen(shlex.split(cmd_str),24 stdout=subprocess.PIPE,25 stderr=subprocess.PIPE)26 stdout, stderr = process.communicate()27 return stdout, stderr, process.returncode28def call_combined(cmd_str):29 """Run process and return output tuple (stdout+stderr, ret code)."""30 trace(cmd_str)31 process = subprocess.Popen(shlex.split(cmd_str),32 stdout=subprocess.PIPE,33 stderr=subprocess.STDOUT)34 stdout, _ = process.communicate()35 return stdout, process.returncode36def call_checked(cmd_str):37 """Run process and get stdout+stderr, raise an exception on trouble."""38 trace(cmd_str)39 return subprocess.check_output(shlex.split(cmd_str),40 stderr=subprocess.STDOUT)41def call_checked_list(cmd_str):42 return call_checked(cmd_str).split('\n')43def call_checked_list_skip(cmd_str):44 out_list = call_checked_list(cmd_str)45 def is_init_line(line):46 if (len(line) >= 3) and (line[0] == "*") and (line[-2] == "*"):47 return True48 else:49 return False50 return [line for line in out_list if not is_init_line(line)]51def get_device_list():52 output = call_checked_list_skip("adb devices")53 dev_list = []54 for line in output[1:]:55 if line.strip() == "":56 continue57 device, _ = line.split()58 dev_list.append(device)59 return dev_list60def get_attached_device_count():61 return len(get_device_list())62def compute_md5(string):63 hsh = hashlib.md5()64 hsh.update(string)65 return hsh.hexdigest()66class HostFile(object):67 def __init__(self, handle, md5):68 self.handle = handle69 self.md5 = md570 self.full_path = handle.name71 self.base_name = os.path.basename(self.full_path)72class DeviceFile(object):73 def __init__(self, md5, full_path):74 self.md5 = md575 self.full_path = full_path76 self.base_name = os.path.basename(self.full_path)77def make_random_host_files(in_dir, num_files, rand_size=True):78 files = {}79 min_size = 1 * (1 << 10)80 max_size = 16 * (1 << 10)81 fixed_size = min_size82 for _ in range(num_files):83 file_handle = tempfile.NamedTemporaryFile(dir=in_dir)84 if rand_size:85 size = random.randrange(min_size, max_size, 1024)86 else:87 size = fixed_size88 rand_str = os.urandom(size)89 file_handle.write(rand_str)90 file_handle.flush()91 md5 = compute_md5(rand_str)92 files[file_handle.name] = HostFile(file_handle, md5)93 return files94def make_random_device_files(adb, in_dir, num_files, rand_size=True):95 files = {}96 min_size = 1 * (1 << 10)97 max_size = 16 * (1 << 10)98 fixed_size = min_size99 for i in range(num_files):100 if rand_size:101 size = random.randrange(min_size, max_size, 1024)102 else:103 size = fixed_size104 base_name = "device_tmpfile" + str(i)105 full_path = in_dir + "/" + base_name106 adb.shell("dd if=/dev/urandom of={} bs={} count=1".format(full_path,107 size))108 dev_md5, _ = adb.shell("md5sum {}".format(full_path)).split()109 files[full_path] = DeviceFile(dev_md5, full_path)110 return files111class AdbWrapper(object):112 """Convenience wrapper object for the adb command."""113 def __init__(self, device=None, out_dir=None):114 self.device = device115 self.out_dir = out_dir116 self.adb_cmd = "adb "117 if self.device:118 self.adb_cmd += "-s {} ".format(device)119 if self.out_dir:120 self.adb_cmd += "-p {} ".format(out_dir)121 def shell(self, cmd):122 return call_checked(self.adb_cmd + "shell " + cmd)123 def shell_nocheck(self, cmd):124 return call_combined(self.adb_cmd + "shell " + cmd)125 def install(self, filename):126 return call_checked(self.adb_cmd + "install {}".format(pipes.quote(filename)))127 def push(self, local, remote):128 return call_checked(self.adb_cmd + "push {} {}".format(local, remote))129 def pull(self, remote, local):130 return call_checked(self.adb_cmd + "pull {} {}".format(remote, local))131 def sync(self, directory=""):132 return call_checked(self.adb_cmd + "sync {}".format(directory))133 def forward(self, local, remote):134 return call_checked(self.adb_cmd + "forward {} {}".format(local,135 remote))136 def tcpip(self, port):137 return call_checked(self.adb_cmd + "tcpip {}".format(port))138 def usb(self):139 return call_checked(self.adb_cmd + "usb")140 def root(self):141 return call_checked(self.adb_cmd + "root")142 def unroot(self):143 return call_checked(self.adb_cmd + "unroot")144 def forward_remove(self, local):145 return call_checked(self.adb_cmd + "forward --remove {}".format(local))146 def forward_remove_all(self):147 return call_checked(self.adb_cmd + "forward --remove-all")148 def connect(self, host):149 return call_checked(self.adb_cmd + "connect {}".format(host))150 def disconnect(self, host):151 return call_checked(self.adb_cmd + "disconnect {}".format(host))152 def reverse(self, remote, local):153 return call_checked(self.adb_cmd + "reverse {} {}".format(remote,154 local))155 def reverse_remove_all(self):156 return call_checked(self.adb_cmd + "reverse --remove-all")157 def reverse_remove(self, remote):158 return call_checked(159 self.adb_cmd + "reverse --remove {}".format(remote))160 def wait(self):161 return call_checked(self.adb_cmd + "wait-for-device")162class AdbBasic(unittest.TestCase):163 def test_shell(self):164 """Check that we can at least cat a file."""165 adb = AdbWrapper()166 out = adb.shell("cat /proc/uptime")167 self.assertEqual(len(out.split()), 2)168 self.assertGreater(float(out.split()[0]), 0.0)169 self.assertGreater(float(out.split()[1]), 0.0)170 def test_help(self):171 """Make sure we get _something_ out of help."""172 out = call_checked("adb help")173 self.assertTrue(len(out) > 0)174 def test_version(self):175 """Get a version number out of the output of adb."""176 out = call_checked("adb version").split()177 version_num = False178 for item in out:179 if re.match(r"[\d+\.]*\d", item):180 version_num = True181 self.assertTrue(version_num)182 def _test_root(self):183 adb = AdbWrapper()184 adb.root()185 adb.wait()186 self.assertEqual("root", adb.shell("id -un").strip())187 def _test_unroot(self):188 adb = AdbWrapper()189 adb.unroot()190 adb.wait()191 self.assertEqual("shell", adb.shell("id -un").strip())192 def test_root_unroot(self):193 """Make sure that adb root and adb unroot work, using id(1)."""194 adb = AdbWrapper()195 original_user = adb.shell("id -un").strip()196 try:197 if original_user == "root":198 self._test_unroot()199 self._test_root()200 elif original_user == "shell":201 self._test_root()202 self._test_unroot()203 finally:204 if original_user == "root":205 adb.root()206 else:207 adb.unroot()208 adb.wait()209 def test_argument_escaping(self):210 """Make sure that argument escaping is somewhat sane."""211 adb = AdbWrapper()212 # http://b/19734868213 # Note that this actually matches ssh(1)'s behavior --- it's214 # converted to "sh -c echo hello; echo world" which sh interprets215 # as "sh -c echo" (with an argument to that shell of "hello"),216 # and then "echo world" back in the first shell.217 result = adb.shell("sh -c 'echo hello; echo world'").splitlines()218 self.assertEqual(["", "world"], result)219 # If you really wanted "hello" and "world", here's what you'd do:220 result = adb.shell("echo hello\;echo world").splitlines()221 self.assertEqual(["hello", "world"], result)222 # http://b/15479704223 self.assertEqual('t', adb.shell("'true && echo t'").strip())224 self.assertEqual('t', adb.shell("sh -c 'true && echo t'").strip())225 # http://b/20564385226 self.assertEqual('t', adb.shell("FOO=a BAR=b echo t").strip())227 self.assertEqual('123Linux', adb.shell("echo -n 123\;uname").strip())228 def test_install_argument_escaping(self):229 """Make sure that install argument escaping works."""230 adb = AdbWrapper()231 # http://b/20323053232 tf = tempfile.NamedTemporaryFile("w", suffix="-text;ls;1.apk")233 self.assertIn("-text;ls;1.apk", adb.install(tf.name))234 # http://b/3090932235 tf = tempfile.NamedTemporaryFile("w", suffix="-Live Hold'em.apk")236 self.assertIn("-Live Hold'em.apk", adb.install(tf.name))237class AdbFile(unittest.TestCase):238 SCRATCH_DIR = "/data/local/tmp"239 DEVICE_TEMP_FILE = SCRATCH_DIR + "/adb_test_file"240 DEVICE_TEMP_DIR = SCRATCH_DIR + "/adb_test_dir"241 def test_push(self):242 """Push a randomly generated file to specified device."""243 kbytes = 512244 adb = AdbWrapper()245 with tempfile.NamedTemporaryFile(mode="w") as tmp:246 rand_str = os.urandom(1024 * kbytes)247 tmp.write(rand_str)248 tmp.flush()249 host_md5 = compute_md5(rand_str)250 adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_FILE))251 try:252 adb.push(local=tmp.name, remote=AdbFile.DEVICE_TEMP_FILE)253 dev_md5, _ = adb.shell(254 "md5sum {}".format(AdbFile.DEVICE_TEMP_FILE)).split()255 self.assertEqual(host_md5, dev_md5)256 finally:257 adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))258 # TODO: write push directory test.259 def test_pull(self):260 """Pull a randomly generated file from specified device."""261 kbytes = 512262 adb = AdbWrapper()263 adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_FILE))264 try:265 adb.shell("dd if=/dev/urandom of={} bs=1024 count={}".format(266 AdbFile.DEVICE_TEMP_FILE, kbytes))267 dev_md5, _ = adb.shell(268 "md5sum {}".format(AdbFile.DEVICE_TEMP_FILE)).split()269 with tempfile.NamedTemporaryFile(mode="w") as tmp_write:270 adb.pull(remote=AdbFile.DEVICE_TEMP_FILE, local=tmp_write.name)271 with open(tmp_write.name) as tmp_read:272 host_contents = tmp_read.read()273 host_md5 = compute_md5(host_contents)274 self.assertEqual(dev_md5, host_md5)275 finally:276 adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))277 def test_pull_dir(self):278 """Pull a randomly generated directory of files from the device."""279 adb = AdbWrapper()280 temp_files = {}281 host_dir = None282 try:283 # create temporary host directory284 host_dir = tempfile.mkdtemp()285 # create temporary dir on device286 adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))287 adb.shell("mkdir -p {}".format(AdbFile.DEVICE_TEMP_DIR))288 # populate device dir with random files289 temp_files = make_random_device_files(290 adb, in_dir=AdbFile.DEVICE_TEMP_DIR, num_files=32)291 adb.pull(remote=AdbFile.DEVICE_TEMP_DIR, local=host_dir)292 for device_full_path in temp_files:293 host_path = os.path.join(294 host_dir, temp_files[device_full_path].base_name)295 with open(host_path) as host_file:296 host_md5 = compute_md5(host_file.read())297 self.assertEqual(host_md5,298 temp_files[device_full_path].md5)299 finally:300 for dev_file in temp_files.values():301 host_path = os.path.join(host_dir, dev_file.base_name)302 os.remove(host_path)303 adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))304 if host_dir:305 os.removedirs(host_dir)306 def test_sync(self):307 """Sync a randomly generated directory of files to specified device."""308 try:309 adb = AdbWrapper()310 temp_files = {}311 # create temporary host directory312 base_dir = tempfile.mkdtemp()313 # create mirror device directory hierarchy within base_dir314 full_dir_path = base_dir + AdbFile.DEVICE_TEMP_DIR315 os.makedirs(full_dir_path)316 # create 32 random files within the host mirror317 temp_files = make_random_host_files(in_dir=full_dir_path,318 num_files=32)319 # clean up any trash on the device320 adb = AdbWrapper(out_dir=base_dir)321 adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))322 # issue the sync323 adb.sync("data")324 # confirm that every file on the device mirrors that on the host325 for host_full_path in temp_files.keys():326 device_full_path = os.path.join(327 AdbFile.DEVICE_TEMP_DIR,328 temp_files[host_full_path].base_name)329 dev_md5, _ = adb.shell(330 "md5sum {}".format(device_full_path)).split()331 self.assertEqual(temp_files[host_full_path].md5, dev_md5)332 finally:333 adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))334 if temp_files:335 for tf in temp_files.values():336 tf.handle.close()337 if base_dir:338 os.removedirs(base_dir + AdbFile.DEVICE_TEMP_DIR)339if __name__ == '__main__':340 random.seed(0)341 dev_count = get_attached_device_count()342 if dev_count:343 suite = unittest.TestLoader().loadTestsFromName(__name__)344 unittest.TextTestRunner(verbosity=3).run(suite)345 else:...
setup_step_interpreter_test.py
Source:setup_step_interpreter_test.py
1# coding=utf-82# Copyright 2022 DeepMind Technologies Limited.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Tests for android_env.components.setup_step_interpreter."""16from unittest import mock17from absl.testing import absltest18from android_env.components import adb_call_parser19from android_env.components import errors20from android_env.components import setup_step_interpreter21from android_env.proto import adb_pb222from android_env.proto import task_pb223from google.protobuf import text_format24def _to_proto(proto_class, text):25 proto = proto_class()26 text_format.Parse(text, proto)27 return proto28class SetupStepInterpreterTest(absltest.TestCase):29 def setUp(self):30 super().setUp()31 self._parser = mock.create_autospec(32 adb_call_parser.AdbCallParser, instance=True)33 def test_empty_setup_steps(self):34 """Simple test where nothing should break, and nothing should be done.35 The test simply expects this test to not crash.36 """37 interpreter = setup_step_interpreter.SetupStepInterpreter(38 adb_call_parser=self._parser)39 interpreter.interpret([])40 def test_none_setup_steps(self):41 """Simple test where nothing should break, and nothing should be done.42 The test simply expects this test to not crash.43 """44 interpreter = setup_step_interpreter.SetupStepInterpreter(45 adb_call_parser=self._parser)46 # Empty setup steps should be ignored.47 interpreter.interpret([None])48 def test_invalid_setup_step(self):49 interpreter = setup_step_interpreter.SetupStepInterpreter(50 adb_call_parser=self._parser)51 # Empty setup steps should be ignored.52 with self.assertRaises(AssertionError):53 interpreter.interpret([_to_proto(task_pb2.SetupStep, '')])54 def test_adb_install_apk_filesystem(self):55 self._parser.parse.return_value = adb_pb2.AdbResponse(56 status=adb_pb2.AdbResponse.Status.OK)57 interpreter = setup_step_interpreter.SetupStepInterpreter(58 adb_call_parser=self._parser)59 interpreter.interpret([60 _to_proto(61 task_pb2.SetupStep, """62adb_request: {63 install_apk: {64 filesystem: {65 path: "/my/favorite/dir/my_apk.apk"66 }67 }68}""")69 ])70 self._parser.parse.assert_called_once_with(71 adb_pb2.AdbRequest(72 install_apk=adb_pb2.AdbRequest.InstallApk(73 filesystem=adb_pb2.AdbRequest.InstallApk.Filesystem(74 path='/my/favorite/dir/my_apk.apk'))))75 def test_adb_force_stop(self):76 self._parser.parse.return_value = adb_pb2.AdbResponse(77 status=adb_pb2.AdbResponse.Status.OK)78 interpreter = setup_step_interpreter.SetupStepInterpreter(79 adb_call_parser=self._parser)80 interpreter.interpret([81 _to_proto(82 task_pb2.SetupStep, """83adb_request: { force_stop: { package_name: "my.app.Activity" } }""")84 ])85 self._parser.parse.assert_called_once_with(86 adb_pb2.AdbRequest(87 force_stop=adb_pb2.AdbRequest.ForceStop(88 package_name='my.app.Activity')))89 def test_adb_start_activity(self):90 self._parser.parse.return_value = adb_pb2.AdbResponse(91 status=adb_pb2.AdbResponse.Status.OK)92 interpreter = setup_step_interpreter.SetupStepInterpreter(93 adb_call_parser=self._parser)94 interpreter.interpret([95 _to_proto(96 task_pb2.SetupStep, """97adb_request: {98 start_activity: {99 full_activity: "my.app.Activity"100 extra_args: "arg1"101 extra_args: "arg2"102 }103}""")104 ])105 self._parser.parse.assert_called_once_with(106 adb_pb2.AdbRequest(107 start_activity=adb_pb2.AdbRequest.StartActivity(108 full_activity='my.app.Activity', extra_args=['arg1', 'arg2'])))109 def test_adb_single_tap(self):110 self._parser.parse.return_value = adb_pb2.AdbResponse(111 status=adb_pb2.AdbResponse.Status.OK)112 interpreter = setup_step_interpreter.SetupStepInterpreter(113 adb_call_parser=self._parser)114 interpreter.interpret([115 _to_proto(task_pb2.SetupStep, """116adb_request: {117 tap: {118 x: 321119 y: 654120 }121}""")122 ])123 self._parser.parse.assert_called_once_with(124 adb_pb2.AdbRequest(tap=adb_pb2.AdbRequest.Tap(x=321, y=654)))125 def test_adb_press_button(self):126 self._parser.parse.return_value = adb_pb2.AdbResponse(127 status=adb_pb2.AdbResponse.Status.OK)128 interpreter = setup_step_interpreter.SetupStepInterpreter(129 adb_call_parser=self._parser)130 interpreter.interpret([131 _to_proto(task_pb2.SetupStep,132 """ adb_request: { press_button: { button: HOME } }""")133 ])134 self._parser.parse.assert_called_once_with(135 adb_pb2.AdbRequest(136 press_button=adb_pb2.AdbRequest.PressButton(137 button=adb_pb2.AdbRequest.PressButton.Button.HOME)))138 self._parser.reset_mock()139 interpreter.interpret([140 _to_proto(task_pb2.SetupStep,141 """ adb_request: { press_button: { button: BACK } }""")142 ])143 self._parser.parse.assert_called_once_with(144 adb_pb2.AdbRequest(145 press_button=adb_pb2.AdbRequest.PressButton(146 button=adb_pb2.AdbRequest.PressButton.Button.BACK)))147 def test_adb_start_screen_pinning(self):148 self._parser.parse.return_value = adb_pb2.AdbResponse(149 status=adb_pb2.AdbResponse.Status.OK)150 interpreter = setup_step_interpreter.SetupStepInterpreter(151 adb_call_parser=self._parser)152 interpreter.interpret([153 _to_proto(154 task_pb2.SetupStep, """155adb_request: {156 start_screen_pinning: {157 full_activity: "my.app.HighlanderApp" # "There can be only one".158 }159}""")160 ])161 self._parser.parse.assert_called_once_with(162 adb_pb2.AdbRequest(163 start_screen_pinning=adb_pb2.AdbRequest.StartScreenPinning(164 full_activity='my.app.HighlanderApp')))165 @mock.patch('time.sleep')166 def test_time_sleep(self, mock_sleep):167 interpreter = setup_step_interpreter.SetupStepInterpreter(168 adb_call_parser=self._parser)169 interpreter.interpret(170 [_to_proto(task_pb2.SetupStep, """sleep: { time_sec: 0.875 }""")])171 assert mock_sleep.call_count == 2172 mock_sleep.assert_has_calls([mock.call(0.875), mock.call(0.5)])173 @mock.patch('time.sleep')174 def test_wait_for_app_screen_empty_activity(self, unused_mock_sleep):175 interpreter = setup_step_interpreter.SetupStepInterpreter(176 adb_call_parser=self._parser)177 with self.assertRaises(errors.StepCommandError):178 interpreter.interpret([179 _to_proto(task_pb2.SetupStep,180 """success_condition: {wait_for_app_screen: { }}""")181 ])182 @mock.patch('time.sleep')183 def test_check_install_not_installed(self, unused_mock_sleep):184 self._parser.parse.return_value = adb_pb2.AdbResponse(185 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(186 list=adb_pb2.AdbResponse.PackageManagerResponse.List(items=[187 'com.some.package',188 'not.what.you.are.looking.for',189 ])))190 interpreter = setup_step_interpreter.SetupStepInterpreter(191 adb_call_parser=self._parser)192 with self.assertRaises(errors.StepCommandError):193 interpreter.interpret([194 _to_proto(195 task_pb2.SetupStep, """196success_condition: {197 check_install: {198 package_name: "faz"199 timeout_sec: 0.0001200 }201}202""")203 ])204 def test_check_install_installed(self):205 self._parser.parse.return_value = adb_pb2.AdbResponse(206 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(207 list=adb_pb2.AdbResponse.PackageManagerResponse.List(items=[208 'com.some.package',209 'baz',210 ])))211 interpreter = setup_step_interpreter.SetupStepInterpreter(212 adb_call_parser=self._parser)213 # The test checks that this command raises no AssertionError.214 interpreter.interpret([215 _to_proto(216 task_pb2.SetupStep, """217success_condition: {218 check_install: {219 package_name: "baz"220 timeout_sec: 0.0001221 }222}""")223 ])224 def test_num_retries_failure(self):225 self._parser.parse.side_effect = [226 adb_pb2.AdbResponse(227 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(228 list=adb_pb2.AdbResponse.PackageManagerResponse.List(229 items=[]))),230 ] * 3231 interpreter = setup_step_interpreter.SetupStepInterpreter(232 adb_call_parser=self._parser)233 with self.assertRaises(errors.StepCommandError):234 interpreter.interpret([235 _to_proto(236 task_pb2.SetupStep, """237success_condition: {238 check_install: {239 package_name: "faz"240 timeout_sec: 0.0001241 }242 num_retries: 3243}""")244 ])245 # We retried 3 times after the first call, so we expect 3+1 calls.246 self.assertEqual(self._parser.parse.call_count, 3)247 @mock.patch('time.sleep')248 def test_num_retries_success(self, unused_mock_sleep):249 self._parser.parse.side_effect = [250 adb_pb2.AdbResponse(251 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(252 list=adb_pb2.AdbResponse.PackageManagerResponse.List(253 items=[]))),254 adb_pb2.AdbResponse(255 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(256 list=adb_pb2.AdbResponse.PackageManagerResponse.List(257 items=[]))),258 adb_pb2.AdbResponse(259 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(260 list=adb_pb2.AdbResponse.PackageManagerResponse.List(items=[261 'com.some.package',262 'bar',263 ]))),264 adb_pb2.AdbResponse(265 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(266 list=adb_pb2.AdbResponse.PackageManagerResponse.List(items=[])))267 ]268 interpreter = setup_step_interpreter.SetupStepInterpreter(269 adb_call_parser=self._parser)270 interpreter.interpret([271 _to_proto(272 task_pb2.SetupStep, """273success_condition: {274 check_install: {275 package_name: "bar"276 timeout_sec: 0.0001277 }278 num_retries: 5279}""")280 ])281 # The check should succeed on the third try.282 self.assertEqual(self._parser.parse.call_count, 3)283 def test_retry_step(self):284 self._parser.parse.side_effect = [285 adb_pb2.AdbResponse(286 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(287 list=adb_pb2.AdbResponse.PackageManagerResponse.List(288 items=[]))),289 adb_pb2.AdbResponse(290 package_manager=adb_pb2.AdbResponse.PackageManagerResponse(291 list=adb_pb2.AdbResponse.PackageManagerResponse.List(items=[292 'com.some.package',293 'bar',294 ]))),295 ]296 interpreter = setup_step_interpreter.SetupStepInterpreter(297 adb_call_parser=self._parser)298 interpreter.interpret([299 _to_proto(300 task_pb2.SetupStep, """301success_condition: {302 check_install: {303 package_name: "bar"304 timeout_sec: 0.0001305 }306 num_retries: 2307}""")308 ])309 # We expect the check to fail once and succeed on the second pass.310 self.assertEqual(self._parser.parse.call_count, 2)311if __name__ == '__main__':...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run ATX automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
