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

#!/usr/bin/env python

import argparse
import os.path
import subprocess
import tempfile
from glob import glob
from subprocess import check_output
from typing import (
    Any,
    Dict,
    List,
)

from galaxy.util import unicodify
from galaxy.util.path import StrPath
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: StrPath
) -> 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()