Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tool_util.deps.mulled.get_tests

#!/usr/bin/env python

"""
searches for tests for packages in the bioconda-recipes repo and on Anaconda, looking in different file locations. If no test can be found for the specified version, it will look for tests for other versions of the same package.

A shallow search (default for singularity and conda generation scripts) just checks once on Anaconda for the specified version.
"""
# import doctest
import json
import logging
import tarfile
from glob import glob

import requests
import yaml
try:
    from jinja2 import Template
    from jinja2.exceptions import UndefinedError
except ImportError:
    Template = None  # type: ignore
    UndefinedError = Exception  # type: ignore

from .util import (
    get_file_from_recipe_url,
    split_container_name,
)

INSTALL_JINJA_EXCEPTION = "This mulled functionality required jinja2 but it is unavailable, install condatesting extras."



[docs]def get_commands_from_yaml(yaml_content):
    """
    Parse tests from Conda's meta.yaml file contents
    """
    if Template is None:
        raise Exception(INSTALL_JINJA_EXCEPTION)
    package_tests = {}

    try:
        # we expect to get an input in bytes, so first decode to string; run the file through the jinja processing; load as yaml
        meta_yaml = yaml.safe_load(Template(yaml_content.decode('utf-8')).render())
    except (yaml.scanner.ScannerError, UndefinedError) as e:  # what about things like {{ compiler('cxx') }}
        logging.info(e, exc_info=True)
        return None
    try:
        if meta_yaml['test']['commands'] != [None] and meta_yaml['test']['commands'] is not None:
            package_tests['commands'] = meta_yaml['test']['commands']
    except (KeyError, TypeError):
        logging.info('Error reading commands')
    try:
        if meta_yaml['test']['imports'] != [None] and meta_yaml['test']['imports'] is not None:
            package_tests['imports'] = meta_yaml['test']['imports']
    except (KeyError, TypeError):
        logging.info('Error reading imports')

    if len(package_tests.get('commands', []) + package_tests.get('imports', [])) == 0:
        return None

    # need to know what scripting languages are needed to run the container
    try:
        requirements = list(meta_yaml['requirements']['run'])
    except (KeyError, TypeError):
        logging.info('Error reading requirements', exc_info=True)
    else:
        for requirement in requirements:
            if requirement.split()[0] == 'perl':
                package_tests['import_lang'] = 'perl -e'
                break
            # elif ... :
                # other languages if necessary ... hopefully python and perl should suffice though
        else:  # python by default
            package_tests['import_lang'] = 'python -c'
    return package_tests




[docs]def get_run_test(file):
    r"""
    Get tests from a run_test.sh file
    """
    package_tests = {}
    package_tests['commands'] = [file.replace('\n', ' && ')]
    return package_tests




[docs]def get_anaconda_url(container, anaconda_channel='bioconda'):
    """
    Download tarball from anaconda for test
    """
    name = split_container_name(container)  # list consisting of [name, version, (build, if present)]
    return "https://anaconda.org/{}/{}/{}/download/linux-64/{}.tar.bz2".format(anaconda_channel, name[0], name[1], '-'.join(name))




[docs]def prepend_anaconda_url(url):
    """
    Take a partial url and prepend 'https://anaconda.org'
    """
    return 'https://anaconda.org%s' % url




[docs]def get_test_from_anaconda(url):
    """
    Given the URL of an anaconda tarball, return tests
    """
    try:
        tarball = get_file_from_recipe_url(url)
    except tarfile.ReadError:
        return None

    try:
        metafile = tarball.extractfile('info/recipe/meta.yaml')
    except (tarfile.ReadError, KeyError, TypeError):
        pass
    else:
        package_tests = get_commands_from_yaml(metafile.read())
        if package_tests:
            return package_tests

    # this part is perhaps unnecessary, but some of the older tarballs have a testfile with .yaml.template ext
    try:
        metafile = tarball.extractfile('info/recipe/meta.yaml.template')
    except (tarfile.ReadError, KeyError, TypeError):
        pass
    else:
        package_tests = get_commands_from_yaml(metafile)
        if package_tests:
            return package_tests

    # if meta.yaml was not present or there were no tests in it, try and get run_test.sh instead
    try:
        run_test = tarball.extractfile('info/recipe/run_test.sh')
        return get_run_test(run_test)
    except KeyError:
        logging.info("run_test.sh file not present.")
        return None




[docs]def find_anaconda_versions(name, anaconda_channel='bioconda'):
    """
    Find a list of available anaconda versions for a given container name
    """
    r = requests.get(f"https://anaconda.org/{anaconda_channel}/{name}/files")
    urls = []
    for line in r.text.split('\n'):
        if 'download/linux' in line:
            urls.append(line.split('"')[1])
    return urls




[docs]def open_recipe_file(file, recipes_path=None, github_repo='bioconda/bioconda-recipes'):
    """
    Open a file at a particular location and return contents as string
    """
    if recipes_path:
        return open(f'{recipes_path}/{file}').read()
    else:  # if no clone of the repo is available locally, download from GitHub
        r = requests.get(f'https://raw.githubusercontent.com/{github_repo}/master/{file}')
        if r.status_code == 404:
            raise OSError
        else:
            return r.content




[docs]def get_alternative_versions(filepath, filename, recipes_path=None, github_repo='bioconda/bioconda-recipes'):
    """
    Return files that match ``filepath/*/filename`` in the bioconda-recipes repository
    """
    if recipes_path:
        return [n.replace('%s/' % recipes_path, '') for n in glob(f'{recipes_path}/{filepath}/*/{filename}')]
    # else use the GitHub API:
    versions = []
    r = json.loads(requests.get(f'https://api.github.com/repos/{github_repo}/contents/{filepath}').text)
    for subfile in r:
        if subfile['type'] == 'dir':
            if requests.get('https://raw.githubusercontent.com/{}/master/{}/{}'.format(github_repo, subfile['path'], filename)).status_code == 200:
                versions.append('{}/{}'.format(subfile['path'], filename))
    return versions




[docs]def try_a_func(func1, func2, param, container):
    """
    Try to perform a function (or actually a combination of two functions: first getting the file and then processing it)
    """
    try:
        result = func1(func2(*param))
    except OSError:
        return None
    if result:
        result['container'] = container
        return result