Use anaconda API to find package download URL

Fix the following failing mulled unit test:

```
FAILED test/unit/tool_util/mulled/test_get_tests.py::test_find_anaconda_versions - AssertionError: assert '/bioconda/2pg_cartesian/1.0.1/download/linux-64/2pg_cartesian-1.0.1-0.tar.bz2' in []
```
This commit is contained in:
Nicola Soranzo
2025-11-25 15:54:41 +00:00
parent ef14b2bc55
commit 42fda3adcd
2 changed files with 50 additions and 41 deletions
+35 -31
View File
@@ -108,13 +108,6 @@ def get_anaconda_url(container, anaconda_channel="bioconda"):
return f"https://anaconda.org/{anaconda_channel}/{name[0]}/{name[1]}/download/linux-64/{'-'.join(name)}.tar.bz2"
def prepend_anaconda_url(url):
"""
Take a partial url and prepend 'https://anaconda.org'
"""
return f"https://anaconda.org{url}"
def get_test_from_anaconda(url: str) -> Optional[Dict[str, Any]]:
"""
Given the URL of an anaconda tarball, return tests
@@ -132,20 +125,26 @@ def get_test_from_anaconda(url: str) -> Optional[Dict[str, Any]]:
return None
def find_anaconda_versions(name, anaconda_channel="bioconda"):
def find_anaconda_download_url(
name: str, version: str, build: Optional[str] = None, anaconda_channel: str = "bioconda"
) -> Optional[str]:
"""
Find a list of available anaconda versions for a given container name
Find the anaconda download url for a given package.
"""
r = requests.get(
f"https://anaconda.org/{anaconda_channel}/{name}/files",
f"https://api.anaconda.org/package/{anaconda_channel}/{name}/files",
timeout=MULLED_SOCKET_TIMEOUT,
)
r.raise_for_status()
urls = []
for line in r.text.splitlines():
if "download/linux" in line:
urls.append(line.split('"')[1])
return urls
package_files = r.json()
for package_file in reversed(package_files):
if (
package_file["version"] == version
and (build is None or package_file["attrs"]["build"] == build)
and package_file["attrs"]["subdir"] in ["linux-64", "noarch"]
):
return f"https:{package_file['download_url']}"
return None
def open_recipe_file(file, recipes_path=None, github_repo="bioconda/bioconda-recipes"):
@@ -211,50 +210,55 @@ def deep_test_search(
"""
Look in bioconda-recipes repo as well as anaconda for the tests, checking in multiple possible locations. If no test is found for the specified version, search if other package versions have a test available.
"""
name = split_container_name(container)
name_tuple = split_container_name(container)
assert len(name_tuple) in (2, 3)
name = name_tuple[0]
version = name_tuple[1]
build = name_tuple[2] if len(name_tuple) == 3 else None
for f in [
(
get_commands_from_yaml,
open_recipe_file,
(f"recipes/{name[0]}/{name[1]}/meta.yaml", recipes_path, github_repo),
(f"recipes/{name}/{version}/meta.yaml", recipes_path, github_repo),
container,
),
(
get_run_test,
open_recipe_file,
(f"recipes/{name[0]}/{name[1]}/run_test.sh", recipes_path, github_repo),
(f"recipes/{name}/{version}/run_test.sh", recipes_path, github_repo),
container,
),
(
get_commands_from_yaml,
open_recipe_file,
(f"recipes/{name[0]}/meta.yaml", recipes_path, github_repo),
(f"recipes/{name}/meta.yaml", recipes_path, github_repo),
container,
),
(get_run_test, open_recipe_file, (f"recipes/{name[0]}/run_test.sh", recipes_path, github_repo), container),
(get_run_test, open_recipe_file, (f"recipes/{name}/run_test.sh", recipes_path, github_repo), container),
(get_test_from_anaconda, get_anaconda_url, (container, anaconda_channel), container),
]:
result = try_a_func(*f)
if result:
return result
versions = get_alternative_versions(f"recipes/{name[0]}", "meta.yaml", recipes_path, github_repo)
for version in versions:
result = try_a_func(get_commands_from_yaml, open_recipe_file, (version, recipes_path, github_repo), container)
alt_versions = get_alternative_versions(f"recipes/{name}", "meta.yaml", recipes_path, github_repo)
for alt_version in alt_versions:
result = try_a_func(
get_commands_from_yaml, open_recipe_file, (alt_version, recipes_path, github_repo), container
)
if result:
return result
versions = get_alternative_versions(f"recipes/{name[0]}", "run_test.sh", recipes_path, github_repo)
for version in versions:
result = try_a_func(get_run_test, open_recipe_file, (version, recipes_path, github_repo), container)
alt_versions = get_alternative_versions(f"recipes/{name}", "run_test.sh", recipes_path, github_repo)
for alt_version in alt_versions:
result = try_a_func(get_run_test, open_recipe_file, (alt_version, recipes_path, github_repo), container)
if result:
return result
versions = find_anaconda_versions(name[0], anaconda_channel)
for version in versions:
result = try_a_func(get_test_from_anaconda, prepend_anaconda_url, (version,), container)
if result:
return result
url = find_anaconda_download_url(name, version, build=build, anaconda_channel=anaconda_channel)
result = try_a_func(get_test_from_anaconda, lambda x: x, (url,), container)
if result:
return result
# if everything fails
return {"container": container}
+15 -10
View File
@@ -2,7 +2,7 @@ from unittest import SkipTest
from galaxy.tool_util.deps.mulled.get_tests import (
deep_test_search,
find_anaconda_versions,
find_anaconda_download_url,
get_alternative_versions,
get_anaconda_url,
get_commands_from_yaml,
@@ -11,7 +11,6 @@ from galaxy.tool_util.deps.mulled.get_tests import (
hashed_test_search,
main_test_search,
open_recipe_file,
prepend_anaconda_url,
)
from galaxy.util import smart_str
from ..util import external_dependency_management
@@ -49,11 +48,6 @@ def test_get_anaconda_url():
assert url == "https://anaconda.org/bioconda/samtools/1.7/download/linux-64/samtools-1.7-1.tar.bz2"
def test_prepend_anaconda_url():
url = prepend_anaconda_url("/bioconda/samtools/0.1.12/download/linux-64/samtools-0.1.12-2.tar.bz2")
assert url == "https://anaconda.org/bioconda/samtools/0.1.12/download/linux-64/samtools-0.1.12-2.tar.bz2"
@external_dependency_management
def test_get_test_from_anaconda():
# test old fashion tar.bz2 package
@@ -81,9 +75,20 @@ def test_get_test_from_anaconda():
@external_dependency_management
def test_find_anaconda_versions():
versions = find_anaconda_versions("2pg_cartesian")
assert "/bioconda/2pg_cartesian/1.0.1/download/linux-64/2pg_cartesian-1.0.1-0.tar.bz2" in versions
def test_find_anaconda_download_url():
download_url = find_anaconda_download_url("2pg_cartesian", "1.0.0")
assert download_url is None
download_url = find_anaconda_download_url("2pg_cartesian", "1.0.1")
assert download_url is not None
assert download_url.startswith(
"https://api.anaconda.org/download/bioconda/2pg_cartesian/1.0.1/linux-64/2pg_cartesian-1.0.1-"
)
download_url = find_anaconda_download_url("2pg_cartesian", "1.0.1", "0")
assert download_url is not None
assert (
download_url
== "https://api.anaconda.org/download/bioconda/2pg_cartesian/1.0.1/linux-64/2pg_cartesian-1.0.1-0.tar.bz2"
)
@external_dependency_management