General Pacbio Tool test runner

tool_runner.py

import os
import logging
import abc
import functools

import networkx as nx
from fabric.api import lcd, local, shell_env, prefix, settings

log = logging.getLogger(__name__)


def compose(*funcs):
    """Functional composition

    [f, g, h] will be f(g(h(x)))
    """
    def compose_two(f, g):
        def c(x):
            return f(g(x))
        return c
    return functools.reduce(compose_two, funcs)


class TestToolBase(object):
    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod
    def to_cmds(self, dir_name):
        """This should return the necessary commands"""
        pass

    def __repr__(self):
        return "<{k} >".format(k=self.__class__.__name__)


class PyTestToolBase(TestToolBase):
    """Base class for all Python Package TestTools"""
    REQUIRED_PYTHON_PACKAGES = ()


class PythonUnitTestTool(PyTestToolBase):
    REQUIRED_PYTHON_PACKAGES = ('nose', )
    EXE = 'nosetests'

    def __init__(self, relative_path, output_file="nosetests.xml", options=None):
        # relative path to tests directory
        self.relative_path = relative_path
        self.output_file = output_file
        # extra args to pass directly to nose
        self.options = options

    def __repr__(self):
        _d = dict(k=self.__class__.__name__,
                  r=self.relative_path,
                  o=self.options)
        return "<{k} {r} options:{o} >".format(**_d)

    def to_cmds(self, dir_name):
        cmds = []
        _d = dict(e=self.__class__.EXE,
                  x=os.path.join(dir_name, self.output_file),
                  t=os.path.join(dir_name, self.relative_path))
        cmds.append("{e} --xunit-file={x} {t}".format(**_d))
        return cmds


class PythonUnitCoverageTestTool(PythonUnitTestTool):
    REQUIRED_PYTHON_PACKAGES = ('nose', 'nosexcover')

    def __init__(self, relative_path, package_name, output_file="nosetests.xml", options=None):
        super(PythonUnitCoverageTestTool, self).__init__(relative_path, output_file=output_file, options=options)
        self.package_name = package_name

    def to_cmds(self, dir_name):
        cmds = []
        o = os.path.join(dir_name, self.output_file)
        # path to tests
        u = os.path.join(dir_name, self.relative_path)
        _d = dict(e=self.__class__.EXE,
                  x=o,
                  p=self.package_name,
                  u=u)
        cmds.append("{e} --xunit-file={x} --with-coverage --cover-package={p} --cover-inclusive {u}".format(**_d))
        return cmds


class PythonPep8Tool(PyTestToolBase):
    EXE = "pep8"
    REQUIRED_PYTHON_PACKAGES = ('pep8', )

    def __init__(self, package_name, ignores=("E501", ), output_file="pep8.out"):
        self.package_name = package_name
        self.pep_ignores = ignores
        self.output_file = output_file

    def __repr__(self):
        _d = dict(k=self.__class__.__name__,
                  i=self.pep_ignores,
                  p=self.package_name)
        return "<{k} {p} ignores:{i} >".format(**_d)

    def to_cmds(self, dir_name):
        """pep8 --ignore=E501 {p} | tee pep8.out"""
        cmds = []
        o = os.path.join(dir_name, self.output_file)
        i = ' --ignore="{x}"'.format(x=",".join(self.pep_ignores)) if self.pep_ignores else " "
        _d = dict(e=self.__class__.EXE,
                  i=i,
                  o=o,
                  p=self.package_name)
        cmds.append("{e} {i} {p} | tee {o}".format(**_d))
        return cmds


class PyLintTool(PyTestToolBase):
    EXE = 'pylint'
    REQUIRED_PYTHON_PACKAGES = ('pylint', )

    def __init__(self, project_name, disable=("C0103", "C0301", "C0111"),
                 output_file="pylint.out"):
        self.project_name = project_name
        self.disable = disable
        self.output_file = output_file

    def __repr__(self):
        _d = dict(k=self.__class__.__name__,
                  p=self.project_name,
                  d=self.disable)
        return "<{k} {p} disable:{d} >".format(**_d)

    def to_cmds(self, dir_name):
        cmds = []
        d = " --disable=\"{x}\"".format(x=','.join(self.disable))
        o = os.path.join(dir_name, self.output_file)
        p = self.project_name
        _d = dict(e=self.__class__.EXE,
                  d=d,
                  p=p,
                  o=o)
        cmds.append("{e} -f parseable {d} {p} | tee {o}".format(**_d))
        return cmds


