Merge pull request #16125 from nsoranzo/release_22.05_fix_get_file_from_conda_package

[22.05] Fix ``get_test_from_anaconda()`` and ``base_image_for_targets()`` functions

.github/workflows/mulled.yaml

@@ -42,6 +42,8 @@ jobs:
         with:
           path: .tox
           key: tox-cache-${{ runner.os }}-${{ steps.full-python-version.outputs.version }}-${{ hashFiles('galaxy root/requirements.txt') }}-mulled
+      - name: Install Apptainer's singularity
+        uses: eWaterCycle/setup-apptainer@v2
       - name: Install tox
         run: pip install tox
       - name: run tests

.github/workflows/toolshed.yaml

@@ -17,7 +17,7 @@ concurrency:
 jobs:
   test:
     name: Test
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version: ['3.7']

doc/source/admin/special_topics/mulled_containers.rst

@@ -153,8 +153,6 @@ Multiple containers can be installed simultaneously by giving ``--containers`` m
 
    $ mulled-update-singularity-containers --containers samtools:1.6--0 bamtools:2.4.1--0 --filepath /tmp/sing/ --installation /usr/local/bin/singularity
 
-.. code-block:: bash
-
 For a large number of containers, it may be more convenient to employ the ``--container-list`` option:
 
 .. code-block:: bash
@@ -163,15 +161,19 @@ For a large number of containers, it may be more convenient to employ the ``--co
 
 Here ``list.txt`` should contain a list of containers, each on a new line.
 
-In order to generate the list file the ``mulled-list`` command may be useful. The following command returns a list of all Docker containers available on the quay.io biocontainers organization, excluding those already available as Singularity containers via https://depot.galaxyproject.org/singularity/.:: bash
+In order to generate the list file the ``mulled-list`` command may be useful. The following command returns a list of all Docker containers available on the quay.io biocontainers organization, excluding those already available as Singularity containers on https://depot.galaxyproject.org/singularity/ .
+
+.. code-block:: bash
 
    $ mulled-list --source docker --not-singularity --blacklist blacklist.txt --file output.txt
 
 The list of containers will be saved as ``output.txt``. The (optional) ``--blacklist`` option may be used to exclude containers which should not included in the output; ``blacklist.txt`` should contain a list of the 'blacklisted' containers, each on a new line.
 
-Containers, once generated, should be tested. This can be achieved by affixing ``--testing test-output.log`` to the command, or alternatively, by use of the dedicated ``mulled-singularity-testing`` tool.:: bash
+The generated containers should also be tested. This can be achieved by affixing ``--testing test-output.log`` to the ``mulled-update-singularity-containers`` command:
 
-   $ mulled-singularity-testing --container-list list.txt --filepath /tmp/sing/ --installation /usr/local/bin/singularity --logfile test-output.txt
+.. code-block:: bash
+
+   $ mulled-update-singularity-containers --container-list list.txt --filepath /tmp/sing/ --installation /usr/local/bin/singularity --testing test-output.log
 
 .. _IUC: https://galaxyproject.org/iuc/
 .. _container annotation:  https://github.com/galaxyproject/galaxy/blob/dev/test/functional/tools/catDocker.xml#L4

lib/galaxy/tool_util/deps/conda_util.py

