path: root/pygitchecker.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Simple, on demand git checker"""
import argparse
import errno
import io
import json
import os
import subprocess
import sys

import git


class ValidationError(Exception):
    """ValidationError"""


def eprint(*arg, **kwargs):
    """stdderr print wrapper"""
    print(*arg, file=sys.stderr, flush=True, **kwargs)


def is_git_repo(abspath):
    """True of abspath points to a git repo, False otherwise"""
    if not os.path.isdir(abspath):
        return False
    try:
        git.Repo(abspath)
        return True
    except git.InvalidGitRepositoryError:
        return False


def modifications_exist(abspath):
    """
    True if modifications exist for git repo representing abspath
    False if no modifications exist

    git.InvalidGitRepositoryError raised if abspath points to illegal repo
    """
    repo = git.Repo(abspath)
    return bool(repo.is_dirty() or repo.index.diff(None))


def print_modification_status(abspath, status, args):
    """Prints the modification status for the given abspath"""
    if status:
        if args.verbose:
            print(f'{abspath} -> modified')
        else:
            print(abspath)
        return status
    if args.verbose:
        print(f'{abspath} -> unchanged')
    return status


def fetch_git_repos(abspath: str):
    """Returns a list of git repo paths under abspath"""
    repo_list = []
    for root, dirs, _ in os.walk(abspath):
        if '/.git' in root:
            continue
        if '.git' in dirs:
            try:
                git.Repo(root)
                repo_list.append(root)
            except git.InvalidGitRepositoryError:
                continue
    return repo_list


def fetch_git_files(args):
    """Fetches new and modified .py-files using git"""
    try:
        repo = git.Repo(args.repo_path)
        modified_files = {
            mfile.a_path
            for mfile in repo.index.diff(None)
            if mfile.a_path.lower().endswith('.py')
        }
        staged_files = {
            sfile.a_path
            for sfile in repo.index.diff('HEAD')
            if sfile.a_path.lower().endswith('.py')
        }
        untracked_files = set(
            filter(lambda f: f.lower().endswith('.py'), repo.untracked_files)
        )
        return list(modified_files | staged_files | untracked_files)
    except git.InvalidGitRepositoryError:
        eprint(f'{args.repo_path} is not a valid git repository')
        sys.exit(1)


def fetch_all_files(abspath):
    """Fetches all .py-files"""
    all_files_list = []
    for root, _, files in os.walk(abspath):
        for f in files:
            if os.path.splitext(f)[1].lower() == '.py':
                all_files_list.append(os.path.join(abspath, root, f))
    return all_files_list


def validate_python_file(pfile, tests, args):
    """
    Validates a single .py file against tests

    :param tests: A sequence of test dicts
    :type tests: collections.Sequence

    :param pfile: Python file (abs. path)
    :type pfile: str

    :param args: Argparse Namespace-object
    :type args: argparse.Namespace

    :return: False if validation fails, False otherwise
    :rtype: bool
    """
    skipped = []
    if args.skip:
        skipped = args.skip.split(',')
    for test in tests:
        if test['name'] in skipped:
            if args.verbose:
                print(f'{pfile} <- {test["name"]} (skipping)')
            continue
        if args.verbose:
            print(f'{pfile} <- {test["name"]}')
        try:
            params = [param.replace('%p', pfile) for param in test['params']]
            subprocess.run(params, check=test['check'])
        except subprocess.CalledProcessError:
            if not args.continue_checks:
                if args.verbose:
                    eprint(f'{pfile}: {test["name"]} validation failed')
                sys.exit(0)


def main(inargs=None):
    """Main entry point"""
    parser = argparse.ArgumentParser(
        description='The following options are available'
    )
    parser.add_argument(
        'repo_path',
        metavar='REPO',
        nargs='?',
        default=os.getcwd(),
        help='Git repo path (default: cwd)',
    )
    parser.add_argument(
        '-a',
        '--all-files',
        action='store_true',
        dest='all_files',
        help='Check all files, not only the modified ones',
    )
    parser.add_argument(
        '-C',
        '--continue-checks',
        action='store_true',
        dest='continue_checks',
        help='Continue with the checks if validation fails',
    )
    parser.add_argument(
        '-c',
        '--config',
        metavar='<file>',
        type=str,
        default=os.path.expanduser('~/.pygitchecker.json'),
        dest='config_file',
        help='Config file (default: ~/.pygitchecker.json)',
    )
    parser.add_argument(
        '-F',
        '--file',
        metavar='<file>',
        type=str,
        default='',
        dest='python_file',
        help='Validate a particular .py file',
    )
    parser.add_argument(
        '-f',
        '--full',
        action='store_true',
        dest='full_search',
        help='Consider repos that contain untracked files (used with -m)',
    )
    parser.add_argument(
        '-l',
        '--list-only',
        action='store_true',
        dest='list_only',
        help='Only lists the target files without performing any checks',
    )
    parser.add_argument(
        '-m',
        '--modifications-only',
        action='store_true',
        dest='mod_only',
        help='Only check for git modifications',
    )
    parser.add_argument(
        '-r',
        '--recursive',
        action='store_true',
        dest='recursive',
        help='Look for several repos recursively',
    )
    parser.add_argument(
        '-s',
        '--skip',
        metavar='<test1[,test2,test3...]>',
        type=str,
        default='',
        dest='skip',
        help='Skip one or more tests',
    )
    parser.add_argument(
        '-v',
        '--verbose',
        action='store_true',
        dest='verbose',
        help='Verbosity',
    )
    args = parser.parse_args(inargs)
    try:
        with io.open(args.config_file, 'r', encoding='utf-8') as fp:
            config_dict = json.load(fp)
    except Exception as e:
        eprint(f'Unable to parse {args.config_file}: {e}')
        sys.exit(errno.EIO)
    if args.python_file:
        validate_python_file(args.python_file, config_dict.get('tests'), args)
        sys.exit(0)
    abs_repo_path = os.path.abspath(os.path.expanduser(args.repo_path))
    if args.mod_only:
        if not args.recursive:
            if not is_git_repo(abs_repo_path):
                eprint(f'{abs_repo_path} is not a valid git repository')
                sys.exit(1)
            print_modification_status(
                abs_repo_path, modifications_exist(abs_repo_path), args
            )
            sys.exit(0)
        # resursive
        repo_list = fetch_git_repos(abs_repo_path)
        if not repo_list:
            print('No git-repos found')
            sys.exit(0)
        repo_list.sort()
        for repo_path in repo_list:
            print_modification_status(
                repo_path, modifications_exist(repo_path), args
            )
        sys.exit(0)
    if args.all_files:
        selected_files = fetch_all_files(abs_repo_path)
    else:
        selected_files = fetch_git_files(args)
    if not selected_files:
        print('No files selected')
        sys.exit(0)
    # now run the real tests
    for target_file in selected_files:
        if args.list_only:
            print(target_file)
            continue
        if args.verbose:
            print(f'Validating {target_file}:')
        validate_python_file(target_file, config_dict.get('tests'), args)


if __name__ == '__main__':
    main()