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 os
import os.path
import subprocess
import tempfile
from glob import glob
from subprocess import check_output
from typing import (
    Any,
    Dict,
    List,
    Union,
)

from galaxy.util import unicodify
from .get_tests import (
    hashed_test_search,
    import_test_to_command_list,
    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", os.path.join(filepath, container), f"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(f"Docker to Singularity conversion failed.\nOutput was:\n{unicodify(e.output)}")
[docs]def singularity_container_test( tests: Dict[str, Dict[str, Any]], installation: str, filepath: Union[str, os.PathLike] ) -> Dict[str, List]: """ Run tests, record if they pass or fail """ test_results: Dict[str, List] = {"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, os.path.join(filepath, container)] test_passed = True errors = [] for test_command in test.get("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 + ["bash", "-c", test_command], stderr=subprocess.STDOUT) except subprocess.CalledProcessError: try: check_output(exec_command + [test_command], stderr=subprocess.STDOUT) except subprocess.CalledProcessError as e: errors.append({"command": test_command, "output": unicodify(e.output)}) test_passed = False for imp in test.get("imports", []): try: check_output( exec_command + import_test_to_command_list(test["import_lang"], 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(f"{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(f"\n\t\t{container}") f.write("\n\tTEST FAILED:") for container in test_results["failed"]: f.write(f"\n\t\t{container['container']}") for error in container["errors"]: f.write( "\n\t\t\tCOMMAND: {}\n\t\t\t\tERROR:{}".format( error.get("command", f"import{error.get('import', 'nothing found')}"), error["output"] ) ) f.write("\n\tNO TEST AVAILABLE:") for container in test_results["notest"]: f.write(f"\n\t\t{container}")
if __name__ == "__main__": main()