blob: 670a7fe7a74518b7185f3b81393c251908ffa536 [file] [log] [blame] [edit]
#!/usr/bin/env vpython
# Copyright 2015 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.
"""Tests generated CIPD packages.
Supposed to be invoked after build.py has run. Uses packages from out/*.cipd and
tests from tests/*.py.
Assumes cipd client is built in out/.cipd_client/cipd_* (true after build.py has
run).
"""
import argparse
import os
import re
import shutil
import subprocess
import sys
import tempfile
# Root of infra.git repository.
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# .exe on Windows.
EXE_SUFFIX = '.exe' if sys.platform == 'win32' else ''
class TestException(Exception):
pass
def print_title(title):
"""Pretty prints a banner to stdout."""
sys.stdout.flush()
sys.stderr.flush()
print
print '-' * 80
print title
print '-' * 80
def get_docstring(test_script):
"""Hacky way to grab a first line of a module docstring using regexps."""
with open(test_script, 'rt') as f:
text = f.read()
m = re.match(r'^.*"""(.*?)"""', text, re.DOTALL)
if not m:
return None
return m.group(1).strip().splitlines()[0]
def find_cipd():
"""Finds a CIPD client to use in PATH."""
exts = ('.exe', '.bat') if sys.platform == 'win32' else ('',)
for p in os.environ.get('PATH', '').split(os.pathsep):
base = os.path.join(p, 'cipd')
for ext in exts:
candidate = base + ext
if os.path.isfile(candidate):
return candidate
return 'cipd' + EXE_SUFFIX
def run_test(cipd_client, package, work_dir, test_script):
"""Extracts a package to a dir and runs test_script with cwd == work_dir."""
print_title('Deploying %s' % os.path.basename(package))
cmd_line = ['cipd', 'pkg-deploy', '-root', work_dir, package]
print ' '.join(cmd_line)
if subprocess.call(args=cmd_line, executable=cipd_client):
raise TestException('Failed to install %s, see logs' % package)
print_title(get_docstring(test_script) or 'Running tests...')
cmd_line = ['python', test_script]
print '%s in %s' % (' '.join(cmd_line), work_dir)
env = os.environ.copy()
env.pop('PYTHONPATH', None)
ret = subprocess.call(
args=cmd_line, executable=sys.executable, env=env, cwd=work_dir)
if ret:
raise TestException('Non zero exit code (%d)' % ret)
def run(
package_out_dir,
package_tests_dir,
work_dir,
packages):
"""Deployes build *.cipd package locally and runs tests against them.
Used to verify the packaged code works when installed as CIPD package, it is
important for infra_python package that has non-trivial structure.
Args:
package_out_dir: where to search for built packages.
work_dir: where to install/update packages into.
packages: names of *.cipd files in package_out_dir or [] for all.
Returns:
0 on success, 1 or error.
"""
# Discover what to test.
paths = []
if not packages:
# Enumerate all known tests in tests/*.py and filter them based on
# availability of corresponding *.cipd package in package_out_dir. It will
# skip any cross-compiled packages, since they have additional '+<platform>'
# suffix in the package file name.
for test in os.listdir(package_tests_dir):
if not test.endswith('.py'):
continue
pkg_file = os.path.join(
package_out_dir, os.path.splitext(test)[0] + '.cipd')
if os.path.exists(pkg_file):
paths.append(pkg_file)
else:
for name in packages:
abs_path = os.path.join(package_out_dir, name)
if not os.path.isfile(abs_path):
raise TestException('No such package file: %s' % name)
paths.append(abs_path)
paths = sorted(paths)
if not paths:
print 'Nothing to test.'
return 0
cipd_client = find_cipd()
if not cipd_client:
return 1
# Run all tests sequentially. Most of the are extra fast.
nuke_temp = False
if not work_dir:
work_dir = tempfile.mkdtemp(suffix='cipd_test')
nuke_temp = True
work_dir = os.path.abspath(work_dir)
try:
fail = False
for path in paths:
name = os.path.splitext(os.path.basename(path))[0]
test_script = os.path.join(package_tests_dir, '%s.py' % name)
if not os.path.isfile(test_script):
print 'Skipping tests for %s - no such file: %s' % (name, test_script)
continue
try:
run_test(
cipd_client=cipd_client,
package=path,
work_dir=os.path.join(work_dir, name),
test_script=test_script)
print ''
print 'PASS'
except TestException as exc:
print >> sys.stderr, ''
print >> sys.stderr, 'FAILED! ' * 10
print >> sys.stderr, 'Tests for %s failed: %s' % (name, exc)
fail = True
return 1 if fail else 0
finally:
if nuke_temp:
try:
shutil.rmtree(work_dir, ignore_errors=True)
except OSError as exc:
print >> sys.stderr, 'Failed to delete %s: %s' % (work_dir, exc)
def main(
args,
package_out_dir=os.path.join(ROOT, 'build', 'out'),
package_tests_dir=os.path.join(ROOT, 'build', 'tests')):
parser = argparse.ArgumentParser(description='Tests infra CIPD packages')
parser.add_argument(
'packages', metavar='NAME', type=str, nargs='*',
help='name of a built package file in build/out/* to deploy and test')
parser.add_argument(
'--work-dir', metavar='DIR', dest='work_dir',
help='directory to deploy packages into (temporary dir by default)')
args = parser.parse_args(args)
return run(
package_out_dir,
package_tests_dir,
args.work_dir,
[n + '.cipd' if not n.endswith('.cipd') else n for n in args.packages])
if __name__ == '__main__':
sys.exit(main(sys.argv[1:]))