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 20 21from acloud import errors 22from acloud.internal.lib import adb_tools 23from acloud.internal.lib import driver_test_lib 24 25 26class AdbToolsTest(driver_test_lib.BaseDriverTest): 27 """Test adb functions.""" 28 DEVICE_ALIVE = ("List of devices attached\n" 29 "127.0.0.1:48451 device product:aosp_cf_x86_phone " 30 "model:Cuttlefish_x86_phone device:vsoc_x86 " 31 "transport_id:98").encode() 32 DEVICE_OFFLINE = ("List of devices attached\n" 33 "127.0.0.1:48451 offline").encode() 34 DEVICE_STATE_ONLY = ("List of devices attached\n" 35 "127.0.0.1:48451\toffline\n" 36 "emulator-5554\tdevice\n").encode() 37 DEVICE_NONE = b"List of devices attached" 38 39 def setUp(self): 40 """Patch the path to adb.""" 41 super().setUp() 42 self.Patch(adb_tools.AdbTools, "_adb_command", "path/adb") 43 44 # pylint: disable=no-member 45 def testGetAdbConnectionStatus(self): 46 """Test get adb connection status.""" 47 fake_adb_port = "48451" 48 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 49 adb_cmd = adb_tools.AdbTools(fake_adb_port) 50 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "device") 51 52 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 53 adb_cmd = adb_tools.AdbTools(fake_adb_port) 54 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "offline") 55 56 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 57 adb_cmd = adb_tools.AdbTools(fake_adb_port) 58 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), None) 59 60 def testGetAdbConnectionStatusFail(self): 61 """Test adb connect status fail.""" 62 fake_adb_port = None 63 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 64 adb_cmd = adb_tools.AdbTools(fake_adb_port) 65 self.assertEqual(adb_cmd.GetAdbConnectionStatus(), None) 66 67 def testGetAdbInformation(self): 68 """Test get adb information.""" 69 fake_adb_port = "48451" 70 dict_device = {'product': 'aosp_cf_x86_phone', 71 'usb': None, 72 'adb_status': 'device', 73 'device': 'vsoc_x86', 74 'model': 'Cuttlefish_x86_phone', 75 'transport_id': '98'} 76 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 77 adb_cmd = adb_tools.AdbTools(fake_adb_port) 78 self.assertEqual(adb_cmd.device_information, dict_device) 79 80 dict_office = {'product': None, 81 'usb': None, 82 'adb_status': 'offline', 83 'device': None, 84 'model': None, 85 'transport_id': None} 86 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 87 adb_cmd = adb_tools.AdbTools(fake_adb_port) 88 self.assertEqual(adb_cmd.device_information, dict_office) 89 90 dict_none = {'product': None, 91 'usb': None, 92 'adb_status': None, 93 'device': None, 94 'model': None, 95 'transport_id': None} 96 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 97 adb_cmd = adb_tools.AdbTools(fake_adb_port) 98 self.assertEqual(adb_cmd.device_information, dict_none) 99 100 # pylint: disable=protected-access 101 def testSetDeviceSerial(self): 102 """Test SetDeviceSerial.""" 103 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 104 adb_port = "6666" 105 adb_tool = adb_tools.AdbTools(adb_port) 106 expected_result = "127.0.0.1:6666" 107 self.assertEqual(adb_tool._device_address, expected_result) 108 109 serial = "0.0.0.0:6520" 110 adb_tool = adb_tools.AdbTools(device_serial=serial) 111 expected_result = "0.0.0.0:6520" 112 self.assertEqual(adb_tool._device_address, expected_result) 113 114 def testGetDeviceSerials(self): 115 """Test parsing the output of adb devices.""" 116 self.Patch(subprocess, "check_output", 117 return_value=self.DEVICE_STATE_ONLY) 118 serials = adb_tools.AdbTools.GetDeviceSerials() 119 self.assertEqual(serials, ["127.0.0.1:48451", "emulator-5554"]) 120 121 # pylint: disable=no-member,protected-access 122 def testConnectAdb(self): 123 """Test connect adb.""" 124 fake_adb_port = "48451" 125 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 126 self.Patch(subprocess, "check_call", return_value=True) 127 adb_cmd = adb_tools.AdbTools(fake_adb_port) 128 adb_cmd.ConnectAdb() 129 self.assertEqual(adb_cmd.IsAdbConnectionAlive(), True) 130 subprocess.check_call.assert_not_called() 131 132 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 133 self.Patch(subprocess, "check_call", return_value=True) 134 subprocess.check_call.call_count = 0 135 adb_cmd = adb_tools.AdbTools(fake_adb_port) 136 adb_cmd.ConnectAdb() 137 self.assertEqual(adb_cmd.IsAdbConnectionAlive(), False) 138 subprocess.check_call.assert_called_with([adb_cmd._adb_command, 139 adb_tools._ADB_CONNECT, 140 adb_cmd._device_serial]) 141 142 # pylint: disable=no-member,protected-access 143 def testDisconnectAdb(self): 144 """Test disconnect adb.""" 145 fake_adb_port = "48451" 146 self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) 147 self.Patch(subprocess, "check_call", return_value=True) 148 adb_cmd = adb_tools.AdbTools(fake_adb_port) 149 150 self.assertEqual(adb_cmd.IsAdbConnected(), True) 151 subprocess.check_call.assert_not_called() 152 153 self.Patch(subprocess, "check_output", side_effect=[self.DEVICE_OFFLINE, 154 self.DEVICE_NONE]) 155 self.Patch(subprocess, "check_call", return_value=True) 156 subprocess.check_call.call_count = 0 157 adb_cmd = adb_tools.AdbTools(fake_adb_port) 158 adb_cmd.DisconnectAdb() 159 self.assertEqual(adb_cmd.IsAdbConnected(), False) 160 subprocess.check_call.assert_called_with([adb_cmd._adb_command, 161 adb_tools._ADB_DISCONNECT, 162 adb_cmd._device_serial]) 163 164 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 165 self.Patch(subprocess, "check_call", return_value=True) 166 subprocess.check_call.call_count = 0 167 adb_cmd = adb_tools.AdbTools(fake_adb_port) 168 adb_cmd.DisconnectAdb() 169 self.assertEqual(adb_cmd.IsAdbConnected(), False) 170 subprocess.check_call.assert_not_called() 171 172 # test raise error if adb still alive after disconnect 173 self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) 174 self.Patch(subprocess, "check_call", return_value=True) 175 subprocess.check_call.call_count = 0 176 adb_cmd = adb_tools.AdbTools(fake_adb_port) 177 with self.assertRaises(errors.AdbDisconnectFailed): 178 adb_cmd.DisconnectAdb() 179 180 def testEmuCommand(self): 181 """Test emu command.""" 182 fake_adb_port = "48451" 183 fake_device_serial = "fake_device_serial" 184 self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) 185 186 mock_popen_obj = mock.Mock(returncode=1) 187 self.Patch(subprocess, "Popen", return_value=mock_popen_obj) 188 189 adb_cmd = adb_tools.AdbTools(adb_port=fake_adb_port, 190 device_serial=fake_device_serial) 191 returncode = adb_cmd.EmuCommand("unit", "test") 192 self.assertEqual(returncode, 1) 193 subprocess.Popen.assert_called_once_with( 194 ["path/adb", "-s", "fake_device_serial", "emu", "unit", "test"], 195 stdin=subprocess.PIPE, 196 stdout=subprocess.PIPE, 197 stderr=subprocess.PIPE) 198 mock_popen_obj.communicate.assert_called_once_with() 199 200 201if __name__ == "__main__": 202 unittest.main() 203