1 #!/usr/bin/env python 2 # 3 # Copyright (C) 2015 The Android Open Source Project 4 # 5 # Licensed under the Apache License, Version 2.0 (the "License"); 6 # you may not use this file except in compliance with the License. 7 # You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, 13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 # 17 """Tests for the adb program itself. 18 19 This differs from things in test_device.py in that there is no API for these 20 things. Most of these tests involve specific error messages or the help text. 21 """ 22 from __future__ import print_function 23 24 import contextlib 25 import os 26 import random 27 import socket 28 import struct 29 import subprocess 30 import threading 31 import unittest 32 33 import adb 34 35 36 class NonApiTest(unittest.TestCase): 37 """Tests for ADB that aren't a part of the AndroidDevice API.""" 38 39 def test_help(self): 40 """Make sure we get _something_ out of help.""" 41 out = subprocess.check_output( 42 ['adb', 'help'], stderr=subprocess.STDOUT) 43 self.assertGreater(len(out), 0) 44 45 def test_version(self): 46 """Get a version number out of the output of adb.""" 47 lines = subprocess.check_output(['adb', 'version']).splitlines() 48 version_line = lines[0] 49 self.assertRegexpMatches( 50 version_line, r'^Android Debug Bridge version \d+\.\d+\.\d+$') 51 if len(lines) == 2: 52 # Newer versions of ADB have a second line of output for the 53 # version that includes a specific revision (git SHA). 54 revision_line = lines[1] 55 self.assertRegexpMatches( 56 revision_line, r'^Revision [0-9a-f]{12}-android$') 57 58 def test_tcpip_error_messages(self): 59 p = subprocess.Popen(['adb', 'tcpip'], stdout=subprocess.PIPE, 60 stderr=subprocess.STDOUT) 61 out, _ = p.communicate() 62 self.assertEqual(1, p.returncode) 63 self.assertIn('help message', out) 64 65 p = subprocess.Popen(['adb', 'tcpip', 'foo'], stdout=subprocess.PIPE, 66 stderr=subprocess.STDOUT) 67 out, _ = p.communicate() 68 self.assertEqual(1, p.returncode) 69 self.assertIn('error', out) 70 71 # Helper method that reads a pipe until it is closed, then sets the event. 72 def _read_pipe_and_set_event(self, pipe, event): 73 x = pipe.read() 74 event.set() 75 76 # Test that launch_server() does not let the adb server inherit 77 # stdin/stdout/stderr handles which can cause callers of adb.exe to hang. 78 # This test also runs fine on unix even though the impetus is an issue 79 # unique to Windows. 80 def test_handle_inheritance(self): 81 # This test takes 5 seconds to run on Windows: if there is no adb server 82 # running on the the port used below, adb kill-server tries to make a 83 # TCP connection to a closed port and that takes 1 second on Windows; 84 # adb start-server does the same TCP connection which takes another 85 # second, and it waits 3 seconds after starting the server. 86 87 # Start adb client with redirected stdin/stdout/stderr to check if it 88 # passes those redirections to the adb server that it starts. To do 89 # this, run an instance of the adb server on a non-default port so we 90 # don't conflict with a pre-existing adb server that may already be 91 # setup with adb TCP/emulator connections. If there is a pre-existing 92 # adb server, this also tests whether multiple instances of the adb 93 # server conflict on adb.log. 94 95 port = 5038 96 # Kill any existing server on this non-default port. 97 subprocess.check_output(['adb', '-P', str(port), 'kill-server'], 98 stderr=subprocess.STDOUT) 99 100 try: 101 # Run the adb client and have it start the adb server. 102 p = subprocess.Popen(['adb', '-P', str(port), 'start-server'], 103 stdin=subprocess.PIPE, stdout=subprocess.PIPE, 104 stderr=subprocess.PIPE) 105 106 # Start threads that set events when stdout/stderr are closed. 107 stdout_event = threading.Event() 108 stdout_thread = threading.Thread( 109 target=self._read_pipe_and_set_event, 110 args=(p.stdout, stdout_event)) 111 stdout_thread.daemon = True 112 stdout_thread.start() 113 114 stderr_event = threading.Event() 115 stderr_thread = threading.Thread( 116 target=self._read_pipe_and_set_event, 117 args=(p.stderr, stderr_event)) 118 stderr_thread.daemon = True 119 stderr_thread.start() 120 121 # Wait for the adb client to finish. Once that has occurred, if 122 # stdin/stderr/stdout are still open, it must be open in the adb 123 # server. 124 p.wait() 125 126 # Try to write to stdin which we expect is closed. If it isn't 127 # closed, we should get an IOError. If we don't get an IOError, 128 # stdin must still be open in the adb server. The adb client is 129 # probably letting the adb server inherit stdin which would be 130 # wrong. 131 with self.assertRaises(IOError): 132 p.stdin.write('x') 133 134 # Wait a few seconds for stdout/stderr to be closed (in the success 135 # case, this won't wait at all). If there is a timeout, that means 136 # stdout/stderr were not closed and and they must be open in the adb 137 # server, suggesting that the adb client is letting the adb server 138 # inherit stdout/stderr which would be wrong. 139 self.assertTrue(stdout_event.wait(5), "adb stdout not closed") 140 self.assertTrue(stderr_event.wait(5), "adb stderr not closed") 141 finally: 142 # If we started a server, kill it. 143 subprocess.check_output(['adb', '-P', str(port), 'kill-server'], 144 stderr=subprocess.STDOUT) 145 146 # Use SO_LINGER to cause TCP RST segment to be sent on socket close. 147 def _reset_socket_on_close(self, sock): 148 # The linger structure is two shorts on Windows, but two ints on Unix. 149 linger_format = 'hh' if os.name == 'nt' else 'ii' 150 l_onoff = 1 151 l_linger = 0 152 153 sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 154 struct.pack(linger_format, l_onoff, l_linger)) 155 # Verify that we set the linger structure properly by retrieving it. 156 linger = sock.getsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 16) 157 self.assertEqual((l_onoff, l_linger), 158 struct.unpack_from(linger_format, linger)) 159 160 def test_emu_kill(self): 161 """Ensure that adb emu kill works. 162 163 Bug: https://code.google.com/p/android/issues/detail?id=21021 164 """ 165 port = 12345 166 167 with contextlib.closing( 168 socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as listener: 169 # Use SO_REUSEADDR so subsequent runs of the test can grab the port 170 # even if it is in TIME_WAIT. 171 listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 172 listener.bind(('127.0.0.1', port)) 173 listener.listen(4) 174 175 # Now that listening has started, start adb emu kill, telling it to 176 # connect to our mock emulator. 177 p = subprocess.Popen( 178 ['adb', '-s', 'emulator-' + str(port), 'emu', 'kill'], 179 stderr=subprocess.STDOUT) 180 181 accepted_connection, addr = listener.accept() 182 with contextlib.closing(accepted_connection) as conn: 183 # If WSAECONNABORTED (10053) is raised by any socket calls, 184 # then adb probably isn't reading the data that we sent it. 185 conn.sendall('Android Console: type \'help\' for a list ' + 186 'of commands\r\n') 187 conn.sendall('OK\r\n') 188 189 with contextlib.closing(conn.makefile()) as f: 190 self.assertEqual('kill\n', f.readline()) 191 self.assertEqual('quit\n', f.readline()) 192 193 conn.sendall('OK: killing emulator, bye bye\r\n') 194 195 # Use SO_LINGER to send TCP RST segment to test whether adb 196 # ignores WSAECONNRESET on Windows. This happens with the 197 # real emulator because it just calls exit() without closing 198 # the socket or calling shutdown(SD_SEND). At process 199 # termination, Windows sends a TCP RST segment for every 200 # open socket that shutdown(SD_SEND) wasn't used on. 201 self._reset_socket_on_close(conn) 202 203 # Wait for adb to finish, so we can check return code. 204 p.communicate() 205 206 # If this fails, adb probably isn't ignoring WSAECONNRESET when 207 # reading the response from the adb emu kill command (on Windows). 208 self.assertEqual(0, p.returncode) 209 210 211 def main(): 212 random.seed(0) 213 if len(adb.get_devices()) > 0: 214 suite = unittest.TestLoader().loadTestsFromName(__name__) 215 unittest.TextTestRunner(verbosity=3).run(suite) 216 else: 217 print('Test suite must be run with attached devices') 218 219 220 if __name__ == '__main__': 221 main() 222
