PyBitmessage-2024-12-19/fabfile/lib.py

# pylint: disable=not-context-manager
"""
A library of functions and constants for tasks to make use of.

"""

import os
import re
import sys
from functools import wraps

from fabric.api import cd, env, hide, run
from fabric.context_managers import settings, shell_env
from fabvenv import virtualenv

FABRIC_ROOT = os.path.dirname(__file__)
PROJECT_ROOT = os.path.dirname(FABRIC_ROOT)
VENV_ROOT = os.path.expanduser(os.path.join('~', '.virtualenvs', 'pybitmessage-devops'))
PYTHONPATH = os.path.join(PROJECT_ROOT, 'src',)


def coerce_list(value):
    """Coerce a value into a list"""
    if isinstance(value, str):
        return value.split(',')
    else:
        sys.exit("Bad string value {}".format(value))


def coerce_bool(value):
    """Coerce a value into a boolean"""
    if isinstance(value, bool):
        return value
    elif any(
            [
                value in [0, '0'],
                value.lower().startswith('n'),
            ]
    ):
        return False
    elif any(
            [
                value in [1, '1'],
                value.lower().startswith('y'),
            ]
    ):
        return True
    else:
        sys.exit("Bad boolean value {}".format(value))


def flatten(data):
    """Recursively flatten lists"""
    result = []
    for item in data:
        if isinstance(item, list):
            result.append(flatten(item))
        else:
            result.append(item)
    return result


def filelist_from_git(rev=None):
    """Return a list of files based on git output"""
    cmd = 'git diff --name-only'
    if rev:
        if rev in ['cached', 'staged']:
            cmd += ' --{}'.format(rev)
        elif rev == 'working':
            pass
        else:
            cmd += ' -r {}'.format(rev)

    with cd(PROJECT_ROOT):
        with hide('running', 'stdout'):
            results = []
            ansi_escape = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
            clean = ansi_escape.sub('', run(cmd))
            clean = re.sub('\n', '', clean)
            for line in clean.split('\r'):
                if line.endswith(".py"):
                    with settings(warn_only=True):
                        with hide('warnings'):
                            if run('stat {}'.format(os.path.abspath(line))).succeeded:
                                results.append(os.path.abspath(line))
                            else:
                                print 'Deleted file {} skipped.'.format(line)
            return results


def pycodestyle(path_to_file):
    """Run pycodestyle on a file"""
    with virtualenv(VENV_ROOT):
        with hide('warnings', 'running', 'stdout', 'stderr'):
            with settings(warn_only=True):
                return run(
                    'pycodestyle --config={0} {1}'.format(
                        os.path.join(
                            PROJECT_ROOT,
                            'setup.cfg',
                        ),
                        path_to_file,
                    ),
                )


def flake8(path_to_file):
    """Run flake8 on a file"""
    with virtualenv(VENV_ROOT):
        with hide('warnings', 'running', 'stdout'):
            with settings(warn_only=True):
                return run(
                    'flake8 --config={0} {1}'.format(
                        os.path.join(
                            PROJECT_ROOT,
                            'setup.cfg',
                        ),
                        path_to_file,
                    ),
                )


def pylint(path_to_file):
    """Run pylint on a file"""
    with virtualenv(VENV_ROOT):
        with hide('warnings', 'running', 'stdout', 'stderr'):
            with settings(warn_only=True):
                with shell_env(PYTHONPATH=PYTHONPATH):
                    return run(
                        'pylint --rcfile={0} {1}'.format(
                            os.path.join(
                                PROJECT_ROOT,
                                'setup.cfg',
                            ),
                            path_to_file,
                        ),
                    )


def isort(path_to_file):
    """Run isort on a file"""
    with virtualenv(VENV_ROOT):
        with hide('warnings', 'running', 'stdout', 'stderr'):
            with settings(warn_only=True):
                with shell_env(PYTHONPATH=PYTHONPATH):
                    returnable = run(
                        'isort {0}'.format(
                            path_to_file,
                        ),
                    )
                    # isort takes the view that a sorted file in is an error
                    returnable.return_code = not returnable.return_code
                    return returnable


def autopep8(path_to_file):
    """Run autopep8 on a file"""
    with virtualenv(VENV_ROOT):
        with hide('running'):
            with settings(warn_only=True):
                return run(
                    "autopep8 --experimental --aggressive --aggressive -i --max-line-length=119 {}".format(
                        path_to_file
                    ),
                )


def get_filtered_pycodestyle_output(path_to_file):
    """Clean up the raw results for pycodestyle"""

    return [
        i
        for i in pycodestyle(path_to_file).split(os.linesep)
        if i
    ]


def get_filtered_flake8_output(path_to_file):
    """Clean up the raw results for flake8"""

    return [
        i
        for i in flake8(path_to_file).split(os.linesep)
        if i
    ]


def get_filtered_pylint_output(path_to_file):
    """Clean up the raw results for pylint"""

    return [
        i
        for i in pylint(path_to_file).split(os.linesep)
        if all([
            i,
            not i.startswith(' '),
            not i.startswith('\r'),
            not i.startswith('-'),
            not i.startswith('Y'),
            not i.startswith('*'),
            not i.startswith('Using config file'),
        ])
    ]


def default_hosts(hosts):
    """Decorator to apply default hosts to a task"""

    def real_decorator(func):
        """Manipulate env"""
        env.hosts = env.hosts or hosts

        @wraps(func)
        def wrapper(*args, **kwargs):
            """Original function called from here"""
            return func(*args, **kwargs)

        return wrapper

    return real_decorator