# Copyright 2019 - The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tests for AdbTools.""" import subprocess import unittest from unittest import mock from acloud import errors from acloud.internal.lib import adb_tools from acloud.internal.lib import driver_test_lib class AdbToolsTest(driver_test_lib.BaseDriverTest): """Test adb functions.""" DEVICE_ALIVE = ("List of devices attached\n" "127.0.0.1:48451 device product:aosp_cf_x86_phone " "model:Cuttlefish_x86_phone device:vsoc_x86 " "transport_id:98").encode() DEVICE_OFFLINE = ("List of devices attached\n" "127.0.0.1:48451 offline").encode() DEVICE_STATE_ONLY = ("List of devices attached\n" "127.0.0.1:48451\toffline\n" "emulator-5554\tdevice\n").encode() DEVICE_NONE = b"List of devices attached" def setUp(self): """Patch the path to adb.""" super().setUp() self.Patch(adb_tools.AdbTools, "_adb_command", "path/adb") # pylint: disable=no-member def testGetAdbConnectionStatus(self): """Test get adb connection status.""" fake_adb_port = "48451" self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "device") self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "offline") self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.GetAdbConnectionStatus(), None) def testGetAdbConnectionStatusFail(self): """Test adb connect status fail.""" fake_adb_port = None self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.GetAdbConnectionStatus(), None) def testGetAdbInformation(self): """Test get adb information.""" fake_adb_port = "48451" dict_device = {'product': 'aosp_cf_x86_phone', 'usb': None, 'adb_status': 'device', 'device': 'vsoc_x86', 'model': 'Cuttlefish_x86_phone', 'transport_id': '98'} self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.device_information, dict_device) dict_office = {'product': None, 'usb': None, 'adb_status': 'offline', 'device': None, 'model': None, 'transport_id': None} self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.device_information, dict_office) dict_none = {'product': None, 'usb': None, 'adb_status': None, 'device': None, 'model': None, 'transport_id': None} self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.device_information, dict_none) # pylint: disable=protected-access def testSetDeviceSerial(self): """Test SetDeviceSerial.""" self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) adb_port = "6666" adb_tool = adb_tools.AdbTools(adb_port) expected_result = "127.0.0.1:6666" self.assertEqual(adb_tool._device_address, expected_result) serial = "0.0.0.0:6520" adb_tool = adb_tools.AdbTools(device_serial=serial) expected_result = "0.0.0.0:6520" self.assertEqual(adb_tool._device_address, expected_result) def testGetDeviceSerials(self): """Test parsing the output of adb devices.""" self.Patch(subprocess, "check_output", return_value=self.DEVICE_STATE_ONLY) serials = adb_tools.AdbTools.GetDeviceSerials() self.assertEqual(serials, ["127.0.0.1:48451", "emulator-5554"]) # pylint: disable=no-member,protected-access def testConnectAdb(self): """Test connect adb.""" fake_adb_port = "48451" self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) self.Patch(subprocess, "check_call", return_value=True) adb_cmd = adb_tools.AdbTools(fake_adb_port) adb_cmd.ConnectAdb() self.assertEqual(adb_cmd.IsAdbConnectionAlive(), True) subprocess.check_call.assert_not_called() self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) self.Patch(subprocess, "check_call", return_value=True) subprocess.check_call.call_count = 0 adb_cmd = adb_tools.AdbTools(fake_adb_port) adb_cmd.ConnectAdb() self.assertEqual(adb_cmd.IsAdbConnectionAlive(), False) subprocess.check_call.assert_called_with([adb_cmd._adb_command, adb_tools._ADB_CONNECT, adb_cmd._device_serial]) # pylint: disable=no-member,protected-access def testDisconnectAdb(self): """Test disconnect adb.""" fake_adb_port = "48451" self.Patch(subprocess, "check_output", return_value=self.DEVICE_ALIVE) self.Patch(subprocess, "check_call", return_value=True) adb_cmd = adb_tools.AdbTools(fake_adb_port) self.assertEqual(adb_cmd.IsAdbConnected(), True) subprocess.check_call.assert_not_called() self.Patch(subprocess, "check_output", side_effect=[self.DEVICE_OFFLINE, self.DEVICE_NONE]) self.Patch(subprocess, "check_call", return_value=True) subprocess.check_call.call_count = 0 adb_cmd = adb_tools.AdbTools(fake_adb_port) adb_cmd.DisconnectAdb() self.assertEqual(adb_cmd.IsAdbConnected(), False) subprocess.check_call.assert_called_with([adb_cmd._adb_command, adb_tools._ADB_DISCONNECT, adb_cmd._device_serial]) self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) self.Patch(subprocess, "check_call", return_value=True) subprocess.check_call.call_count = 0 adb_cmd = adb_tools.AdbTools(fake_adb_port) adb_cmd.DisconnectAdb() self.assertEqual(adb_cmd.IsAdbConnected(), False) subprocess.check_call.assert_not_called() # test raise error if adb still alive after disconnect self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE) self.Patch(subprocess, "check_call", return_value=True) subprocess.check_call.call_count = 0 adb_cmd = adb_tools.AdbTools(fake_adb_port) with self.assertRaises(errors.AdbDisconnectFailed): adb_cmd.DisconnectAdb() def testEmuCommand(self): """Test emu command.""" fake_adb_port = "48451" fake_device_serial = "fake_device_serial" self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE) mock_popen_obj = mock.Mock(returncode=1) self.Patch(subprocess, "Popen", return_value=mock_popen_obj) adb_cmd = adb_tools.AdbTools(adb_port=fake_adb_port, device_serial=fake_device_serial) returncode = adb_cmd.EmuCommand("unit", "test") self.assertEqual(returncode, 1) subprocess.Popen.assert_called_once_with( ["path/adb", "-s", "fake_device_serial", "emu", "unit", "test"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) mock_popen_obj.communicate.assert_called_once_with() if __name__ == "__main__": unittest.main()