| # 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 |
| import mock |
| from six import b |
| |
| from acloud import errors |
| from acloud.internal.lib import adb_tools |
| from acloud.internal.lib import driver_test_lib |
| from acloud.internal.lib import utils |
| |
| |
| class AdbToolsTest(driver_test_lib.BaseDriverTest): |
| """Test adb functions.""" |
| DEVICE_ALIVE = b("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") |
| DEVICE_OFFLINE = b("List of devices attached\n" |
| "127.0.0.1:48451 offline") |
| DEVICE_NONE = b("List of devices attached") |
| |
| # 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=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(utils, "FindExecutable", return_value="path/adb") |
| 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() |
