forked from denoland/deno
-
Notifications
You must be signed in to change notification settings - Fork 0
/
util.py
413 lines (339 loc) · 13.1 KB
/
util.py
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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
# Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
import collections
import os
import re
import shutil
import select
import stat
import sys
import subprocess
import tempfile
import time
if os.environ.get("NO_COLOR", None):
RESET = FG_READ = FG_GREEN = ""
else:
RESET = "\x1b[0m"
FG_RED = "\x1b[31m"
FG_GREEN = "\x1b[32m"
executable_suffix = ".exe" if os.name == "nt" else ""
root_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
libdeno_path = os.path.join(root_path, "core", "libdeno")
tests_path = os.path.join(root_path, "tests")
third_party_path = os.path.join(root_path, "third_party")
def make_env(merge_env=None, env=None):
if env is None:
env = os.environ
env = env.copy()
if merge_env is None:
merge_env = {}
for key in merge_env.keys():
env[key] = merge_env[key]
return env
def add_env_path(add, env, key="PATH", prepend=False):
dirs_left = env[key].split(os.pathsep) if key in env else []
dirs_right = add.split(os.pathsep) if isinstance(add, str) else add
if prepend:
dirs_left, dirs_right = dirs_right, dirs_left
for d in dirs_right:
if not d in dirs_left:
dirs_left += [d]
env[key] = os.pathsep.join(dirs_left)
def run(args, quiet=False, cwd=None, env=None, merge_env=None, shell=None):
args[0] = os.path.normpath(args[0])
env = make_env(env=env, merge_env=merge_env)
if shell is None:
# Use the default value for 'shell' parameter.
# - Posix: do not use shell.
# - Windows: use shell; this makes .bat/.cmd files work.
shell = os.name == "nt"
if not quiet:
print " ".join([shell_quote(arg) for arg in args])
rc = subprocess.call(args, cwd=cwd, env=env, shell=shell)
if rc != 0:
sys.exit(rc)
CmdResult = collections.namedtuple('CmdResult', ['out', 'err', 'code'])
def run_output(args,
quiet=False,
cwd=None,
env=None,
merge_env=None,
exit_on_fail=False):
if merge_env is None:
merge_env = {}
args[0] = os.path.normpath(args[0])
if not quiet:
print " ".join(args)
env = make_env(env=env, merge_env=merge_env)
shell = os.name == "nt" # Run through shell to make .bat/.cmd files work.
p = subprocess.Popen(
args,
cwd=cwd,
env=env,
shell=shell,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
try:
out, err = p.communicate()
except subprocess.CalledProcessError as e:
p.kill()
p.wait()
raise e
retcode = p.poll()
if retcode and exit_on_fail:
sys.exit(retcode)
# Ignore Windows CRLF (\r\n).
return CmdResult(
out.replace('\r\n', '\n'), err.replace('\r\n', '\n'), retcode)
def shell_quote_win(arg):
if re.search(r'[\x00-\x20"^%~!@&?*<>|()=]', arg):
# Double all " quote characters.
arg = arg.replace('"', '""')
# Wrap the entire string in " quotes.
arg = '"' + arg + '"'
# Double any N backslashes that are immediately followed by a " quote.
arg = re.sub(r'(\\+)(?=")', r'\1\1', arg)
return arg
def shell_quote(arg):
if os.name == "nt":
return shell_quote_win(arg)
else:
# Python 2 has posix shell quoting built in, albeit in a weird place.
from pipes import quote
return quote(arg)
def symlink(target, name, target_is_dir=False):
if os.name == "nt":
from ctypes import WinDLL, WinError, GetLastError
from ctypes.wintypes import BOOLEAN, DWORD, LPCWSTR
kernel32 = WinDLL('kernel32', use_last_error=False)
CreateSymbolicLinkW = kernel32.CreateSymbolicLinkW
CreateSymbolicLinkW.restype = BOOLEAN
CreateSymbolicLinkW.argtypes = (LPCWSTR, LPCWSTR, DWORD)
# File-type symlinks can only use backslashes as separators.
target = os.path.normpath(target)
# If the symlink points at a directory, it needs to have the appropriate
# flag set, otherwise the link will be created but it won't work.
if target_is_dir:
type_flag = 0x01 # SYMBOLIC_LINK_FLAG_DIRECTORY
else:
type_flag = 0
# Before Windows 10, creating symlinks requires admin privileges.
# As of Win 10, there is a flag that allows anyone to create them.
# Initially, try to use this flag.
unpriv_flag = 0x02 # SYMBOLIC_LINK_FLAG_ALLOW_UNPRIVILEGED_CREATE
r = CreateSymbolicLinkW(name, target, type_flag | unpriv_flag)
# If it failed with ERROR_INVALID_PARAMETER, try again without the
# 'allow unprivileged create' flag.
if not r and GetLastError() == 87: # ERROR_INVALID_PARAMETER
r = CreateSymbolicLinkW(name, target, type_flag)
# Throw if unsuccessful even after the second attempt.
if not r:
raise WinError()
else:
os.symlink(target, name)
def touch(fname):
if os.path.exists(fname):
os.utime(fname, None)
else:
open(fname, 'a').close()
# Recursively list all files in (a subdirectory of) a git worktree.
# * Optionally, glob patterns may be specified to e.g. only list files with a
# certain extension.
# * Untracked files are included, unless they're listed in .gitignore.
# * Directory names themselves are not listed (but the files inside are).
# * Submodules and their contents are ignored entirely.
# * This function fails if the query matches no files.
def git_ls_files(base_dir, patterns=None):
base_dir = os.path.abspath(base_dir)
args = [
"git", "-C", base_dir, "ls-files", "-z", "--exclude-standard",
"--cached", "--modified", "--others"
]
if patterns:
args += ["--"] + patterns
output = subprocess.check_output(args)
files = [
os.path.normpath(os.path.join(base_dir, f)) for f in output.split("\0")
if f != ""
]
if not files:
raise RuntimeError("git_ls_files: no files in '%s'" % base_dir +
(" matching %s" % patterns if patterns else ""))
return files
# The Python equivalent of `rm -rf`.
def rmtree(directory):
# On Windows, shutil.rmtree() won't delete files that have a readonly bit.
# Git creates some files that do. The 'onerror' callback deals with those.
def rm_readonly(func, path, _):
os.chmod(path, stat.S_IWRITE)
func(path)
shutil.rmtree(directory, onerror=rm_readonly)
def build_mode(default="debug"):
if "DENO_BUILD_MODE" in os.environ:
return os.environ["DENO_BUILD_MODE"]
elif "--release" in sys.argv:
return "release"
else:
return default
# E.G. "target/debug"
def build_path():
if "DENO_BUILD_PATH" in os.environ:
return os.environ["DENO_BUILD_PATH"]
else:
return os.path.join(root_path, "target", build_mode())
def parse_exit_code(s):
codes = [int(d or 1) for d in re.findall(r'error(\d*)', s)]
if len(codes) > 1:
assert False, "doesn't support multiple error codes."
elif len(codes) == 1:
return codes[0]
else:
return 0
# Attempts to enable ANSI escape code support.
# Returns True if successful, False if not supported.
def enable_ansi_colors():
if os.name != 'nt':
return True # On non-windows platforms this just works.
return enable_ansi_colors_win10()
# The windows 10 implementation of enable_ansi_colors.
def enable_ansi_colors_win10():
import ctypes
# Function factory for errcheck callbacks that raise WinError on failure.
def raise_if(error_result):
def check(result, _func, args):
if result == error_result:
raise ctypes.WinError(ctypes.get_last_error())
return args
return check
# Windows API types.
from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR, LPVOID
LPDWORD = ctypes.POINTER(DWORD)
# Generic constants.
NULL = ctypes.c_void_p(0).value
INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value
ERROR_INVALID_PARAMETER = 87
# CreateFile flags.
# yapf: disable
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
FILE_SHARE_READ = 0x01
FILE_SHARE_WRITE = 0x02
OPEN_EXISTING = 3
# yapf: enable
# Get/SetConsoleMode flags.
ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x04
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
# HANDLE CreateFileW(...)
CreateFileW = kernel32.CreateFileW
CreateFileW.restype = HANDLE
CreateFileW.errcheck = raise_if(INVALID_HANDLE_VALUE)
# yapf: disable
CreateFileW.argtypes = (LPCWSTR, # lpFileName
DWORD, # dwDesiredAccess
DWORD, # dwShareMode
LPVOID, # lpSecurityAttributes
DWORD, # dwCreationDisposition
DWORD, # dwFlagsAndAttributes
HANDLE) # hTemplateFile
# yapf: enable
# BOOL CloseHandle(HANDLE hObject)
CloseHandle = kernel32.CloseHandle
CloseHandle.restype = BOOL
CloseHandle.errcheck = raise_if(False)
CloseHandle.argtypes = (HANDLE, )
# BOOL GetConsoleMode(HANDLE hConsoleHandle, LPDWORD lpMode)
GetConsoleMode = kernel32.GetConsoleMode
GetConsoleMode.restype = BOOL
GetConsoleMode.errcheck = raise_if(False)
GetConsoleMode.argtypes = (HANDLE, LPDWORD)
# BOOL SetConsoleMode(HANDLE hConsoleHandle, DWORD dwMode)
SetConsoleMode = kernel32.SetConsoleMode
SetConsoleMode.restype = BOOL
SetConsoleMode.errcheck = raise_if(False)
SetConsoleMode.argtypes = (HANDLE, DWORD)
# Open the console output device.
conout = CreateFileW("CONOUT$", GENERIC_READ | GENERIC_WRITE,
FILE_SHARE_READ | FILE_SHARE_WRITE, NULL,
OPEN_EXISTING, 0, 0)
# Get the current mode.
mode = DWORD()
GetConsoleMode(conout, ctypes.byref(mode))
# Try to set the flag that controls ANSI escape code support.
try:
SetConsoleMode(conout, mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING)
except WindowsError as e: # pylint:disable=undefined-variable
if e.winerror == ERROR_INVALID_PARAMETER:
return False # Not supported, likely an older version of Windows.
raise
finally:
CloseHandle(conout)
return True
def extract_number(pattern, string):
matches = re.findall(pattern, string)
if len(matches) != 1:
return None
return int(matches[0])
def extract_max_latency_in_milliseconds(pattern, string):
matches = re.findall(pattern, string)
if len(matches) != 1:
return None
num = float(matches[0][0])
unit = matches[0][1]
if (unit == 'ms'):
return num
elif (unit == 'us'):
return num / 1000
elif (unit == 's'):
return num * 1000
def parse_wrk_output(output):
stats = {}
stats['req_per_sec'] = None
stats['max_latency'] = None
for line in output.split("\n"):
if stats['req_per_sec'] is None:
stats['req_per_sec'] = extract_number(r'Requests/sec:\s+(\d+)',
line)
if stats['max_latency'] is None:
stats['max_latency'] = extract_max_latency_in_milliseconds(
r'\s+99%(?:\s+(\d+.\d+)([a-z]+))', line)
return stats
def platform():
return {"linux2": "linux", "darwin": "mac", "win32": "win"}[sys.platform]
def mkdtemp():
# On Windows, set the base directory that mkdtemp() uses explicitly. If not,
# it'll use the short (8.3) path to the temp dir, which triggers the error
# 'TS5009: Cannot find the common subdirectory path for the input files.'
temp_dir = os.environ["TEMP"] if os.name == 'nt' else None
return tempfile.mkdtemp(dir=temp_dir)
# This function is copied from:
# https://gist.github.com/hayd/4f46a68fc697ba8888a7b517a414583e
# https://stackoverflow.com/q/52954248/1240268
def tty_capture(cmd, bytes_input, timeout=5):
"""Capture the output of cmd with bytes_input to stdin,
with stdin, stdout and stderr as TTYs."""
# pty is not available on windows, so we import it within this function.
import pty
mo, so = pty.openpty() # provide tty to enable line-buffering
me, se = pty.openpty()
mi, si = pty.openpty()
fdmap = {mo: 'stdout', me: 'stderr', mi: 'stdin'}
timeout_exact = time.time() + timeout
p = subprocess.Popen(
cmd, bufsize=1, stdin=si, stdout=so, stderr=se, close_fds=True)
os.write(mi, bytes_input)
select_timeout = .04 #seconds
res = {'stdout': b'', 'stderr': b''}
while True:
ready, _, _ = select.select([mo, me], [], [], select_timeout)
if ready:
for fd in ready:
data = os.read(fd, 512)
if not data:
break
res[fdmap[fd]] += data
elif p.poll() is not None or time.time(
) > timeout_exact: # select timed-out
break # p exited
for fd in [si, so, se, mi, mo, me]:
os.close(fd) # can't do it sooner: it leads to errno.EIO error
p.wait()
return p.returncode, res['stdout'], res['stderr']