Refactor AdbTools into its own module.
Also fix up device connection status issue.
Bug: 122826931
Test: atest acloud_test &&
acloud create
adb kill-server
acloud list should display "not connected" status
Change-Id: I27b22f5f2368790c22136140432a1ae5b2947e07
diff --git a/internal/lib/adb_tools.py b/internal/lib/adb_tools.py
new file mode 100644
index 0000000..3bd2db5
--- /dev/null
+++ b/internal/lib/adb_tools.py
@@ -0,0 +1,144 @@
+# 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.
+"""A tool that help to run adb to check device status."""
+
+from distutils.spawn import find_executable
+
+import re
+import subprocess
+
+from acloud import errors
+from acloud.internal import constants
+from acloud.internal.lib import utils
+
+_ADB_CONNECT = "connect"
+_ADB_DEVICE = "devices"
+_ADB_DISCONNECT = "disconnect"
+_ADB_STATUS_DEVICE = "device"
+
+
+class AdbTools(object):
+ """Adb tools.
+
+ Args:
+ adb_port: String of adb port number.
+ """
+ def __init__(self, adb_port=None):
+ self._adb_command = ""
+ self._adb_port = adb_port
+ self._device_serial = ""
+ self._SetDeviceSerial()
+ self._CheckAdb()
+
+ def _SetDeviceSerial(self):
+ """Set device serial."""
+ if self._adb_port:
+ self._device_serial = "127.0.0.1:%s" % self._adb_port
+
+ def _CheckAdb(self):
+ """Find adb bin path.
+
+ Raises:
+ errors.NoExecuteCmd: Can't find the execute adb bin.
+ """
+ self._adb_command = find_executable(constants.ADB_BIN)
+ if not self._adb_command:
+ raise errors.NoExecuteCmd("Can't find the adb command.")
+
+ def GetAdbConnectionStatus(self):
+ """Get Adb connect status.
+
+ We determine adb connection in below manner:
+ 1. Check if self._adb_port is null (ssh tunnel is broken).
+ 2. Check adb devices command to get the connection status of the
+ adb devices. When the attached field is device, then device is returned,
+ if it is offline, then offline is returned. If no device is found,
+ the None is returned.
+
+ e.g.
+ Case 1: return device
+ List of devices attached
+ 127.0.0.1:48451 device
+
+ Case 2: return offline
+ List of devices attached
+ 127.0.0.1:48451 offline
+
+ Case 3: return None
+ List of devices attached
+
+ Returns:
+ String, the result of adb connection.
+ """
+ if not self._adb_port:
+ return None
+
+ adb_cmd = [self._adb_command, _ADB_DEVICE]
+ device_info = subprocess.check_output(adb_cmd)
+ for device in device_info.splitlines():
+ match = re.match(r"%s\s(?P<adb_status>.+)" % self._device_serial, device)
+ if match:
+ return match.group("adb_status")
+ return None
+
+ def IsAdbConnectionAlive(self):
+ """Check devices connect alive.
+
+ Returns:
+ Boolean, True if adb status is device. False otherwise.
+ """
+ return self.GetAdbConnectionStatus() == _ADB_STATUS_DEVICE
+
+ def IsAdbConnected(self):
+ """Check devices connected or not.
+
+ If adb connected and the status is device or offline, return True.
+ If there is no any connection, return False.
+
+ Returns:
+ Boolean, True if adb status not none. False otherwise.
+ """
+ return self.GetAdbConnectionStatus() is not None
+
+ def DisconnectAdb(self):
+ """Disconnect adb.
+
+ Only disconnect if the devices shows up in adb devices.
+ """
+ try:
+ if self.IsAdbConnected():
+ adb_disconnect_args = [self._adb_command,
+ _ADB_DISCONNECT,
+ self._device_serial]
+ subprocess.check_call(adb_disconnect_args)
+ except subprocess.CalledProcessError:
+ utils.PrintColorString("Failed to adb disconnect %s" %
+ self._device_serial,
+ utils.TextColors.FAIL)
+
+ def ConnectAdb(self):
+ """Connect adb.
+
+ Only connect if adb connection is not alive.
+ """
+ try:
+ if not self.IsAdbConnectionAlive():
+ adb_connect_args = [self._adb_command,
+ _ADB_CONNECT,
+ self._device_serial]
+ subprocess.check_call(adb_connect_args)
+ except subprocess.CalledProcessError:
+ utils.PrintColorString("Failed to adb connect %s" %
+ self._device_serial,
+ utils.TextColors.FAIL)
diff --git a/internal/lib/adb_tools_test.py b/internal/lib/adb_tools_test.py
new file mode 100644
index 0000000..777354a
--- /dev/null
+++ b/internal/lib/adb_tools_test.py
@@ -0,0 +1,97 @@
+# 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 unittest
+import subprocess
+
+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")
+ DEVICE_OFFLINE = ("List of devices attached\n"
+ "127.0.0.1:48451 offline")
+ DEVICE_NONE = ("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)
+ self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "offline")
+
+ self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE)
+ self.assertEqual(adb_cmd.GetAdbConnectionStatus(), 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", 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.DisconnectAdb()
+ self.assertEqual(adb_cmd.IsAdbConnected(), True)
+ 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()
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/list/instance.py b/list/instance.py
index 51d7817..d726839 100644
--- a/list/instance.py
+++ b/list/instance.py
@@ -38,6 +38,7 @@
from acloud.internal import constants
from acloud.internal.lib import utils
+from acloud.internal.lib.adb_tools import AdbTools
logger = logging.getLogger(__name__)
@@ -259,8 +260,15 @@
def _ProcessGceInstance(self, gce_instance):
"""Parse the required data from gce_instance to local variables.
- We also gather more details on client side including the forwarding adb port
- and vnc port which will be used to determine the status of connection.
+ We also gather more details on client side including the forwarding adb
+ port and vnc port which will be used to determine the status of ssh
+ tunnel connection.
+
+ The status of gce instance will be displayed in _fullname property:
+ - Connected: If gce instance and ssh tunnel and adb connection are all
+ active.
+ - No connected: If ssh tunnel or adb connection is not found.
+ - Terminated: If we can't retrieve the public ip from gce instance.
Args:
gce_instance: dict object queried from gce.
@@ -294,14 +302,15 @@
self._ip = ip
self._adb_port = forwarded_ports.adb_port
self._vnc_port = forwarded_ports.vnc_port
- if self._adb_port:
- self._ssh_tunnel_is_connected = True
+ self._ssh_tunnel_is_connected = self._adb_port is not None
+
+ adb_device = AdbTools(self._adb_port)
+ if adb_device.IsAdbConnected():
self._fullname = (_FULL_NAME_STRING %
{"device_serial": "127.0.0.1:%d" % self._adb_port,
"instance_name": self._name,
"elapsed_time": self._elapsed_time})
else:
- self._ssh_tunnel_is_connected = False
self._fullname = (_FULL_NAME_STRING %
{"device_serial": "not connected",
"instance_name": self._name,
diff --git a/list/instance_test.py b/list/instance_test.py
index 3178adc..e0f4ffd 100644
--- a/list/instance_test.py
+++ b/list/instance_test.py
@@ -14,6 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for instance class."""
+
import collections
import datetime
import subprocess
@@ -27,6 +28,7 @@
from acloud.internal import constants
from acloud.internal.lib import driver_test_lib
+from acloud.internal.lib.adb_tools import AdbTools
from acloud.list import instance
@@ -120,6 +122,7 @@
"GetAdbVncPortFromSSHTunnel",
return_value=forwarded_ports(vnc_port=fake_vnc, adb_port=fake_adb))
self.Patch(instance, "_GetElapsedTime", return_value="fake_time")
+ self.Patch(AdbTools, "IsAdbConnected", return_value=True)
# test ssh_tunnel_is_connected will be true if ssh tunnel connection is found
instance_info = instance.RemoteInstance(gce_instance)
@@ -130,6 +133,14 @@
fake_adb, gce_instance[constants.INS_KEY_NAME], "fake_time")
self.assertEqual(expected_full_name, instance_info.fullname)
+ # test ssh tunnel is connected but adb is disconnected
+ self.Patch(AdbTools, "IsAdbConnected", return_value=False)
+ instance_info = instance.RemoteInstance(gce_instance)
+ self.assertTrue(instance_info.ssh_tunnel_is_connected)
+ expected_full_name = "device serial: not connected (%s) elapsed time: %s" % (
+ instance_info.name, "fake_time")
+ self.assertEqual(expected_full_name, instance_info.fullname)
+
# test ssh_tunnel_is_connected will be false if ssh tunnel connection is not found
self.Patch(
instance.RemoteInstance,
diff --git a/reconnect/reconnect.py b/reconnect/reconnect.py
index df01d85..413b146 100644
--- a/reconnect/reconnect.py
+++ b/reconnect/reconnect.py
@@ -20,23 +20,18 @@
"""
from __future__ import print_function
+
from collections import namedtuple
-from distutils.spawn import find_executable
import getpass
import re
-import subprocess
-from acloud import errors as root_errors
from acloud.delete import delete
from acloud.internal import constants
from acloud.internal.lib import utils
+from acloud.internal.lib.adb_tools import AdbTools
from acloud.list import list as list_instance
from acloud.public import config
-_ADB_CONNECT = "connect"
-_ADB_DEVICE = "devices"
-_ADB_DISCONNECT = "disconnect"
-_ADB_STATUS_DEVICE = "device"
_RE_DISPLAY = re.compile(r"([\d]+)x([\d]+)\s.*")
_VNC_STARTED_PATTERN = "ssvnc vnc://127.0.0.1:%(vnc_port)d"
# TODO(b/122929848): merge all definition of ForwardedPorts into one spot.
@@ -50,117 +45,6 @@
}
-class AdbTools(object):
- """Adb tools.
-
- Args:
- adb_port: String of adb port number.
- """
- def __init__(self, adb_port=None):
- self._adb_command = ""
- self._adb_port = adb_port
- self._device_serial = ""
- self._SetDeviceSerial()
- self._CheckAdb()
-
- def _SetDeviceSerial(self):
- """Set device serial."""
- if self._adb_port:
- self._device_serial = "127.0.0.1:%s" % self._adb_port
-
- def _CheckAdb(self):
- """Find adb bin path.
-
- Raises:
- root_errors.NoExecuteCmd: Can't find the execute adb bin.
- """
- self._adb_command = find_executable(constants.ADB_BIN)
- if not self._adb_command:
- raise root_errors.NoExecuteCmd("Can't find the adb command.")
-
- def GetAdbConnectionStatus(self):
- """Get Adb connect status.
-
- Users uses adb devices command to get the connection status of the
- adb devices. When the attached field is device, then device is returned,
- if it is offline, then offline is returned. If no device is found,
- the None is returned.
-
- e.g.
- Case 1: return device
- List of devices attached
- 127.0.0.1:48451 device
-
- Case 2: return offline
- List of devices attached
- 127.0.0.1:48451 offline
-
- Case 3: return None
- List of devices attached
-
- Returns:
- String, the result of adb connection.
- """
- adb_cmd = [self._adb_command, _ADB_DEVICE]
- device_info = subprocess.check_output(adb_cmd)
- for device in device_info.splitlines():
- match = re.match(r"%s\s(?P<adb_status>.+)" % self._device_serial, device)
- if match:
- return match.group("adb_status")
- return None
-
- def IsAdbConnectionAlive(self):
- """Check devices connect alive.
-
- Returns:
- Boolean, True if adb status is device. False otherwise.
- """
- return self.GetAdbConnectionStatus() == _ADB_STATUS_DEVICE
-
- def IsAdbConnected(self):
- """Check devices connected or not.
-
- If adb connected and the status is device or offline, return True.
- If there is no any connection, return False.
-
- Returns:
- Boolean, True if adb status not none. False otherwise.
- """
- return self.GetAdbConnectionStatus() is not None
-
- def DisconnectAdb(self):
- """Disconnect adb.
-
- Only disconnect if the devices shows up in adb devices.
- """
- try:
- if self.IsAdbConnected():
- adb_disconnect_args = [self._adb_command,
- _ADB_DISCONNECT,
- self._device_serial]
- subprocess.check_call(adb_disconnect_args)
- except subprocess.CalledProcessError:
- utils.PrintColorString("Failed to adb disconnect %s" %
- self._device_serial,
- utils.TextColors.FAIL)
-
- def ConnectAdb(self):
- """Connect adb.
-
- Only connect if adb connection is not alive.
- """
- try:
- if not self.IsAdbConnectionAlive():
- adb_connect_args = [self._adb_command,
- _ADB_CONNECT,
- self._device_serial]
- subprocess.check_call(adb_connect_args)
- except subprocess.CalledProcessError:
- utils.PrintColorString("Failed to adb connect %s" %
- self._device_serial,
- utils.TextColors.FAIL)
-
-
def StartVnc(vnc_port, display):
"""Start vnc connect to AVD.
diff --git a/reconnect/reconnect_test.py b/reconnect/reconnect_test.py
index 58a49b7..b00c131 100644
--- a/reconnect/reconnect_test.py
+++ b/reconnect/reconnect_test.py
@@ -23,6 +23,7 @@
from acloud.internal import constants
from acloud.internal.lib import driver_test_lib
from acloud.internal.lib import utils
+from acloud.internal.lib.adb_tools import AdbTools
from acloud.reconnect import reconnect
ForwardedPorts = collections.namedtuple("ForwardedPorts",
@@ -45,8 +46,8 @@
self.Patch(subprocess, "check_call", return_value=True)
self.Patch(utils, "LaunchVncClient")
self.Patch(utils, "AutoConnect")
- self.Patch(reconnect.AdbTools, "IsAdbConnected", return_value=False)
- self.Patch(reconnect.AdbTools, "IsAdbConnectionAlive", return_value=False)
+ self.Patch(AdbTools, "IsAdbConnected", return_value=False)
+ self.Patch(AdbTools, "IsAdbConnectionAlive", return_value=False)
self.Patch(utils, "IsCommandRunning", return_value=False)
#test ssh tunnel not connected, remote instance.
@@ -145,78 +146,5 @@
utils.LaunchVncClient.assert_called_with(5555, "888", "777")
-class AdbToolsTest(driver_test_lib.BaseDriverTest):
- """Test adb functions."""
- DEVICE_ALIVE = ("List of devices attached\n"
- "127.0.0.1:48451 device")
- DEVICE_OFFLINE = ("List of devices attached\n"
- "127.0.0.1:48451 offline")
- DEVICE_NONE = ("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 = reconnect.AdbTools(fake_adb_port)
- self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "device")
-
- self.Patch(subprocess, "check_output", return_value=self.DEVICE_OFFLINE)
- self.assertEqual(adb_cmd.GetAdbConnectionStatus(), "offline")
-
- self.Patch(subprocess, "check_output", return_value=self.DEVICE_NONE)
- self.assertEqual(adb_cmd.GetAdbConnectionStatus(), 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 = reconnect.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 = reconnect.AdbTools(fake_adb_port)
- adb_cmd.ConnectAdb()
- self.assertEqual(adb_cmd.IsAdbConnectionAlive(), False)
- subprocess.check_call.assert_called_with([adb_cmd._adb_command,
- reconnect._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 = reconnect.AdbTools(fake_adb_port)
-
- self.assertEqual(adb_cmd.IsAdbConnected(), 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 = reconnect.AdbTools(fake_adb_port)
- adb_cmd.DisconnectAdb()
- self.assertEqual(adb_cmd.IsAdbConnected(), True)
- subprocess.check_call.assert_called_with([adb_cmd._adb_command,
- reconnect._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 = reconnect.AdbTools(fake_adb_port)
- adb_cmd.DisconnectAdb()
- self.assertEqual(adb_cmd.IsAdbConnected(), False)
- subprocess.check_call.assert_not_called()
-
-
if __name__ == "__main__":
unittest.main()