1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
# Copyright 2018 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import os
import platform
import signal
import socket
import subprocess
import sys
import time
import threading
DIR_SOURCE_ROOT = os.path.abspath(
os.path.join(os.path.dirname(__file__), os.pardir, os.pardir))
IMAGES_ROOT = os.path.join(
DIR_SOURCE_ROOT, 'third_party', 'fuchsia-sdk', 'images')
SDK_ROOT = os.path.join(DIR_SOURCE_ROOT, 'third_party', 'fuchsia-sdk', 'sdk')
def EnsurePathExists(path):
"""Checks that the file |path| exists on the filesystem and returns the path
if it does, raising an exception otherwise."""
if not os.path.exists(path):
raise IOError('Missing file: ' + path)
return path
def GetHostOsFromPlatform():
host_platform = sys.platform
if host_platform.startswith('linux'):
return 'linux'
elif host_platform.startswith('darwin'):
return 'mac'
raise Exception('Unsupported host platform: %s' % host_platform)
def GetHostArchFromPlatform():
host_arch = platform.machine()
if host_arch == 'x86_64':
return 'x64'
elif host_arch == 'aarch64':
return 'arm64'
raise Exception('Unsupported host architecture: %s' % host_arch)
def GetHostToolPathFromPlatform(tool):
host_arch = platform.machine()
return os.path.join(SDK_ROOT, 'tools', GetHostArchFromPlatform(), tool)
def GetEmuRootForPlatform(emulator):
return os.path.join(
DIR_SOURCE_ROOT, 'third_party', '{0}-{1}-{2}'.format(
emulator, GetHostOsFromPlatform(), GetHostArchFromPlatform()))
def ConnectPortForwardingTask(target, local_port, remote_port = 0):
"""Establishes a port forwarding SSH task to a localhost TCP endpoint hosted
at port |local_port|. Blocks until port forwarding is established.
Returns the remote port number."""
forwarding_flags = ['-O', 'forward', # Send SSH mux control signal.
'-R', '%d:localhost:%d' % (remote_port, local_port),
'-v', # Get forwarded port info from stderr.
'-NT'] # Don't execute command; don't allocate terminal.
if remote_port != 0:
# Forward to a known remote port.
task = target.RunCommand([], ssh_args=forwarding_flags)
if task.returncode != 0:
raise Exception('Could not establish a port forwarding connection.')
return
task = target.RunCommandPiped([],
ssh_args=forwarding_flags,
stdout=subprocess.PIPE,
stderr=open('/dev/null'))
output = task.stdout.readlines()
task.wait()
if task.returncode != 0:
raise Exception('Got an error code when requesting port forwarding: %d' %
task.returncode)
parsed_port = int(output[0].strip())
logging.debug('Port forwarding established (local=%d, device=%d)' %
(local_port, parsed_port))
return parsed_port
def GetAvailableTcpPort():
"""Finds a (probably) open port by opening and closing a listen socket."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("", 0))
port = sock.getsockname()[1]
sock.close()
return port
def SubprocessCallWithTimeout(command, silent=False, timeout_secs=None):
"""Helper function for running a command.
Args:
command: The command to run.
silent: If true, stdout and stderr of the command will not be printed.
timeout_secs: Maximum amount of time allowed for the command to finish.
Returns:
A tuple of (return code, stdout, stderr) of the command. Raises
an exception if the subprocess times out.
"""
if silent:
devnull = open(os.devnull, 'w')
process = subprocess.Popen(command, stdout=devnull, stderr=devnull)
else:
process = subprocess.Popen(command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
timeout_timer = None
if timeout_secs:
def interrupt_process():
process.send_signal(signal.SIGKILL)
timeout_timer = threading.Timer(timeout_secs, interrupt_process)
# Ensure that keyboard interrupts are handled properly (crbug/1198113).
timeout_timer.daemon = True
timeout_timer.start()
out, err = process.communicate()
if timeout_timer:
timeout_timer.cancel()
if process.returncode == -9:
raise Exception('Timeout when executing \"%s\".' % ' '.join(command))
return process.returncode, out, err
|