class PythonCoverageTool(PyTestToolBase):
    def __init__(self, xml=True, html=True):
        self.xml = xml
        self.html = html

    def to_cmds(self, dir_name):
        cmds = []
        if self.xml:
            cmds.append("python -m coverage xml")
        if self.html:
            cmds.append("python -m coverage html")
        return cmds


class CramRunnerTool(PyTestToolBase):
    EXE = 'run_cram_unit.py'
    REQUIRED_PYTHON_PACKAGES = ("CramUnit", )

    def __init__(self, relative_path, output_file="cram_xunit.xml",
                 debug=True, verbose=True):
        self.relative_path = relative_path
        self.output_file = output_file
        self.debug = debug
        self.verbose = verbose

    def __repr__(self):
        _d = dict(k=self.__class__.__name__,
                  r=self.relative_path)
        return "<{k} {r} >".format(**_d)

    def to_cmds(self, dir_name):
        cmds = []
        v = ' --verbose ' if self.verbose else " "
        d = ' --debug ' if self.debug else ""
        o = os.path.join(dir_name, self.output_file)
        p = os.path.join(dir_name, self.relative_path)
        _d = dict(e=self.__class__.EXE, v=v, d=d, o=o, p=p)
        cmds.append("{e} {v} {d} -x {o} {p}".format(**_d))
        return cmds


class SloccountTool(TestToolBase):
    EXE = '/usr/bin/sloccount'

    def __init__(self, output_file="sloccount.sc"):
        self.output_file = output_file

    def to_cmds(self, dir_name):
        cmds = []
        o = os.path.join(dir_name, self.output_file)
        _d = dict(e=self.__class__.__name__,
                  d=dir_name,
                  o=o)
        cmds.append("{e} --duplicates --wide --details {d} > {o}".format(**_d))
        return cmds


class PacBioTool(object):

    def __init__(self, project_path, project_id, dependencies, tool_tests):
        """

        :type project_id: str
        :type dependencies: list
        :type tool_tests: list
        """
        # Abspath to project. Not sure I like this.
        self.project_path = project_path
        # this is essentially the package name
        self.project_id = project_id
        # the order matters
        self.dependencies = dependencies
        # this should be a list of TestTool instances
        self.tool_tests = tool_tests

    def __repr__(self):
        _d = dict(k=self.__class__.__name__,
                  i=self.project_id,
                  d=",".join(self.dependencies),
                  n=len(self.tool_tests),
                  p=self.project_path)
        return "<{k} {i} ntest:{n} deps:{d} {p} >".format(**_d)


def create_virtualenv(python_path, virtualenv_path,
                      virtualenv_exe='virtualenv.py', system_site_packages=False):
    pass


def uninstall_pacbio_tool(pacbio_tool):
    with lcd(pacbio_tool.project_path):
        # ignore error if the package isn't installed
        with settings(warn_only=True):
            local("pip uninstall -y {x}".format(x=pacbio_tool.project_id))


def install_pacbio_tool(pacbio_tool):
    """This should be generalize to non-python tools

    This should probably install pip install -r REQUIREMENTS.txt if
    it exists, then pip install .
    """
    with lcd(pacbio_tool.project_path):
        local("pip install .")


def uninstall_pacbio_tool_in_virtualenv(pacbio_tool, virtualenv_path):
    with prefix("source {f}/bin/activate".format(f=virtualenv_path)):
        uninstall_pacbio_tool(pacbio_tool)


def install_pacbio_tool_in_virtualenv(pacbio_tool, virtualenv_path):
    with prefix("source {p}/bin/activate".format(p=virtualenv_path)):
        install_pacbio_tool(pacbio_tool)


def install_pacbio_tool_in_seymour_with_virtualenv(pacbio_tool, seymour_home, virtualenv_path):
    with shell_env(SEYMOUR_HOME=seymour_home):
        install_pacbio_tool_in_virtualenv(pacbio_tool, virtualenv_path)


def run_pacbio_tool(pacbio_tool):
    log.info("Running tool {t}".format(t=repr(pacbio_tool)))
    with lcd(pacbio_tool.project_path):
        for tool_test in pacbio_tool.tool_tests:
            cmds = tool_test.to_cmds(pacbio_tool.project_path)
            for cmd in cmds:
                log.info(cmd)
                local(cmd)
            log.info("completed running {x}".format(x=tool_test))
    log.info("completed running tool {t}".format(t=repr(pacbio_tool)))


def to_dependency_graph(pacbio_tools):
    g = nx.DiGraph()
    for pacbio_tool in pacbio_tools:
        g.add_node(pacbio_tool.project_id)
        for d in pacbio_tool.dependencies:
            g.add_edge(d, pacbio_tool.project_id)
    return g


