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"""A tool that help to run adb to check device status."""
15
16import re
17import subprocess
18
19from acloud import errors
20from acloud.internal import constants
21from acloud.internal.lib import utils
22
23_ADB_CONNECT = "connect"
24_ADB_DEVICE = "devices"
25_ADB_DISCONNECT = "disconnect"
26_ADB_STATUS_DEVICE = "device"
27_ADB_STATUS_DEVICE_ARGS = "-l"
28_RE_ADB_DEVICE_INFO = (r"%s\s*(?P<adb_status>[\S]+)? ?"
29                       r"(usb:(?P<usb>[\S]+))? ?"
30                       r"(product:(?P<product>[\S]+))? ?"
31                       r"(model:(?P<model>[\S]+))? ?"
32                       r"(device:(?P<device>[\S]+))? ?"
33                       r"(transport_id:(?P<transport_id>[\S]+))? ?")
34_DEVICE_ATTRIBUTES = ["adb_status", "usb", "product", "model", "device", "transport_id"]
35_MAX_RETRIES_ON_WAIT_ADB_GONE = 5
36#KEY_CODE 82 = KEY_MENU
37_UNLOCK_SCREEN_KEYEVENT = ("%(adb_bin)s -s %(device_serial)s "
38                           "shell input keyevent 82")
39_WAIT_ADB_RETRY_BACKOFF_FACTOR = 1.5
40_WAIT_ADB_SLEEP_MULTIPLIER = 2
41
42
43class AdbTools(object):
44    """Adb tools.
45
46    Attributes:
47        _adb_command: String, combine adb commands then execute it.
48        _adb_port: Integer, Specified adb port to establish connection.
49        _device_address: String, the device's host and port for adb to connect
50                         to. For example, adb connect 127.0.0.1:5555.
51        _device_serial: String, adb device's serial number. The value can be
52                        different from _device_address. For example,
53                        adb -s emulator-5554 shell.
54        _device_information: Dict, will be added to adb information include usb,
55                            product model, device and transport_id
56    """
57    def __init__(self, adb_port=None, device_serial=""):
58        """Initialize.
59
60        Args:
61            adb_port: String of adb port number.
62            device_serial: String, adb device's serial number.
63        """
64        self._adb_command = ""
65        self._adb_port = adb_port
66        self._device_address = ""
67        self._device_serial = ""
68        self._SetDeviceSerial(device_serial)
69        self._device_information = {}
70        self._CheckAdb()
71        self._GetAdbInformation()
72
73    def _SetDeviceSerial(self, device_serial):
74        """Set device serial and address.
75
76        Args:
77            device_serial: String, the device's serial number. If this
78                           argument is empty, the serial number is set to the
79                           network address.
80        """
81        self._device_address = ("127.0.0.1:%s" % self._adb_port if
82                                self._adb_port else "")
83        self._device_serial = (device_serial if device_serial else
84                               self._device_address)
85
86    def _CheckAdb(self):
87        """Find adb bin path.
88
89        Raises:
90            errors.NoExecuteCmd: Can't find the execute adb bin.
91        """
92        self._adb_command = utils.FindExecutable(constants.ADB_BIN)
93        if not self._adb_command:
94            raise errors.NoExecuteCmd("Can't find the adb command.")
95
96    def GetAdbConnectionStatus(self):
97        """Get Adb connect status.
98
99        Check if self._adb_port is null (ssh tunnel is broken).
100
101        Returns:
102            String, the result of adb connection.
103        """
104        if not self._adb_port:
105            return None
106
107        return self._device_information["adb_status"]
108
109    def _GetAdbInformation(self):
110        """Get Adb connect information.
111
112        1. Check adb devices command to get the connection information.
113
114        2. Gather information include usb, product model, device and transport_id
115        when the attached field is device.
116
117        e.g.
118            Case 1:
119            List of devices attached
120            127.0.0.1:48451 device product:aosp_cf model:Cuttlefish device:vsoc_x86 transport_id:147
121            _device_information = {"adb_status":"device",
122                                   "usb":None,
123                                   "product":"aosp_cf",
124                                   "model":"Cuttlefish",
125                                   "device":"vsoc_x86",
126                                   "transport_id":"147"}
127
128            Case 2:
129            List of devices attached
130            127.0.0.1:48451 offline
131            _device_information = {"adb_status":"offline",
132                                   "usb":None,
133                                   "product":None,
134                                   "model":None,
135                                   "device":None,
136                                   "transport_id":None}
137
138            Case 3:
139            List of devices attached
140            _device_information = {"adb_status":None,
141                                   "usb":None,
142                                   "product":None,
143                                   "model":None,
144                                   "device":None,
145                                   "transport_id":None}
146        """
147        adb_cmd = [self._adb_command, _ADB_DEVICE, _ADB_STATUS_DEVICE_ARGS]
148        device_info = subprocess.check_output(adb_cmd)
149        self._device_information = {
150            attribute: None for attribute in _DEVICE_ATTRIBUTES}
151
152        for device in device_info.splitlines():
153            match = re.match(_RE_ADB_DEVICE_INFO % self._device_serial, device)
154            if match:
155                self._device_information = {
156                    attribute: match.group(attribute) if match.group(attribute)
157                               else None for attribute in _DEVICE_ATTRIBUTES}
158
159    def IsAdbConnectionAlive(self):
160        """Check devices connect alive.
161
162        Returns:
163            Boolean, True if adb status is device. False otherwise.
164        """
165        return self.GetAdbConnectionStatus() == _ADB_STATUS_DEVICE
166
167    def IsAdbConnected(self):
168        """Check devices connected or not.
169
170        If adb connected and the status is device or offline, return True.
171        If there is no any connection, return False.
172
173        Returns:
174            Boolean, True if adb status not none. False otherwise.
175        """
176        return self.GetAdbConnectionStatus() is not None
177
178    def _DisconnectAndRaiseError(self):
179        """Disconnect adb.
180
181        Disconnect from the device's network address if it shows up in adb
182        devices. For example, adb disconnect 127.0.0.1:5555.
183
184        Raises:
185            errors.WaitForAdbDieError: adb is alive after disconnect adb.
186        """
187        try:
188            if self.IsAdbConnected():
189                adb_disconnect_args = [self._adb_command,
190                                       _ADB_DISCONNECT,
191                                       self._device_address]
192                subprocess.check_call(adb_disconnect_args)
193                # check adb device status
194                self._GetAdbInformation()
195                if self.IsAdbConnected():
196                    raise errors.AdbDisconnectFailed(
197                        "adb disconnect failed, device is still connected and "
198                        "has status: [%s]" % self.GetAdbConnectionStatus())
199
200        except subprocess.CalledProcessError:
201            utils.PrintColorString("Failed to adb disconnect %s" %
202                                   self._device_address,
203                                   utils.TextColors.FAIL)
204
205    def DisconnectAdb(self, retry=False):
206        """Retry to disconnect adb.
207
208        When retry=True, this method will retry to disconnect adb until adb
209        device is completely gone.
210
211        Args:
212            retry: Boolean, True to retry disconnect on error.
213        """
214        retry_count = _MAX_RETRIES_ON_WAIT_ADB_GONE if retry else 0
215        # Wait for adb device is reset and gone.
216        utils.RetryExceptionType(exception_types=errors.AdbDisconnectFailed,
217                                 max_retries=retry_count,
218                                 functor=self._DisconnectAndRaiseError,
219                                 sleep_multiplier=_WAIT_ADB_SLEEP_MULTIPLIER,
220                                 retry_backoff_factor=
221                                 _WAIT_ADB_RETRY_BACKOFF_FACTOR)
222
223    def ConnectAdb(self):
224        """Connect adb.
225
226        Connect adb to the device's network address if the connection is not
227        alive. For example, adb connect 127.0.0.1:5555.
228        """
229        try:
230            if not self.IsAdbConnectionAlive():
231                adb_connect_args = [self._adb_command,
232                                    _ADB_CONNECT,
233                                    self._device_address]
234                subprocess.check_call(adb_connect_args)
235        except subprocess.CalledProcessError:
236            utils.PrintColorString("Failed to adb connect %s" %
237                                   self._device_address,
238                                   utils.TextColors.FAIL)
239
240    def AutoUnlockScreen(self):
241        """Auto unlock screen.
242
243        Auto unlock screen after invoke vnc client.
244        """
245        try:
246            adb_unlock_args = _UNLOCK_SCREEN_KEYEVENT % {
247                "adb_bin": self._adb_command,
248                "device_serial": self._device_serial}
249            subprocess.check_call(adb_unlock_args.split())
250        except subprocess.CalledProcessError:
251            utils.PrintColorString("Failed to unlock screen."
252                                   "(adb_port: %s)" % self._adb_port,
253                                   utils.TextColors.WARNING)
254
255    def EmuCommand(self, *args):
256        """Send an emulator command to the device.
257
258        Args:
259            args: List of strings, the emulator command.
260
261        Returns:
262            Integer, the return code of the adb command.
263            The return code is 0 if adb successfully sends the command to
264            emulator. It is irrelevant to the result of the command.
265        """
266        adb_cmd = [self._adb_command, "-s", self._device_serial, "emu"]
267        adb_cmd.extend(args)
268        proc = subprocess.Popen(adb_cmd, stdin=subprocess.PIPE,
269                                stdout=subprocess.PIPE,
270                                stderr=subprocess.PIPE)
271        proc.communicate()
272        return proc.returncode
273
274    @property
275    def device_information(self):
276        """Return the device information."""
277        return self._device_information
278