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