def __init__(self, cwd, env):
self.cwd = cwd
- self.env = env
+ self.env = dict(env)
def executeShCmd(cmd, shenv, results):
if isinstance(cmd, ShUtil.Seq):
# Handle shell builtins first.
if cmd.commands[0].args[0] == 'cd':
- # Update the cwd in the environment.
+ if len(cmd.commands) != 1:
+ raise ValueError("'cd' cannot be part of a pipeline")
if len(cmd.commands[0].args) != 2:
- raise ValueError('cd supports only one argument')
+ raise ValueError("'cd' supports only one argument")
newdir = cmd.commands[0].args[1]
+ # Update the cwd in the parent environment.
if os.path.isabs(newdir):
shenv.cwd = newdir
else:
shenv.cwd = os.path.join(shenv.cwd, newdir)
+ # The cd builtin always succeeds. If the directory does not exist, the
+ # following Popen calls will fail instead.
return 0
procs = []
# output. This is null until we have seen some output using
# stderr.
for i,j in enumerate(cmd.commands):
+ # Reference the global environment by default.
+ cmd_shenv = shenv
+ if j.args[0] == 'env':
+ # Create a copy of the global environment and modify it for this one
+ # command. There might be multiple envs in a pipeline:
+ # env FOO=1 llc < %s | env BAR=2 llvm-mc | FileCheck %s
+ cmd_shenv = ShellEnvironment(shenv.cwd, shenv.env)
+ arg_idx = 1
+ for arg_idx, arg in enumerate(j.args[1:]):
+ # Partition the string into KEY=VALUE.
+ key, eq, val = arg.partition('=')
+ # Stop if there was no equals.
+ if eq == '':
+ break
+ cmd_shenv.env[key] = val
+ j.args = j.args[arg_idx+1:]
+
# Apply the redirections, we use (N,) as a sentinel to indicate stdin,
# stdout, stderr for N equal to 0, 1, or 2 respectively. Redirects to or
# from a file are represented with a list [file, mode, file-object]
r[2] = tempfile.TemporaryFile(mode=r[1])
else:
# Make sure relative paths are relative to the cwd.
- redir_filename = os.path.join(shenv.cwd, r[0])
+ redir_filename = os.path.join(cmd_shenv.cwd, r[0])
r[2] = open(redir_filename, r[1])
# Workaround a Win32 and/or subprocess bug when appending.
#
# Resolve the executable path ourselves.
args = list(j.args)
- executable = lit.util.which(args[0], shenv.env['PATH'])
+ executable = None
+ # For paths relative to cwd, use the cwd of the shell environment.
+ if args[0].startswith('.'):
+ exe_in_cwd = os.path.join(cmd_shenv.cwd, args[0])
+ if os.path.isfile(exe_in_cwd):
+ executable = exe_in_cwd
+ if not executable:
+ executable = lit.util.which(args[0], cmd_shenv.env['PATH'])
if not executable:
raise InternalShellError(j, '%r: command not found' % j.args[0])
args[i] = f.name
try:
- procs.append(subprocess.Popen(args, cwd=shenv.cwd,
+ procs.append(subprocess.Popen(args, cwd=cmd_shenv.cwd,
executable = executable,
stdin = stdin,
stdout = stdout,
stderr = stderr,
- env = shenv.env,
+ env = cmd_shenv.env,
close_fds = kUseCloseFDs))
except OSError as e:
- raise InternalShellError(j, 'Could not create process due to {}'.format(e))
+ raise InternalShellError(j, 'Could not create process ({}) due to {}'.format(executable, e))
# Immediately close stdin for any process taking stdin from us.
if stdin == subprocess.PIPE:
return lit.util.executeCommand(command, cwd=cwd,
env=test.config.environment)
-def parseIntegratedTestScriptCommands(source_path):
+def parseIntegratedTestScriptCommands(source_path, keywords):
"""
parseIntegratedTestScriptCommands(source_path) -> commands
# remaining code can work with "strings" agnostic of the executing Python
# version.
- keywords = ['RUN:', 'XFAIL:', 'REQUIRES:', 'UNSUPPORTED:', 'END.']
keywords_re = re.compile(
to_bytes("(%s)(.*)\n" % ("|".join(k for k in keywords),)))
finally:
f.close()
-
-def parseIntegratedTestScript(test, normalize_slashes=False,
- extra_substitutions=[], require_script=True):
- """parseIntegratedTestScript - Scan an LLVM/Clang style integrated test
- script and extract the lines to 'RUN' as well as 'XFAIL' and 'REQUIRES'
- and 'UNSUPPORTED' information. The RUN lines also will have variable
- substitution performed. If 'require_script' is False an empty script may be
- returned. This can be used for test formats where the actual script is
- optional or ignored.
- """
-
- # Get the temporary location, this is always relative to the test suite
- # root, not test source root.
- #
- # FIXME: This should not be here?
- sourcepath = test.getSourcePath()
- sourcedir = os.path.dirname(sourcepath)
+def getTempPaths(test):
+ """Get the temporary location, this is always relative to the test suite
+ root, not test source root."""
execpath = test.getExecPath()
execdir,execbase = os.path.split(execpath)
tmpDir = os.path.join(execdir, 'Output')
tmpBase = os.path.join(tmpDir, execbase)
+ return tmpDir, tmpBase
+
+def getDefaultSubstitutions(test, tmpDir, tmpBase, normalize_slashes=False):
+ sourcepath = test.getSourcePath()
+ sourcedir = os.path.dirname(sourcepath)
# Normalize slashes, if requested.
if normalize_slashes:
tmpBase = tmpBase.replace('\\', '/')
# We use #_MARKER_# to hide %% while we do the other substitutions.
- substitutions = list(extra_substitutions)
+ substitutions = []
substitutions.extend([('%%', '#_MARKER_#')])
substitutions.extend(test.config.substitutions)
substitutions.extend([('%s', sourcepath),
('%/t', tmpBase.replace('\\', '/') + '.tmp'),
('%/T', tmpDir.replace('\\', '/')),
])
+ return substitutions
+def applySubstitutions(script, substitutions):
+ """Apply substitutions to the script. Allow full regular expression syntax.
+ Replace each matching occurrence of regular expression pattern a with
+ substitution b in line ln."""
+ def processLine(ln):
+ # Apply substitutions
+ for a,b in substitutions:
+ if kIsWindows:
+ b = b.replace("\\","\\\\")
+ ln = re.sub(a, b, ln)
+
+ # Strip the trailing newline and any extra whitespace.
+ return ln.strip()
+ # Note Python 3 map() gives an iterator rather than a list so explicitly
+ # convert to list before returning.
+ return list(map(processLine, script))
+
+def parseIntegratedTestScript(test, require_script=True):
+ """parseIntegratedTestScript - Scan an LLVM/Clang style integrated test
+ script and extract the lines to 'RUN' as well as 'XFAIL' and 'REQUIRES'
+ and 'UNSUPPORTED' information. If 'require_script' is False an empty script
+ may be returned. This can be used for test formats where the actual script
+ is optional or ignored.
+ """
# Collect the test lines from the script.
+ sourcepath = test.getSourcePath()
script = []
requires = []
unsupported = []
+ keywords = ['RUN:', 'XFAIL:', 'REQUIRES:', 'UNSUPPORTED:', 'END.']
for line_number, command_type, ln in \
- parseIntegratedTestScriptCommands(sourcepath):
+ parseIntegratedTestScriptCommands(sourcepath, keywords):
if command_type == 'RUN':
# Trim trailing whitespace.
ln = ln.rstrip()
raise ValueError("unknown script command type: %r" % (
command_type,))
- # Apply substitutions to the script. Allow full regular
- # expression syntax. Replace each matching occurrence of regular
- # expression pattern a with substitution b in line ln.
- def processLine(ln):
- # Apply substitutions
- for a,b in substitutions:
- if kIsWindows:
- b = b.replace("\\","\\\\")
- ln = re.sub(a, b, ln)
-
- # Strip the trailing newline and any extra whitespace.
- return ln.strip()
- script = [processLine(ln)
- for ln in script]
-
# Verify the script contains a run line.
if require_script and not script:
return lit.Test.Result(Test.UNRESOLVED, "Test has no run line!")
return lit.Test.Result(Test.UNSUPPORTED,
"Test is unsupported with the following features: %s" % msg)
- return script,tmpBase,execdir
-
-def _runShTest(test, litConfig, useExternalSh,
- script, tmpBase, execdir):
+ unsupported_targets = [f for f in unsupported
+ if f in test.suite.config.target_triple]
+ if unsupported_targets:
+ return lit.Test.Result(Test.UNSUPPORTED,
+ "Test is unsupported with the following triple: %s" % (
+ test.suite.config.target_triple,))
+
+ if test.config.limit_to_features:
+ # Check that we have one of the limit_to_features features in requires.
+ limit_to_features_tests = [f for f in test.config.limit_to_features
+ if f in requires]
+ if not limit_to_features_tests:
+ msg = ', '.join(test.config.limit_to_features)
+ return lit.Test.Result(Test.UNSUPPORTED,
+ "Test requires one of the limit_to_features features %s" % msg)
+
+ return script
+
+def _runShTest(test, litConfig, useExternalSh, script, tmpBase):
# Create the output directory if it does not already exist.
lit.util.mkdir_p(os.path.dirname(tmpBase))
+ execdir = os.path.dirname(test.getExecPath())
if useExternalSh:
res = executeScript(test, litConfig, tmpBase, script, execdir)
else:
if test.config.unsupported:
return (Test.UNSUPPORTED, 'Test is unsupported')
- res = parseIntegratedTestScript(test, useExternalSh, extra_substitutions)
- if isinstance(res, lit.Test.Result):
- return res
+ script = parseIntegratedTestScript(test)
+ if isinstance(script, lit.Test.Result):
+ return script
if litConfig.noExecute:
return lit.Test.Result(Test.PASS)
- script, tmpBase, execdir = res
- return _runShTest(test, litConfig, useExternalSh, script, tmpBase, execdir)
-
+ tmpDir, tmpBase = getTempPaths(test)
+ substitutions = list(extra_substitutions)
+ substitutions += getDefaultSubstitutions(test, tmpDir, tmpBase,
+ normalize_slashes=useExternalSh)
+ script = applySubstitutions(script, substitutions)
+
+ # Re-run failed tests up to test_retry_attempts times.
+ attempts = 1
+ if hasattr(test.config, 'test_retry_attempts'):
+ attempts += test.config.test_retry_attempts
+ for i in range(attempts):
+ res = _runShTest(test, litConfig, useExternalSh, script, tmpBase)
+ if res.code != Test.FAIL:
+ break
+ # If we had to run the test more than once, count it as a flaky pass. These
+ # will be printed separately in the test summary.
+ if i > 0 and res.code == Test.PASS:
+ res.code = Test.FLAKYPASS
+ return res
projects / oota-llvm.git / blobdiff