Warning

This document is for an in-development version 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.mulled_update_singularity_containers

#!/usr/bin/env python

import argparse
import subprocess
import tempfile
from glob import glob
from subprocess import check_output

from galaxy.util import unicodify
from .get_tests import hashed_test_search, main_test_search


[docs]def get_list_from_file(filename): """ Returns a list of containers stored in a file (one on each line) """ with open(filename) as fh: return [_ for _ in fh.read().splitlines() if _] # if blank lines are in the file empty strings must be removed
[docs]def docker_to_singularity(container, installation, filepath, no_sudo=False): """ Convert docker to singularity container. """ cmd = [installation, 'build', '/'.join((filepath, container)), 'docker://quay.io/biocontainers/' + container] try: if no_sudo: check_output(cmd, stderr=subprocess.STDOUT) else: check_output(cmd.insert(0, 'sudo'), stderr=subprocess.STDOUT) check_output(['sudo', 'rm', '-rf', '/root/.singularity/docker/'], stderr=subprocess.STDOUT) except subprocess.CalledProcessError as e: raise Exception("Docker to Singularity conversion failed.\nOutput was:\n%s" % unicodify(e.output))
[docs]def singularity_container_test(tests, installation, filepath): """ Run tests, record if they pass or fail """ test_results = {'passed': [], 'failed': [], 'notest': []} # create a 'sanitised home' directory in which the containers may be mounted - see http://singularity.lbl.gov/faq#solution-1-specify-the-home-to-mount with tempfile.TemporaryDirectory() as tmpdirname: for container, test in tests.items(): if 'commands' not in test and 'imports' not in test: test_results['notest'].append(container) else: exec_command = [installation, 'exec', '-H', tmpdirname, '/'.join((filepath, container))] test_passed = True errors = [] if test.get('commands', False): for test_command in test['commands']: test_command = test_command.replace('$PREFIX', '/usr/local/') test_command = test_command.replace('${PREFIX}', '/usr/local/') test_command = test_command.replace('$R ', 'Rscript ') try: check_output(exec_command.extend(['bash', '-c', test_command]), stderr=subprocess.STDOUT) except subprocess.CalledProcessError: try: check_output(exec_command.append(test_command), stderr=subprocess.STDOUT) except subprocess.CalledProcessError as e: errors.append( {'command': test_command, 'output': unicodify(e.output)}) test_passed = False if test.get('imports', False): for imp in test['imports']: try: check_output(exec_command.extend([test['import_lang'], 'import ' + imp]), stderr=subprocess.STDOUT) except subprocess.CalledProcessError as e: errors.append({'import': imp, 'output': unicodify(e.output)}) test_passed = False if test_passed: test_results['passed'].append(container) else: test['errors'] = errors test_results['failed'].append(test) return test_results
[docs]def main(): parser = argparse.ArgumentParser( description='Updates index of singularity containers.') parser.add_argument('-c', '--containers', dest='containers', nargs='+', default=None, help="Containers to be generated. If the number of containers is large, it may be simpler to use the --containers-list option.") parser.add_argument('-l', '--container-list', dest='container_list', default=None, help="Name of file containing list of containers to be generated. Alternative to --containers.") parser.add_argument('-f', '--filepath', dest='filepath', help="File path where newly-built Singularity containers are placed.") parser.add_argument('-i', '--installation', dest='installation', help="File path of Singularity installation.") parser.add_argument('--no-sudo', dest='no_sudo', action='store_true', help="Build containers without sudo.") parser.add_argument('--testing', '-t', dest='testing', default=None, help="Performs testing automatically - a name for the output file should be provided. (Alternatively, testing may be done using the separate testing tool.") args = parser.parse_args() if args.containers: containers = args.containers elif args.container_list: containers = get_list_from_file(args.container_list) else: print("Either --containers or --container-list should be selected.") return for container in containers: docker_to_singularity(container, args.installation, args.filepath, args.no_sudo) if args.testing: container_testing({'anaconda_channel': 'bioconda', 'installation': args.installation, 'filepath': args.filepath, 'github_repo': 'bioconda/bioconda-recipes', 'deep_search': False, 'github_local_path': None, 'logfile': args.testing, 'containers': containers})
[docs]def container_testing(args=None): if not args: # i.e. if testing is called directly from CLI and not via main() parser = argparse.ArgumentParser(description='Tests.') parser.add_argument('-c', '--containers', dest='containers', nargs='+', default=None, help="Containers to be tested. If the number of containers is large, it may be simpler to use the --containers-list option.") parser.add_argument('-l', '--container-list', dest='container_list', default=None, help="Name of file containing list of containers to be tested. Alternative to --containers.") parser.add_argument('-f', '--filepath', dest='filepath', help="File path where the containers to be tested are located.") parser.add_argument('-o', '--logfile', dest='logfile', default='singularity.log', help="Filename for a log to be written to.") parser.add_argument('-i', '--installation', dest='installation', help="File path of Singularity installation.") parser.add_argument('--deep-search', dest='deep_search', action='store_true', help="Perform a more extensive, but probably slower, search for tests.") parser.add_argument('--anaconda-channel', dest='anaconda_channel', default='bioconda', help="Anaconda channel to search for tests (default: bioconda).") parser.add_argument('--github-repo', dest='github_repo', default='bioconda/bioconda-recipes', help="Github repository to search for tests - only relevant if --deep-search is activated (default: bioconda/bioconda-recipes") parser.add_argument('--github-local-path', dest='github_local_path', default=None, help="If the bioconda-recipes repository (or other repository containing tests) is available locally, provide the path here. Only relevant if --deep-search is activated.") args = vars(parser.parse_args()) if args['containers']: containers = args['containers'] elif args['container_list']: containers = get_list_from_file(args['container_list']) else: # if no containers are specified, test everything in the filepath containers = [n.split(args['filepath'])[1] for n in glob('%s*' % args['filepath'])] with open(args['logfile'], 'w') as f: f.write("SINGULARITY CONTAINERS GENERATED:") tests = {} for container in containers: if container[0:6] == 'mulled': # if it is a 'hashed container' tests[container] = hashed_test_search( container, args['github_local_path'], args['deep_search'], args['anaconda_channel'], args['github_repo']) else: tests[container] = main_test_search( container, args['github_local_path'], args['deep_search'], args['anaconda_channel'], args['github_repo']) test_results = singularity_container_test( tests, args['installation'], args['filepath']) f.write('\n\tTEST PASSED:') for container in test_results['passed']: f.write('\n\t\t%s' % container) f.write('\n\tTEST FAILED:') for container in test_results['failed']: f.write('\n\t\t%s' % container['container']) for error in container['errors']: f.write('\n\t\t\tCOMMAND: {}\n\t\t\t\tERROR:{}'.format(error.get( 'command', 'import' + error.get('import', 'nothing found')), error['output'])) f.write('\n\tNO TEST AVAILABLE:') for container in test_results['notest']: f.write('\n\t\t%s' % container)
if __name__ == "__main__": main()