Mock quay.io API calls, pass install=False to not pull docker images

This commit is contained in:
mvdbeek
2022-08-29 16:38:31 +02:00
parent c4f576ccf3
commit 60cda23970
2 changed files with 24 additions and 16 deletions
+19 -14
View File
@@ -59,6 +59,24 @@ def quay_repository(namespace, pkg_name, session=None):
return data
def _get_namespace(namespace):
next_page = None
repo_names = []
repos_headers = {"Accept-encoding": "gzip", "Accept": "application/json"}
while True:
repos_parameters = {"public": "true", "namespace": namespace, "next_page": next_page}
repos_response = requests.get(
QUAY_REPOSITORY_API_ENDPOINT, headers=repos_headers, params=repos_parameters, timeout=MULLED_SOCKET_TIMEOUT
)
repos_response_json = repos_response.json()
repos = repos_response_json["repositories"]
repo_names += [r["name"] for r in repos]
next_page = repos_response_json.get("next_page")
if not next_page:
break
return repo_names
def _namespace_has_repo_name(namespace, repo_name, resolution_cache):
"""
Get all quay containers in the biocontainers repo
@@ -74,20 +92,7 @@ def _namespace_has_repo_name(namespace, repo_name, resolution_cache):
except KeyError:
# mulled_resolution_cache may be beaker CacheManager instance, which raises KeyError if key is not present on `.get`
pass
next_page = None
repo_names = []
repos_headers = {"Accept-encoding": "gzip", "Accept": "application/json"}
while True:
repos_parameters = {"public": "true", "namespace": namespace, "next_page": next_page}
repos_response = requests.get(
QUAY_REPOSITORY_API_ENDPOINT, headers=repos_headers, params=repos_parameters, timeout=MULLED_SOCKET_TIMEOUT
)
repos_response_json = repos_response.json()
repos = repos_response_json["repositories"]
repo_names += [r["name"] for r in repos]
next_page = repos_response_json.get("next_page")
if not next_page:
break
repo_names = _get_namespace(namespace)
if resolution_cache is not None:
resolution_cache[cache_key] = repo_names
return repo_name in repo_names
@@ -29,10 +29,13 @@ def container_registry():
return ContainerRegistry(app_info)
def test_container_registry(container_registry):
def test_container_registry(container_registry, mocker):
mocker.patch('galaxy.tool_util.deps.mulled.util._get_namespace', return_value=["samtools"])
tool_info = ToolInfo(requirements=[ToolRequirement(name="samtools", version="1.10", type="package")])
container_description = container_registry.find_best_container_description(
[DOCKER_CONTAINER_TYPE], tool_info
[DOCKER_CONTAINER_TYPE],
tool_info,
install=False,
)
assert container_description.type == "docker"
assert "samtools:1.10" in container_description.identifier