160:fda1f8cab608
Anton Shestakov <av6@dwimlabs.net>, Mon, 18 Jul 2016 16:04:12 +0800
tests: test utils.lookup_option()

next change 178:290623d1275c
previous change 157:1770e08a2431

checker.py

Permissions: -rwxr-xr-x

Other formats: Feeds:
#!/usr/bin/env python -u
from __future__ import absolute_import, print_function
import os
from argparse import ArgumentParser, FileType
from fnmatch import fnmatchcase
from shutil import rmtree
from subprocess import check_call, check_output, CalledProcessError
from tempfile import mkdtemp
import yaml
from candolint.utils import rel, timestamp
def run_ignore_codes(fn, args, codes):
try:
return fn(args)
except CalledProcessError as e:
if e.returncode not in codes:
raise
return e.output
def run(args, silent=False, get_output=False, ignore_codes=None):
if get_output:
fn = check_output
else:
fn = check_call
try:
if not silent:
print('$ ' + ' '.join(args))
if ignore_codes:
result = run_ignore_codes(fn, args, ignore_codes)
else:
result = fn(args)
if get_output:
return result
else:
return True
except Exception as e:
print('# C&O error: {}'.format(e))
return False
def read_linter_config(name, path=rel('linters')):
try:
fd = open(os.path.join(path, '{}.yml'.format(name)))
except Exception as e:
print("# C&O couldn't load linter config: {}".format(e))
return None
with fd:
try:
return yaml.safe_load(fd)
except Exception as e:
print("# C&O couldn't parse linter config: {}".format(e))
return None
def prevalidate(config):
ok = True
if 'url' not in config:
print("# C&O config doesn't have 'url' defined.")
ok = False
if 'scm' not in config:
print("# C&O config doesn't have 'scm' defined.")
ok = False
elif config['scm'] not in ('git', 'hg'):
print("# C&O checker doesn't support {} yet.".format(config['scm']))
ok = False
return ok
def git_clone(url, dest):
return run(['git', 'clone', url, dest])
def hg_clone(url, dest):
return run(['hg', 'clone', url, dest])
def git_checkout(commit):
return run(['git', 'checkout', commit])
def hg_update(commit):
return run(['hg', 'update', commit])
def git_status():
return run(['git', 'status'])
def hg_summary():
return run(['hg', 'sum'])
def git_show():
template = (r'# C&O commit: %H%n'
r'# C&O commit ref names: %D%n'
r'# C&O commit date: %ai%n'
r'# C&O commit author: %aN%n'
r'# C&O commit message: %s%n')
return run(['git', 'show', '-q', '--format=format:' + template], silent=True)
def hg_log():
template = (r'# C&O commit: {rev}:{node}\n'
r'# C&O commit branch: {branch}\n'
r'# C&O commit date: {date|isodatesec}\n'
r'# C&O commit author: {author|person}\n'
r'# C&O commit message: {desc|firstline}\n')
return run(['hg', 'log', '-r', '.', '-T', template], silent=True)
def git_files(include, exclude):
cmd = ['git', 'ls-files']
cmd.extend(include)
files = run(cmd, silent=True, get_output=True)
if files is not False:
files = files.splitlines()
for e in exclude:
files = [f for f in files if not fnmatchcase(f, e)]
return files
def hg_files(include, exclude):
cmd = ['hg', 'files']
for pat in include:
cmd.extend(('-I', pat))
for pat in exclude:
cmd.extend(('-X', pat))
files = run(cmd, silent=True, get_output=True, ignore_codes=(1,))
if files is not False:
files = files.splitlines()
return files
def execute(tmp, config, args):
if not prevalidate(config):
return False
os.chdir(tmp)
source = './source'
print('# C&O task: clone')
print('# C&O project URL: {}'.format(config['url']))
if config['scm'] == 'git':
os.environ['LC_ALL'] = 'en_US.UTF-8'
elif config['scm'] == 'hg':
os.environ['HGPLAIN'] = '1'
if config['scm'] == 'git':
if not git_clone(config['url'], source):
return False
elif config['scm'] == 'hg':
if not hg_clone(config['url'], source):
return False
print('$ cd {}'.format(source))
os.chdir(source)
if args.commit is not None:
if config['scm'] == 'git':
if not git_checkout(args.commit):
return False
elif config['scm'] == 'hg':
if not hg_update(args.commit):
return False
if config['scm'] == 'git':
if not git_status():
return False
elif config['scm'] == 'hg':
if not hg_summary():
return False
if config['scm'] == 'git':
if not git_show():
return False
elif config['scm'] == 'hg':
if not hg_log():
return False
print('# C&O task: setup')
requested = []
for linter in config.get('linters', []):
if 'name' in linter and linter['name'] not in requested:
requested.append(linter['name'])
print('# C&O linters requested: {}'.format(' '.join(requested)))
linter_config = {}
for linter in config.get('linters', []):
if 'name' not in linter:
continue
name = linter['name']
if name in linter_config:
continue
lc = read_linter_config(name)
if lc is None or 'exec' not in lc:
continue
linter_config[name] = lc
for item in lc.get('setup', []):
if not run(item):
return False
if 'version' in lc and not run(lc['exec'] + lc['version']):
return False
print('# C&O linters installed: {}'.format(' '.join(linter_config.keys())))
print('# C&O task: checks')
for linter in config.get('linters', []):
if 'name' not in linter or 'include' not in linter:
continue
if linter['name'] not in linter_config:
continue
if config['scm'] == 'git':
files = git_files(linter['include'], linter.get('exclude', []))
elif config['scm'] == 'hg':
files = hg_files(linter['include'], linter.get('exclude', []))
if files is False:
return False
lc = linter_config[linter['name']]
for f in files:
cmd = lc['exec']
flags = lc.get('flags', []) + linter.get('flags', [])
pf = lc.get('post_flags', []) + linter.get('post_flags', [])
codes = lc.get('codes', (1,))
if not run(cmd + flags + [f] + pf, ignore_codes=codes):
return False
return True
def wrapper(config, args):
print('# C&O job started: {}'.format(timestamp()))
tmp = mkdtemp(prefix='candolint.')
if not execute(tmp, config, args):
print('# C&O job failed')
print('# C&O task: cleanup')
rmtree(tmp)
print('# C&O job finished: {}'.format(timestamp()))
def bad_exit():
print('# C&O job failed (exception not caught)')
print('# C&O job finished: {}'.format(timestamp()))
def main():
parser = ArgumentParser()
parser.add_argument(
'config', type=FileType('r'),
help='project configuration file (YAML)')
parser.add_argument('commit', nargs='?', help='commit to update to')
args = parser.parse_args()
config = yaml.safe_load(args.config)
try:
wrapper(config, args)
except:
bad_exit()
raise
if __name__ == '__main__':
main()