test.py 文件源码

def _run_ansible_lint(organization):
    al_bin = os.path.join(aeriscloud_path, 'venv/bin/ansible-lint')

    env = ansible_env(os.environ.copy())

    if organization:
        environment_files = glob.glob(get_env_path(organization) + '/*.yml')
    else:
        environment_files = glob.glob(organization_path + '/*/*.yml')

    if not environment_files:
        return 0

    args = environment_files + ['-r', os.path.join(ansible_path, 'rules')]

    click.echo('Running ansible-lint ... ', nl=False)

    errors = 0
    try:
        python(al_bin, *args,
               _env=env, _err_to_out=True, _ok_code=[0])
        click.echo('[%s]' % click.style('OK', fg='green'))
    except ErrorReturnCode as e:
        parser = re.compile(
            r'^\[(?P<error_code>[^\]]+)\] (?P<error_message>[^\n]+)\n'
            r'%s(?P<file_name>[^:]+):(?P<line_number>[0-9]+)\n'
            r'Task/Handler: (?P<task_name>[^\n]+)\n\n' % (ansible_path + '/'),
            re.MULTILINE
        )

        click.echo('[%s]\n' % click.style('FAIL', fg='red'))

        last_file = None
        pos = 0
        while pos < len(e.stdout):
            match = parser.match(e.stdout, pos)
            if not match:
                click.secho("Error: %s" % e.stdout)
                errors += 1
                break
            error = match.groupdict()

            if error['file_name'] != last_file:
                click.secho('  Errors in file: %s' % error['file_name'],
                            fg='blue', bold=True)
                last_file = error['file_name']

            click.echo('    line %s task %s: %s %s' % (
                click.style(error['line_number'], fg='green'),
                click.style(error['task_name'], fg='green'),
                click.style(error['error_code'], fg='red'),
                click.style(error['error_message'], fg='red'),
            ))
            errors += 1
            pos = match.end()
    return errors