def _resolve_pacbio_tools_dependency(pacbio_tools):
    """Create dependency graph and return the topologically sorted
    pacbio tool instances"""

    # simple look up mechanism
    d = {p.project_id: p for p in pacbio_tools}

    g = to_dependency_graph(pacbio_tools)
    sorted_project_ids = nx.topological_sort(g)
    log.info("sorted dependency graph:")
    log.info(g)
    return [d[project_id] for project_id in sorted_project_ids]


def run_pacbio_tools(pacbio_tools):
    """Run pacbio tools in a the current users context"""
    log.debug("Running {n} tools".format(n=len(pacbio_tools)))

    # build dependency graph
    ordered_pacbio_tools = _resolve_pacbio_tools_dependency(pacbio_tools)

    for pacbio_tool in ordered_pacbio_tools:
        run_pacbio_tool(pacbio_tool)


def run_pacbio_tools_with_seymour_home(pacbio_tools, seymour_home):

    with shell_env(SEYMOUR_HOME=seymour_home):
        with prefix("source $SEYMOUR_HOME/etc/setup.sh"):
            run_pacbio_tools(pacbio_tools)


def run_pacbio_tools_with_virtualenv(pacbio_tools, virtualenv_path):

    with prefix("source {p}/bin/activate".format(p=virtualenv_path)):
        run_pacbio_tools(pacbio_tools)


def run_pacbio_tools_with_seymour_and_virtualenv(pacbio_tools, seymour_home, virtualenv_path):

    with shell_env(SEYMOUR_HOME=seymour_home):
        with prefix("source $SEYMOUR_HOME/etc/setup.sh"):
            with prefix("source {p}/bin/activate".format(p=virtualenv_path)):
                run_pacbio_tools(pacbio_tools)

_REGISTERED_TOOLS = {}


def register_tool(tool_name):
    """Deco to Register tools with pre-configured to be used via a
    factory-esque setup"""

    def wrapped_func(func):
        if tool_name in _REGISTERED_TOOLS:
            raise KeyError("tool '{t}' has already been added to registry.".format(t=tool_name))

        _REGISTERED_TOOLS[tool_name] = func

        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper

    return wrapped_func


# this could be XML or .cfg files
@register_tool('pbcore')
def to_pbcore(project_path):
    tool_tests = [PythonUnitCoverageTestTool('tests', 'pbcore'),
                  PythonPep8Tool('pbcore'),
                  PyLintTool('pbcore')]

    tool = PacBioTool(project_path, 'pbcore', dependencies=[], tool_tests=tool_tests)
    return tool


@register_tool('pbsystem')
def to_pbsystem(project_path):
    tool_tests = [PythonUnitCoverageTestTool('tests', 'pbsystem',
                                             options=" --logging-config log_nose.cfg "),
                  PythonPep8Tool('pbsystem'),
                  PyLintTool('pbsystem')]

    t = PacBioTool(project_path, 'pbsystem', dependencies=['pbcore'], tool_tests=tool_tests)
    return t

@register_tool('pbh5tools')
def to_pbh5tools(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'pbh5tools'),
             PythonPep8Tool('pbh5tools'),
             PyLintTool('pbh5tools'),
             CramRunnerTool('tests/cram')]
    t = PacBioTool(project_path, 'pbh5tools', dependencies=['pbcore'], tool_tests=tests)
    return t


@register_tool('pbbarcode')
def to_pbbarcocde(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'pbbarcode'),
             PythonPep8Tool('pbbarcode'),
             PyLintTool('pbbarcode')]
             #CramRunnerTool('tests/cram')]
    # the sanity.t cram test doesn't work for some reason
    t = PacBioTool(project_path, 'pbbarcode', dependencies=['pbcore', 'pbh5tools'], tool_tests=tests)
    return t


@register_tool('pbsamtools')
def to_pbsamtools(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'pbsamtools'),
             PythonPep8Tool('pbsamtools'),
             PyLintTool('pbsamtools'),
             CramRunnerTool('tests/cram')]
    t = PacBioTool(project_path, 'pbsamtools', dependencies=['pbcore'], tool_tests=tests)
    return t


@register_tool('kineticsTools')
def to_pbbarcocde_pacbio_tool(project_path):
    tests = [PythonUnitCoverageTestTool('test', 'kineticsTools'),
             PythonPep8Tool('kineticsTools'),
             PyLintTool('kineticsTools'),
             CramRunnerTool('test/cram')]
    t = PacBioTool(project_path, 'kineticsTools', dependencies=['pbcore'], tool_tests=tests)
    return t


