1# Copyright 2019 - The Android Open Source Project 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); 4# you may not use this file except in compliance with the License. 5# You may obtain a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12# See the License for the specific language governing permissions and 13# limitations under the License. 14"""Tests for AdbTools.""" 15 16import subprocess 17import unittest 18 19from unittest import mock 20from six import b 21 22from acloud import errors 23from acloud.internal.lib import adb_tools 24from acloud.internal.lib import driver_test_lib 25 26 27class AdbToolsTest(driver_test_lib.BaseDriverTest): 28 """Test adb functions.""" 29 DEVICE_ALIVE = b("List of devices attached\n" 30 "127.0.0.1:48451 device product:aosp_cf_x86_phone " 31 "model:Cuttlefish_x86_phone device:vsoc_x86 " 32 "transport_id:98") 33 DEVICE_OFFLINE = b("List of devices attached\n" 34 "127.0.0.1:48451 offline") 35 DEVICE_STATE_ONLY = b("List of devices attached\n" 36 "127.0.0.1:48451\toffline\n" 37 "emulator-5554\tdevice\n") 38 DEVICE_NONE = b("List of devices attached") 39 40 def setUp(self): 41 """Patch the path to adb.""" 42 super(AdbToolsTest, self).setUp() 43 self.Patch(adb_tools.AdbTools, "_adb_command", "path/adb") 44 45 # pylint: disable=no-member 46 def testGetAdbConnectionStatus(self): 47 """Test get adb connection status.""" 48 fake_adb_port = "48451" 49 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 50 adb_cmd = adb_tools.AdbTools(fake_adb_port) 51 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "device") 52 53 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 54 adb_cmd = adb_tools.AdbTools(fake_adb_port) 55 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "offline") 56 57 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 58 adb_cmd = adb_tools.AdbTools(fake_adb_port) 59 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), None) 60 61 def testGetAdbConnectionStatusFail(self): 62 """Test adb connect status fail.""" 63 fake_adb_port = None 64 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 65 adb_cmd = adb_tools.AdbTools(fake_adb_port) 66 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), None) 67 68 def testGetAdbInformation(self): 69 """Test get adb information.""" 70 fake_adb_port = "48451" 71 dict_device = {'product': 'aosp_cf_x86_phone', 72 'usb': None, 73 'adb_status': 'device', 74 'device': 'vsoc_x86', 75 'model': 'Cuttlefish_x86_phone', 76 'transport_id': '98'} 77 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 78 adb_cmd = adb_tools.AdbTools(fake_adb_port) 79 self.assertEqual(adb_cmd.device_information, dict_device) 80 81 dict_office = {'product': None, 82 'usb': None, 83 'adb_status': 'offline', 84 'device': None, 85 'model': None, 86 'transport_id': None} 87 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 88 adb_cmd = adb_tools.AdbTools(fake_adb_port) 89 self.assertEqual(adb_cmd.device_information, dict_office) 90 91 dict_none = {'product': None, 92 'usb': None, 93 'adb_status': None, 94 'device': None, 95 'model': None, 96 'transport_id': None} 97 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 98 adb_cmd = adb_tools.AdbTools(fake_adb_port) 99 self.assertEqual(adb_cmd.device_information, dict_none) 100 101 def testGetDeviceSerials(self): 102 """Test parsing the output of adb devices.""" 103 self.Patch(subprocess, "check_output", 104 return_value=self.DEVICE_STATE_ONLY) 105 serials = adb_tools.AdbTools.GetDeviceSerials() 106 self.assertEqual(serials, ["127.0.0.1:48451", "emulator-5554"]) 107 108 # pylint: disable=no-member,protected-access 109 def testConnectAdb(self): 110 """Test connect adb.""" 111 fake_adb_port = "48451" 112 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 113 self.Patch(subprocess, "check_call", return_value=True) 114 adb_cmd = adb_tools.AdbTools(fake_adb_port) 115 adb_cmd.ConnectAdb() 116 self.assertEqual(adb_cmd.IsAdbConnectionAlive(), True) 117 subprocess.check_call.assert_not_called() 118 119 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 120 self.Patch(subprocess, "check_call", return_value=True) 121 subprocess.check_call.call_count = 0 122 adb_cmd = adb_tools.AdbTools(fake_adb_port) 123 adb_cmd.ConnectAdb() 124 self.assertEqual(adb_cmd.IsAdbConnectionAlive(), False) 125 subprocess.check_call.assert_called_with([adb_cmd._adb_command, 126 adb_tools._ADB_CONNECT, 127 adb_cmd._device_serial]) 128 129 # pylint: disable=no-member,protected-access 130 def testDisconnectAdb(self): 131 """Test disconnect adb.""" 132 fake_adb_port = "48451" 133 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 134 self.Patch(subprocess, "check_call", return_value=True) 135 adb_cmd = adb_tools.AdbTools(fake_adb_port) 136 137 self.assertEqual(adb_cmd.IsAdbConnected(), True) 138 subprocess.check_call.assert_not_called() 139 140 self.Patch(subprocess, "check_output", side_effect=[self.DEVICE_OFFLINE, 141 self.DEVICE_NONE]) 142 self.Patch(subprocess, "check_call", return_value=True) 143 subprocess.check_call.call_count = 0 144 adb_cmd = adb_tools.AdbTools(fake_adb_port) 145 adb_cmd.DisconnectAdb() 146 self.assertEqual(adb_cmd.IsAdbConnected(), False) 147 subprocess.check_call.assert_called_with([adb_cmd._adb_command, 148 adb_tools._ADB_DISCONNECT, 149 adb_cmd._device_serial]) 150 151 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 152 self.Patch(subprocess, "check_call", return_value=True) 153 subprocess.check_call.call_count = 0 154 adb_cmd = adb_tools.AdbTools(fake_adb_port) 155 adb_cmd.DisconnectAdb() 156 self.assertEqual(adb_cmd.IsAdbConnected(), False) 157 subprocess.check_call.assert_not_called() 158 159 # test raise error if adb still alive after disconnect 160 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 161 self.Patch(subprocess, "check_call", return_value=True) 162 subprocess.check_call.call_count = 0 163 adb_cmd = adb_tools.AdbTools(fake_adb_port) 164 with self.assertRaises(errors.AdbDisconnectFailed): 165 adb_cmd.DisconnectAdb() 166 167 def testEmuCommand(self): 168 """Test emu command.""" 169 fake_adb_port = "48451" 170 fake_device_serial = "fake_device_serial" 171 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 172 173 mock_popen_obj = mock.Mock(returncode=1) 174 self.Patch(subprocess, "Popen", return_value=mock_popen_obj) 175 176 adb_cmd = adb_tools.AdbTools(adb_port=fake_adb_port, 177 device_serial=fake_device_serial) 178 returncode = adb_cmd.EmuCommand("unit", "test") 179 self.assertEqual(returncode, 1) 180 subprocess.Popen.assert_called_once_with( 181 ["path/adb", "-s", "fake_device_serial", "emu", "unit", "test"], 182 stdin=subprocess.PIPE, 183 stdout=subprocess.PIPE, 184 stderr=subprocess.PIPE) 185 mock_popen_obj.communicate.assert_called_once_with() 186 187 188if __name__ == "__main__": 189 unittest.main() 190