@@ -517,6 +517,8 @@ def best_search_result(
 
     Return ``None`` if no results match.
     """
+    # Cannot specify the version here (i.e. conda_target.package_specifier)
+    # because if the version is not found, the exec_search() call would fail.
     search_args = [conda_target.package]
     try:
         res = conda_context.exec_search(search_args, json=True, offline=offline, platform=platform)

lib/galaxy/tool_util/deps/mulled/get_tests.py

@@ -12,6 +12,7 @@ from glob import glob
 from typing import (
     Any,
     Dict,
+    List,
     Optional,
 )
 
@@ -26,8 +27,9 @@ except ImportError:
     UndefinedError = Exception  # type: ignore[assignment,misc]
 
 from galaxy.util import unicodify
+from galaxy.util.commands import argv_to_str
 from .util import (
-    get_file_from_conda_package,
+    get_files_from_conda_package,
     MULLED_SOCKET_TIMEOUT,
     split_container_name,
 )
@@ -66,6 +68,7 @@ def get_commands_from_yaml(yaml_content: bytes) -> Optional[Dict[str, Any]]:
         return None
 
     # need to know what scripting languages are needed to run the container
+    package_tests["import_lang"] = "python -c"  # python by default
     try:
         requirements = list(meta_yaml["requirements"]["run"])
     except (KeyError, TypeError):
@@ -77,8 +80,6 @@ def get_commands_from_yaml(yaml_content: bytes) -> Optional[Dict[str, Any]]:
                 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
 
 
@@ -110,15 +111,16 @@ def get_test_from_anaconda(url: str) -> Optional[Dict[str, Any]]:
     """
     Given the URL of an anaconda tarball, return tests
     """
-    name, content = get_file_from_conda_package(
+    content_dict = get_files_from_conda_package(
         url, ["info/recipe/meta.yaml", "info/recipe/meta.yaml.template", "info/recipe/run_test.sh"]
     )
-    if name and content and name.startswith("info/recipe/meta.yaml"):
+    content = content_dict.get("info/recipe/meta.yaml", content_dict.get("info/recipe/meta.yaml.template"))
+    if content:
         package_tests = get_commands_from_yaml(content)
         if package_tests:
             return package_tests
-    if name and content and name == "info/recipe/run_test.sh":
-        return get_run_test(unicodify(content))
+    if "info/recipe/run_test.sh" in content_dict:
+        return get_run_test(unicodify(content_dict["info/recipe/run_test.sh"]))
     return None
 
 
@@ -128,7 +130,7 @@ def find_anaconda_versions(name, anaconda_channel="bioconda"):
     """
     r = requests.get(f"https://anaconda.org/{anaconda_channel}/{name}/files", timeout=MULLED_SOCKET_TIMEOUT)
     urls = []
-    for line in r.text.split("\n"):
+    for line in r.text.splitlines():
         if "download/linux" in line:
             urls.append(line.split('"')[1])
     return urls
@@ -259,23 +261,34 @@ def main_test_search(
     return {"container": container}
 
 
+def import_test_to_command_list(import_lang: str, import_: str) -> List[str]:
+    if import_lang == "python -c":
+        return ["python", "-c", f"import {import_}"]
+    elif import_lang == "perl -e":
+        return ["perl", "-e", f"use {import_}"]
+    else:
+        raise ValueError(f"Unsupported import_lang '{import_lang}'")
+
+
 def hashed_test_search(
-    container, recipes_path=None, deep=False, anaconda_channel="bioconda", github_repo="bioconda/bioconda-recipes"
-):
+    container: str, recipes_path=None, deep=False, anaconda_channel="bioconda", github_repo="bioconda/bioconda-recipes"
+) -> Dict[str, Any]:
     """
     Get test for hashed containers
     """
-    package_tests = {"commands": [], "imports": [], "container": container, "import_lang": "python -c"}
+    package_tests: Dict[str, Any] = {"commands": [], "imports": [], "container": container, "import_lang": "python -c"}
 
-    githubpage = requests.get(
+    response = requests.get(
         f"https://raw.githubusercontent.com/BioContainers/multi-package-containers/master/combinations/{container}.tsv",
         timeout=MULLED_SOCKET_TIMEOUT,
     )
-    if githubpage.status_code == 200:
-        packages = githubpage.text.split(",")  # get names of packages from github
-        packages = [package.split("=") for package in packages]
-    else:
-        packages = []
+    response.raise_for_status()
+    for line in response.text.splitlines():
+        if not line.startswith("#"):
+            break
+    concatenated_targets = line.split("\t")[0]
+    targets = concatenated_targets.split(",")
+    packages = [target.split("=") for target in targets]
 
     containers = []
     for package in packages:
@@ -293,7 +306,8 @@ def hashed_test_search(
     for container in containers:
         tests = main_test_search(container, recipes_path, deep, anaconda_channel, github_repo)
         package_tests["commands"] += tests.get("commands", [])  # not a very nice solution but probably the simplest
+        # Given that this could be a mix of Python and Perl packages, translate imports to commands
         for imp in tests.get("imports", []):
-            package_tests["imports"].append(f"{tests['import_lang']} 'import {imp}'")
+            package_tests["commands"].append(argv_to_str(import_test_to_command_list(tests["import_lang"], imp)))
 
     return package_tests

lib/galaxy/tool_util/deps/mulled/mulled_build.py

@@ -43,7 +43,7 @@ from .util import (
     conda_build_target_str,
     create_repository,
     default_mulled_conda_channels_from_env,
-    get_file_from_conda_package,
+    get_files_from_conda_package,
     PrintProgress,
     quay_repository,
     v1_image_name,
@@ -170,14 +170,16 @@ def base_image_for_targets(targets: List["Target"], conda_context: CondaContext)
     hits = get_conda_hits_for_targets(targets, conda_context)
     for hit in hits:
         try:
-            name, content = get_file_from_conda_package(hit["url"], ["info/about.json", "info/recipe/meta.yaml"])
-            strcontent = unicodify(content)
-            if name == "info/about.json" and json.loads(strcontent).get("extra", {}).get("container", {}).get(
-                "extended-base", False
-            ):
+            content_dict = get_files_from_conda_package(hit["url"], ["info/about.json", "info/recipe/meta.yaml"])
+            if "info/about.json" in content_dict and json.loads(unicodify(content_dict["info/about.json"])).get(
+                "extra", {}
+            ).get("container", {}).get("extended-base", False):
                 return DEFAULT_EXTENDED_BASE_IMAGE
-            elif name == "info/recipe/meta.yaml" and (
-                yaml.safe_load(strcontent).get("extra", {}).get("container", {}).get("extended-base", False)
+            elif "info/recipe/meta.yaml" in content_dict and (
+                yaml.safe_load(unicodify(content_dict["info/recipe/meta.yaml"]))
+                .get("extra", {})
+                .get("container", {})
+                .get("extended-base", False)
             ):
                 return DEFAULT_EXTENDED_BASE_IMAGE
         except Exception:

lib/galaxy/tool_util/deps/mulled/mulled_update_singularity_containers.py

@@ -1,14 +1,23 @@
 #!/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,
 )
 
@@ -25,7 +34,7 @@ def docker_to_singularity(container, installation, filepath, no_sudo=False):
     """
     Convert docker to singularity container.
     """
-    cmd = [installation, "build", "/".join((filepath, container)), f"docker://quay.io/biocontainers/{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)
@@ -36,47 +45,47 @@ def docker_to_singularity(container, installation, filepath, no_sudo=False):
         raise Exception(f"Docker to Singularity conversion failed.\nOutput was:\n{unicodify(e.output)}")
 
 
-def singularity_container_test(tests, installation, filepath):
+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 = {"passed": [], "failed": [], "notest": []}
+    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, "/".join((filepath, container))]
+                exec_command = [installation, "exec", "-H", tmpdirname, os.path.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 ")
+                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.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"], f"import {imp}"]), stderr=subprocess.STDOUT
-                            )
+                            check_output(exec_command + [test_command], stderr=subprocess.STDOUT)
                         except subprocess.CalledProcessError as e:
-                            errors.append({"import": imp, "output": unicodify(e.output)})
+                            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:

lib/galaxy/tool_util/deps/mulled/util.py

@@ -8,12 +8,10 @@ import re
 import sys
 import threading
 from typing import (
+    Dict,
     Iterable,
     List,
-    Optional,
-    Tuple,
     TYPE_CHECKING,
-    Union,
 )
 
 import packaging.version
@@ -350,42 +348,31 @@ def v2_image_name(targets, image_build=None, name_override=None):
         return f"mulled-v2-{package_hash.hexdigest()}{suffix}"
 
 
-def get_file_from_conda_package(
-    url: str, checklist: Union[str, Iterable[str]]
-) -> Tuple[Optional[str], Optional[bytes]]:
+def get_files_from_conda_package(url: str, filepaths: Iterable[str]) -> Dict[str, bytes]:
     """
-    Get file content for an element in a conda package.
+    Get content of specified files in a conda package.
     The url can be a path to a local file or an url.
-    The checklist can be the name of the element to etract
-    or a Iterable of potentially desired elements the content
-    of the first that is contained in the conda package
-    is returned.
-    Return the name and content (bytes) of the first found element
-    and None, None otherwise
+    The filepaths is an iterable of paths to extract from the conda package, if
+    found in it.
+    Return a dictionary mapping each found filepath to the corresponding content
+    (as bytes).
 
-    >>> name, content = get_file_from_conda_package("https://anaconda.org/conda-forge/chopin2/1.0.6/download/noarch/chopin2-1.0.6-pyhd8ed1ab_0.tar.bz2", "info/recipe/meta.yaml")
-    >>> assert name == "info/recipe/meta.yaml"
-    >>> assert isinstance(content, bytes)
-    >>> name, content = get_file_from_conda_package("https://anaconda.org/conda-forge/chopin2/1.0.7/download/noarch/chopin2-1.0.7-pyhd8ed1ab_1.conda", ["info/about.json", "info/recipe/meta.yaml"])
-    >>> assert name == "info/recipe/meta.yaml"
-    >>> assert isinstance(content, bytes)
+    >>> content_dict = get_files_from_conda_package("https://anaconda.org/conda-forge/chopin2/1.0.6/download/noarch/chopin2-1.0.6-pyhd8ed1ab_0.tar.bz2", ["info/recipe/meta.yaml"])
+    >>> assert "info/recipe/meta.yaml" in content_dict, content_dict
+    >>> assert isinstance(content_dict["info/recipe/meta.yaml"], bytes)
+    >>> content_dict = get_files_from_conda_package("https://anaconda.org/conda-forge/chopin2/1.0.7/download/noarch/chopin2-1.0.7-pyhd8ed1ab_1.conda", ["info/about.json", "info/recipe/meta.yaml", "foo/bar"])
+    >>> assert sorted(content_dict.keys()) == ["info/about.json", "info/recipe/meta.yaml"], content_dict
     """
 
-    if isinstance(checklist, str):
-        checklist = set([checklist])
-    else:
-        checklist = set(checklist)
-    # print(checklist)
     try:
         stream = stream_conda_info(url)
     except FileNotFoundError:
         stream = stream_conda_info_from_url(url)
+    ret = {}
     for tar, member in stream:
-        # print(member.name)
-        if member.name in checklist:
-            return member.name, tar.extractfile(member).read()
-    # print("None")
-    return None, None
+        if member.name in filepaths:
+            ret[member.name] = tar.extractfile(member).read()
+    return ret
 
 
 def split_container_name(name):
@@ -423,6 +410,7 @@ image_name = v1_image_name  # deprecated
 __all__ = (
     "build_target",
     "conda_build_target_str",
+    "get_files_from_conda_package",
     "image_name",
     "mulled_tags_for",
     "quay_versions",

test/unit/tool_util/mulled/test_get_tests.py

@@ -42,7 +42,6 @@ def test_get_run_test():
     assert commands["commands"] == [' #!/bin/bash && pslScore 2> /dev/null || [[ "$?" == 255 ]]']
 
 
-@external_dependency_management
 def test_get_anaconda_url():
     url = get_anaconda_url("samtools:1.7--1")
     assert url == "https://anaconda.org/bioconda/samtools/1.7/download/linux-64/samtools-1.7-1.tar.bz2"
@@ -72,9 +71,11 @@ def test_get_test_from_anaconda():
 
     # test for package defining tests in info/recipe/run_test.sh
     tests = get_test_from_anaconda(
-        "https://anaconda.org/bioconda/ucsc-pslmap/366/download/linux-64/ucsc-pslmap-366-hdd26221_0.tar.bz2"
+        "https://anaconda.org/bioconda/mugsy/1.2.3/download/noarch/mugsy-1.2.3-hdfd78af_4.tar.bz2"
     )
-    assert tests and tests["commands"] == ['#!/bin/bash && pslMap 2> /dev/null || [[ "$?" == 255 ]] && ']
+    assert tests and tests["commands"] == [
+        "#!/bin/bash &&  && export MUGSY_INSTALL=${PREFIX}/bin && mugsy -h | grep mugsy > /dev/null && mugsyWGA  --version && synchain-mugsy 2>&1 | grep mugsy > /dev/null && "
+    ]  # This script is clearly broken, but the whole get_tests module is currently far from usable
 
 
 @external_dependency_management
@@ -91,8 +92,8 @@ def test_open_recipe_file():
 
 @external_dependency_management
 def test_get_alternative_versions():
-    versions = get_alternative_versions("recipes/bamtools", "meta.yaml")
-    assert versions == ["recipes/bamtools/2.3.0/meta.yaml"]
+    versions = get_alternative_versions("recipes/bioblend", "meta.yaml")
+    assert versions == ["recipes/bioblend/0.7.0/meta.yaml"]
 
 
 @external_dependency_management

test/unit/tool_util/mulled/test_mulled_update_singularity_containers.py

@@ -1,9 +1,8 @@
 import os
-import shutil
-import tempfile
 
 import pytest
 
+from galaxy.tool_util.deps.mulled.get_tests import main_test_search
 from galaxy.tool_util.deps.mulled.mulled_update_singularity_containers import (
     docker_to_singularity,
     get_list_from_file,
@@ -13,59 +12,42 @@ from galaxy.util import which
 from ..util import external_dependency_management
 
 
-def test_get_list_from_file():
-    test_dir = tempfile.mkdtemp()
-    try:
-        list_file = os.path.join(test_dir, "list_file.txt")
-        with open(list_file, "w") as f:
-            f.write("bbmap:36.84--0\nbiobambam:2.0.42--0\nconnor:0.5.1--py35_0\ndiamond:0.8.26--0\nedd:1.1.18--py27_0")
-        assert get_list_from_file(list_file) == [
-            "bbmap:36.84--0",
-            "biobambam:2.0.42--0",
-            "connor:0.5.1--py35_0",
-            "diamond:0.8.26--0",
-            "edd:1.1.18--py27_0",
-        ]
-    finally:
-        shutil.rmtree(test_dir)
+def test_get_list_from_file(tmp_path) -> None:
+    list_file = os.path.join(tmp_path, "list_file.txt")
+    with open(list_file, "w") as f:
+        f.write("bbmap:36.84--0\nbiobambam:2.0.42--0\nconnor:0.5.1--py35_0\ndiamond:0.8.26--0\nedd:1.1.18--py27_0")
+    assert get_list_from_file(list_file) == [
+        "bbmap:36.84--0",
+        "biobambam:2.0.42--0",
+        "connor:0.5.1--py35_0",
+        "diamond:0.8.26--0",
+        "edd:1.1.18--py27_0",
+    ]
 
 
 @external_dependency_management
 @pytest.mark.skipif(not which("singularity"), reason="requires singularity but singularity not on PATH")
-def test_docker_to_singularity(tmp_path):
-    tmp_dir = str(tmp_path)
-    docker_to_singularity("abundancebin:1.0.1--0", "singularity", tmp_dir, no_sudo=True)
+def test_docker_to_singularity(tmp_path) -> None:
+    docker_to_singularity("abundancebin:1.0.1--0", "singularity", tmp_path, no_sudo=True)
     assert tmp_path.joinpath("abundancebin:1.0.1--0").exists()
 
 
 @external_dependency_management
 @pytest.mark.skipif(not which("singularity"), reason="requires singularity but singularity not on PATH")
-def test_singularity_container_test(tmp_path):
-    test_dir = tempfile.mkdtemp()
-    try:
-        for n in ["pybigwig:0.1.11--py36_0", "samtools:1.0--1", "yasm:1.3.0--0"]:
-            docker_to_singularity(n, "singularity", test_dir, no_sudo=True)
-        results = singularity_container_test(
-            {
-                "pybigwig:0.1.11--py36_0": {
-                    "imports": ["pyBigWig"],
-                    "commands": [
-                        'python -c "import pyBigWig; assert(pyBigWig.numpy == 1); assert(pyBigWig.remote == 1)"'
-                    ],
-                    "import_lang": "python -c",
-                },
-                "samtools:1.0--1": {
-                    "commands": ["samtools --help"],
-                    "import_lang": "python -c",
-                    "container": "samtools:1.0--1",
-                },
-                "yasm:1.3.0--0": {},
-            },
-            "singularity",
-            test_dir,
-        )
-        assert "samtools:1.0--1" in results["passed"]
-        assert results["failed"][0]["imports"] == ["pyBigWig"]
-        assert "yasm:1.3.0--0" in results["notest"]
-    finally:
-        shutil.rmtree(test_dir)
+def test_singularity_container_test(tmp_path) -> None:
+    containers = [
+        "pybigwig:0.3.22--py36h54a71a5_0",  # test Python imports
+        "samtools:1.0--1",
+        "yasm:1.3.0--0",  # test missing tests
+    ]
+    tests = {container: main_test_search(container) for container in containers}
+    for n in tests.keys():
+        docker_to_singularity(n, "singularity", tmp_path, no_sudo=True)
+    results = singularity_container_test(
+        tests,
+        "singularity",
+        tmp_path,
+    )
+    assert "samtools:1.0--1" in results["passed"]
+    assert "pybigwig:0.3.22--py36h54a71a5_0" in results["passed"]
+    assert "yasm:1.3.0--0" in results["notest"]

test/unit/tool_util/test_conda_resolution.py

@@ -2,9 +2,13 @@ import os
 import shutil
 from tempfile import mkdtemp
 
-from galaxy.tool_util.deps import (
-    conda_util,
-    DependencyManager,
+from galaxy.tool_util.deps import DependencyManager
+from galaxy.tool_util.deps.conda_util import (
+    best_search_result,
+    CondaContext,
+    CondaTarget,
+    install_conda,
+    installed_conda_targets,
 )
 from galaxy.tool_util.deps.requirements import ToolRequirement
 from galaxy.tool_util.deps.resolvers.conda import CondaDependencyResolver
@@ -51,11 +55,30 @@ def test_against_conda_prefix_regression():
             use_path_exec=False,  # For the test ensure this is always a clean install
         )
         conda_context = resolver.conda_context
-        assert len(list(conda_util.installed_conda_targets(conda_context))) == 0
+        assert len(list(installed_conda_targets(conda_context))) == 0
         req = ToolRequirement(name="samtools", version="0.1.16", type="package")
         dependency = resolver.resolve(req, job_directory=job_dir)
         assert dependency.shell_commands() is not None  # install should not fail anymore
-        installed_targets = list(conda_util.installed_conda_targets(conda_context))
+        installed_targets = list(installed_conda_targets(conda_context))
         assert len(installed_targets) > 0
     finally:
         shutil.rmtree(base_path)
+
+
+@external_dependency_management
+def test_best_search_result(tmp_path) -> None:
+    conda_context = CondaContext(
+        conda_prefix=os.path.join(tmp_path, "_conda"), condarc_override=os.path.join(tmp_path, "_condarc")
+    )
+    install_conda(conda_context)
+    (hit, exact) = best_search_result(CondaTarget("samtools"), conda_context)
+    assert hit["name"] == "samtools"
+    assert exact is True
+    (hit, exact) = best_search_result(CondaTarget("samtools", version="1.3.1"), conda_context)
+    assert hit["name"] == "samtools"
+    assert hit["version"] == "1.3.1"
+    assert exact is True
+    # Search non-existent version
+    (hit, exact) = best_search_result(CondaTarget("samtools", version="1.16"), conda_context)
+    assert hit["name"] == "samtools"
+    assert exact is False