from __future__ import absolute_import
import os, signal, subprocess, sys
import re
import platform
import tempfile
import lit.ShUtil as ShUtil
import lit.Test as Test
import lit.util
class InternalShellError(Exception):
def __init__(self, command, message):
self.command = command
self.message = message
kIsWindows = platform.system() == 'Windows'
kUseCloseFDs = not kIsWindows
kAvoidDevNull = kIsWindows
def executeShCmd(cmd, cfg, cwd, results):
if isinstance(cmd, ShUtil.Seq):
if cmd.op == ';':
res = executeShCmd(cmd.lhs, cfg, cwd, results)
return executeShCmd(cmd.rhs, cfg, cwd, results)
if cmd.op == '&':
raise InternalShellError(cmd,"unsupported shell operator: '&'")
if cmd.op == '||':
res = executeShCmd(cmd.lhs, cfg, cwd, results)
if res != 0:
res = executeShCmd(cmd.rhs, cfg, cwd, results)
return res
if cmd.op == '&&':
res = executeShCmd(cmd.lhs, cfg, cwd, results)
if res is None:
return res
if res == 0:
res = executeShCmd(cmd.rhs, cfg, cwd, results)
return res
raise ValueError('Unknown shell command: %r' % cmd.op)
assert isinstance(cmd, ShUtil.Pipeline)
procs = []
input = subprocess.PIPE
stderrTempFiles = []
opened_files = []
named_temp_files = []
for i,j in enumerate(cmd.commands):
redirects = [(0,), (1,), (2,)]
for r in j.redirects:
if r[0] == ('>',2):
redirects[2] = [r[1], 'w', None]
elif r[0] == ('>>',2):
redirects[2] = [r[1], 'a', None]
elif r[0] == ('>&',2) and r[1] in '012':
redirects[2] = redirects[int(r[1])]
elif r[0] == ('>&',) or r[0] == ('&>',):
redirects[1] = redirects[2] = [r[1], 'w', None]
elif r[0] == ('>',):
redirects[1] = [r[1], 'w', None]
elif r[0] == ('>>',):
redirects[1] = [r[1], 'a', None]
elif r[0] == ('<',):
redirects[0] = [r[1], 'r', None]
else:
raise InternalShellError(j,"Unsupported redirect: %r" % (r,))
final_redirects = []
for index,r in enumerate(redirects):
if r == (0,):
result = input
elif r == (1,):
if index == 0:
raise InternalShellError(j,"Unsupported redirect for stdin")
elif index == 1:
result = subprocess.PIPE
else:
result = subprocess.STDOUT
elif r == (2,):
if index != 2:
raise InternalShellError(j,"Unsupported redirect on stdout")
result = subprocess.PIPE
else:
if r[2] is None:
if kAvoidDevNull and r[0] == '/dev/null':
r[2] = tempfile.TemporaryFile(mode=r[1])
else:
r[2] = open(r[0], r[1])
if r[1] == 'a':
r[2].seek(0, 2)
opened_files.append(r[2])
result = r[2]
final_redirects.append(result)
stdin, stdout, stderr = final_redirects
if (stderr == subprocess.STDOUT and stdout != subprocess.PIPE):
stderr = subprocess.PIPE
stderrIsStdout = True
else:
stderrIsStdout = False
if stderr == subprocess.PIPE and j != cmd.commands[-1]:
stderr = tempfile.TemporaryFile(mode='w+b')
stderrTempFiles.append((i, stderr))
args = list(j.args)
executable = lit.util.which(args[0], cfg.environment['PATH'])
if not executable:
raise InternalShellError(j, '%r: command not found' % j.args[0])
if kAvoidDevNull:
for i,arg in enumerate(args):
if arg == "/dev/null":
f = tempfile.NamedTemporaryFile(delete=False)
f.close()
named_temp_files.append(f.name)
args[i] = f.name
procs.append(subprocess.Popen(args, cwd=cwd,
executable = executable,
stdin = stdin,
stdout = stdout,
stderr = stderr,
env = cfg.environment,
close_fds = kUseCloseFDs))
if stdin == subprocess.PIPE:
procs[-1].stdin.close()
procs[-1].stdin = None
if stdout == subprocess.PIPE:
input = procs[-1].stdout
elif stderrIsStdout:
input = procs[-1].stderr
else:
input = subprocess.PIPE
for f in opened_files:
f.close()
procData = [None] * len(procs)
procData[-1] = procs[-1].communicate()
for i in range(len(procs) - 1):
if procs[i].stdout is not None:
out = procs[i].stdout.read()
else:
out = ''
if procs[i].stderr is not None:
err = procs[i].stderr.read()
else:
err = ''
procData[i] = (out,err)
for i,f in stderrTempFiles:
f.seek(0, 0)
procData[i] = (procData[i][0], f.read())
def to_string(bytes):
if isinstance(bytes, str):
return bytes
return bytes.encode('utf-8')
exitCode = None
for i,(out,err) in enumerate(procData):
res = procs[i].wait()
if res == -signal.SIGINT:
raise KeyboardInterrupt
try:
out = to_string(out.decode('utf-8'))
except:
out = str(out)
try:
err = to_string(err.decode('utf-8'))
except:
err = str(err)
results.append((cmd.commands[i], out, err, res))
if cmd.pipe_err:
if exitCode is None:
exitCode = res
elif res < 0:
exitCode = min(exitCode, res)
else:
exitCode = max(exitCode, res)
else:
exitCode = res
for f in named_temp_files:
try:
os.remove(f)
except OSError:
pass
if cmd.negate:
exitCode = not exitCode
return exitCode
def executeScriptInternal(test, litConfig, tmpBase, commands, cwd):
cmds = []
for ln in commands:
try:
cmds.append(ShUtil.ShParser(ln, litConfig.isWindows,
test.config.pipefail).parse())
except:
return lit.Test.Result(Test.FAIL, "shell parser error on: %r" % ln)
cmd = cmds[0]
for c in cmds[1:]:
cmd = ShUtil.Seq(cmd, '&&', c)
results = []
try:
exitCode = executeShCmd(cmd, test.config, cwd, results)
except InternalShellError:
e = sys.exc_info()[1]
exitCode = 127
results.append((e.command, '', e.message, exitCode))
out = err = ''
for i,(cmd, cmd_out,cmd_err,res) in enumerate(results):
out += 'Command %d: %s\n' % (i, ' '.join('"%s"' % s for s in cmd.args))
out += 'Command %d Result: %r\n' % (i, res)
out += 'Command %d Output:\n%s\n\n' % (i, cmd_out)
out += 'Command %d Stderr:\n%s\n\n' % (i, cmd_err)
return out, err, exitCode
def executeScript(test, litConfig, tmpBase, commands, cwd):
bashPath = litConfig.getBashPath();
isWin32CMDEXE = (litConfig.isWindows and not bashPath)
script = tmpBase + '.script'
if isWin32CMDEXE:
script += '.bat'
mode = 'w'
if litConfig.isWindows and not isWin32CMDEXE:
mode += 'b' f = open(script, mode)
if isWin32CMDEXE:
f.write('\nif %ERRORLEVEL% NEQ 0 EXIT\n'.join(commands))
else:
if test.config.pipefail:
f.write('set -o pipefail;')
f.write('{ ' + '; } &&\n{ '.join(commands) + '; }')
f.write('\n')
f.close()
if isWin32CMDEXE:
command = ['cmd','/c', script]
else:
if bashPath:
command = [bashPath, script]
else:
command = ['/bin/sh', script]
if litConfig.useValgrind:
command = litConfig.valgrindArgs + command
return lit.util.executeCommand(command, cwd=cwd,
env=test.config.environment)
def parseIntegratedTestScriptCommands(source_path):
"""
parseIntegratedTestScriptCommands(source_path) -> commands
Parse the commands in an integrated test script file into a list of
(line_number, command_type, line).
"""
def to_bytes(str):
return str.encode('utf-8')
def to_string(bytes):
if isinstance(bytes, str):
return bytes
return to_bytes(bytes)
keywords = ('RUN:', 'XFAIL:', 'REQUIRES:', 'END.')
keywords_re = re.compile(
to_bytes("(%s)(.*)\n" % ("|".join(k for k in keywords),)))
f = open(source_path, 'rb')
try:
data = f.read()
line_number = 1
last_match_position = 0
for match in keywords_re.finditer(data):
match_position = match.start()
line_number += data.count(to_bytes('\n'), last_match_position,
match_position)
last_match_position = match_position
keyword,ln = match.groups()
yield (line_number, to_string(keyword[:-1].decode('utf-8')),
to_string(ln.decode('utf-8')))
finally:
f.close()
def parseIntegratedTestScript(test, normalize_slashes=False,
extra_substitutions=[]):
"""parseIntegratedTestScript - Scan an LLVM/Clang style integrated test
script and extract the lines to 'RUN' as well as 'XFAIL' and 'REQUIRES'
information. The RUN lines also will have variable substitution performed.
"""
sourcepath = test.getSourcePath()
sourcedir = os.path.dirname(sourcepath)
execpath = test.getExecPath()
execdir,execbase = os.path.split(execpath)
tmpDir = os.path.join(execdir, 'Output')
tmpBase = os.path.join(tmpDir, execbase)
if normalize_slashes:
sourcepath = sourcepath.replace('\\', '/')
sourcedir = sourcedir.replace('\\', '/')
tmpDir = tmpDir.replace('\\', '/')
tmpBase = tmpBase.replace('\\', '/')
substitutions = list(extra_substitutions)
substitutions.extend([('%%', '#_MARKER_#')])
substitutions.extend(test.config.substitutions)
substitutions.extend([('%s', sourcepath),
('%S', sourcedir),
('%p', sourcedir),
('%{pathsep}', os.pathsep),
('%t', tmpBase + '.tmp'),
('%T', tmpDir),
('#_MARKER_#', '%')])
substitutions.extend([
('%/s', sourcepath.replace('\\', '/')),
('%/S', sourcedir.replace('\\', '/')),
('%/p', sourcedir.replace('\\', '/')),
('%/t', tmpBase.replace('\\', '/') + '.tmp'),
('%/T', tmpDir.replace('\\', '/')),
])
script = []
requires = []
for line_number, command_type, ln in \
parseIntegratedTestScriptCommands(sourcepath):
if command_type == 'RUN':
ln = ln.rstrip()
ln = re.sub('%\(line\)', str(line_number), ln)
def replace_line_number(match):
if match.group(1) == '+':
return str(line_number + int(match.group(2)))
if match.group(1) == '-':
return str(line_number - int(match.group(2)))
ln = re.sub('%\(line *([\+-]) *(\d+)\)', replace_line_number, ln)
if script and script[-1][-1] == '\\':
script[-1] = script[-1][:-1] + ln
else:
script.append(ln)
elif command_type == 'XFAIL':
test.xfails.extend([s.strip() for s in ln.split(',')])
elif command_type == 'REQUIRES':
requires.extend([s.strip() for s in ln.split(',')])
elif command_type == 'END':
if not ln.strip():
break
else:
raise ValueError("unknown script command type: %r" % (
command_type,))
def processLine(ln):
for a,b in substitutions:
if kIsWindows:
b = b.replace("\\","\\\\")
ln = re.sub(a, b, ln)
return ln.strip()
script = [processLine(ln)
for ln in script]
if not script:
return lit.Test.Result(Test.UNRESOLVED, "Test has no run line!")
if script[-1][-1] == '\\':
return lit.Test.Result(Test.UNRESOLVED,
"Test has unterminated run lines (with '\\')")
missing_required_features = [f for f in requires
if f not in test.config.available_features]
if missing_required_features:
msg = ', '.join(missing_required_features)
return lit.Test.Result(Test.UNSUPPORTED,
"Test requires the following features: %s" % msg)
return script,tmpBase,execdir
def executeShTest(test, litConfig, useExternalSh,
extra_substitutions=[]):
if test.config.unsupported:
return (Test.UNSUPPORTED, 'Test is unsupported')
res = parseIntegratedTestScript(test, useExternalSh, extra_substitutions)
if isinstance(res, lit.Test.Result):
return res
if litConfig.noExecute:
return lit.Test.Result(Test.PASS)
script, tmpBase, execdir = res
lit.util.mkdir_p(os.path.dirname(tmpBase))
if useExternalSh:
res = executeScript(test, litConfig, tmpBase, script, execdir)
else:
res = executeScriptInternal(test, litConfig, tmpBase, script, execdir)
if isinstance(res, lit.Test.Result):
return res
out,err,exitCode = res
if exitCode == 0:
status = Test.PASS
else:
status = Test.FAIL
output = """Script:\n--\n%s\n--\nExit Code: %d\n\n""" % (
'\n'.join(script), exitCode)
if out:
output += """Command Output (stdout):\n--\n%s\n--\n""" % (out,)
if err:
output += """Command Output (stderr):\n--\n%s\n--\n""" % (err,)
return lit.Test.Result(status, output)