@register_tool('pbaha')
def to_pbaha(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'pbaha'),
             PythonPep8Tool('pbaha'),
             PyLintTool('pbaha'),
             CramRunnerTool('tests/cram')]
    t = PacBioTool(project_path, 'pbaha', dependencies=['pbcore', 'pbreports'], tool_tests=tests)
    return t


@register_tool('pbalign')
def to_pbalign(project_path):
    tests = [PythonUnitCoverageTestTool('tests/unit', 'pbalign'),
             PythonPep8Tool('pbalign'),
             PyLintTool('pbalign'),
             CramRunnerTool('tests/cram')]
    t = PacBioTool(project_path, 'pbalign', dependencies=['pbcore'], tool_tests=tests)
    return t

# This has a blasr dependency. It can't be installed without the *real* dependencies
# @register_tool('pbbridgemapper')
# def to_pbbridgemapper(project_path):
#     tests = [PythonUnitCoverageTestTool('tests/unit', 'pbbridgemapper'),
#              PythonPep8Tool('pbbridgemapper'),
#              PyLintTool('pbbridgemapper')]
#     t = PacBioTool(project_path, 'pbbridgemapper', dependencies=['pbcore', 'blasr'], tool_tests=tests)
#     return t


@register_tool('pbfilter')
def to_pbbridgemapper(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'pbfilter'),
             PythonPep8Tool('pbfilter'),
             PyLintTool('pbfilter')]
    t = PacBioTool(project_path, 'pbfilter', dependencies=['pbcore'], tool_tests=tests)
    return t


@register_tool('pbreports')
def to_pbreports(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'pbreports'),
             PythonPep8Tool('pbreports'),
             PyLintTool('pbreports')]
    t = PacBioTool(project_path, 'pbreports', dependencies=['pbcore', 'pbsystem'], tool_tests=tests)
    return t

# Fails because of cython and references to '../../third_party
# @register_tool('pbtranscript')
# def to_pbtranscript(project_path):
#     tests = [PythonUnitCoverageTestTool('tests', 'pbtranscript'),
#              PythonPep8Tool('pbtranscript'),
#              PyLintTool('pbtranscript'),
#              CramRunnerTool('tests/cram')]
#     t = PacBioTool(project_path, 'pbtranscript', dependencies=['pbcore'], tool_tests=tests)
#     return t


@register_tool('CorrelatedVariants')
def to_correlated_variants(project_path):
    tests = [PythonUnitCoverageTestTool('tests', 'CorrelatedVariants'),
             PythonPep8Tool('CorrelatedVariants'),
             PyLintTool('CorrelatedVariants'),
             CramRunnerTool('tests')]
    t = PacBioTool(project_path, 'CorrelatedVariants', dependencies=['pbcore'], tool_tests=tests)
    return t


def pacbio_tool_factory(root_biofx_path, package_name):

    # hacky look up
    _LIB_TOOLS = ('pbcore', 'pbsystem')

    if package_name in _REGISTERED_TOOLS:
        func = _REGISTERED_TOOLS[package_name]
        if package_name in _LIB_TOOLS:
            p = os.path.join(root_biofx_path, 'lib', 'python', package_name)
        else:
            p = os.path.join(root_biofx_path, 'tools', package_name)

        # the tool should be responsible for checking the path
        return func(p)
    else:
        _d = dict(p=package_name, x=",".join(_REGISTERED_TOOLS.keys()))
        raise KeyError("Unable to find '{p}'. Registered tools/libs {x}".format(**_d))


def get_registered_tools(root_biofx_path):
    pacbio_tools = []
    for package_name in _REGISTERED_TOOLS.keys():
        pacbio_tool = pacbio_tool_factory(root_biofx_path, package_name)
        pacbio_tools.append(pacbio_tool)

    # resolve dependencies and topo sort the tools
    sorted_pacbio_tools = _resolve_pacbio_tools_dependency(pacbio_tools)
    return sorted_pacbio_tools


def create_virtualenv_and_run(python_path, root_biofx_path, virtualenv_path):

    pacbio_tools = get_registered_tools(root_biofx_path)

    create_virtualenv(python_path, virtualenv_path, system_site_packages=True)

    run_pacbio_tools_with_virtualenv(pacbio_tools, virtualenv_path)


def create_virtualenv_from_seymour_and_run(seymour_home, root_project_path, virtualenv_path):
    python_path = os.path.join(seymour_home, 'redist/python2.7/bin/python')
    return create_virtualenv_and_run(python_path, root_project_path, virtualenv_path)