542 lines
19 KiB
Python
542 lines
19 KiB
Python
|
# Copyright (c) 2018 The Pooch Developers.
|
||
|
|
# Distributed under the terms of the BSD 3-Clause License.
|
||
|
|
# SPDX-License-Identifier: BSD-3-Clause
|
||
|
|
#
|
||
|
|
# This code is part of the Fatiando a Terra project (https://www.fatiando.org)
|
||
|
|
#
|
||
|
|
"""
|
||
|
|
Test the downloader classes and functions separately from the Pooch core.
|
||
|
|
"""
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
from tempfile import TemporaryDirectory
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
try:
|
||
|
|
import tqdm
|
||
|
|
except ImportError:
|
||
|
|
tqdm = None
|
||
|
|
|
||
|
|
try:
|
||
|
|
import paramiko
|
||
|
|
except ImportError:
|
||
|
|
paramiko = None
|
||
|
|
|
||
|
|
from .. import Pooch
|
||
|
|
from ..downloaders import (
|
||
|
|
HTTPDownloader,
|
||
|
|
FTPDownloader,
|
||
|
|
SFTPDownloader,
|
||
|
|
DOIDownloader,
|
||
|
|
choose_downloader,
|
||
|
|
FigshareRepository,
|
||
|
|
ZenodoRepository,
|
||
|
|
DataverseRepository,
|
||
|
|
doi_to_url,
|
||
|
|
)
|
||
|
|
from ..processors import Unzip
|
||
|
|
from .utils import (
|
||
|
|
pooch_test_url,
|
||
|
|
check_large_data,
|
||
|
|
check_tiny_data,
|
||
|
|
data_over_ftp,
|
||
|
|
pooch_test_figshare_url,
|
||
|
|
pooch_test_zenodo_url,
|
||
|
|
pooch_test_zenodo_with_slash_url,
|
||
|
|
pooch_test_dataverse_url,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
BASEURL = pooch_test_url()
|
||
|
|
FIGSHAREURL = pooch_test_figshare_url()
|
||
|
|
ZENODOURL = pooch_test_zenodo_url()
|
||
|
|
ZENODOURL_W_SLASH = pooch_test_zenodo_with_slash_url()
|
||
|
|
DATAVERSEURL = pooch_test_dataverse_url()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.skipif(tqdm is None, reason="requires tqdm")
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"url",
|
||
|
|
[
|
||
|
|
BASEURL + "tiny-data.txt", # HTTPDownloader
|
||
|
|
FIGSHAREURL, # DOIDownloader
|
||
|
|
],
|
||
|
|
)
|
||
|
|
def test_progressbar_kwarg_passed(url):
|
||
|
|
"""The progressbar keyword argument must pass through choose_downloader"""
|
||
|
|
downloader = choose_downloader(url, progressbar=True)
|
||
|
|
assert downloader.progressbar is True
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.skipif(paramiko is None, reason="requires paramiko")
|
||
|
|
def test_progressbar_kwarg_passed_sftp():
|
||
|
|
"""The progressbar keyword argument must pass through choose_downloader"""
|
||
|
|
url = "sftp://test.rebex.net/pub/example/pocketftp.png"
|
||
|
|
downloader = choose_downloader(url, progressbar=True)
|
||
|
|
assert downloader.progressbar is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_unsupported_protocol():
|
||
|
|
"Should raise ValueError when protocol is not supported"
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
choose_downloader("httpup://some-invalid-url.com")
|
||
|
|
# Simulate the DOI format
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
choose_downloader("doii:XXX/XXX/file")
|
||
|
|
|
||
|
|
|
||
|
|
def test_invalid_doi_repository():
|
||
|
|
"Should fail if data repository is not supported"
|
||
|
|
with pytest.raises(ValueError) as exc:
|
||
|
|
# Use the DOI of the Pooch paper in JOSS (not a data repository)
|
||
|
|
DOIDownloader()(
|
||
|
|
url="doi:10.21105/joss.01943/file_name.txt", output_file=None, pooch=None
|
||
|
|
)
|
||
|
|
assert "Invalid data repository 'joss.theoj.org'" in str(exc.value)
|
||
|
|
|
||
|
|
|
||
|
|
def test_doi_url_not_found():
|
||
|
|
"Should fail if the DOI is not found"
|
||
|
|
with pytest.raises(ValueError) as exc:
|
||
|
|
doi_to_url(doi="NOTAREALDOI")
|
||
|
|
assert "Is the DOI correct?" in str(exc.value)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"repository,doi",
|
||
|
|
[
|
||
|
|
(FigshareRepository, "10.6084/m9.figshare.14763051.v1"),
|
||
|
|
(ZenodoRepository, "10.5281/zenodo.4924875"),
|
||
|
|
(DataverseRepository, "10.11588/data/TKCFEF"),
|
||
|
|
],
|
||
|
|
ids=["figshare", "zenodo", "dataverse"],
|
||
|
|
)
|
||
|
|
def test_figshare_url_file_not_found(repository, doi):
|
||
|
|
"Should fail if the file is not found in the archive"
|
||
|
|
with pytest.raises(ValueError) as exc:
|
||
|
|
url = doi_to_url(doi)
|
||
|
|
repo = repository.initialize(doi, url)
|
||
|
|
repo.download_url(file_name="bla.txt")
|
||
|
|
assert "File 'bla.txt' not found" in str(exc.value)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"url",
|
||
|
|
[FIGSHAREURL, ZENODOURL, DATAVERSEURL],
|
||
|
|
ids=["figshare", "zenodo", "dataverse"],
|
||
|
|
)
|
||
|
|
def test_doi_downloader(url):
|
||
|
|
"Test the DOI downloader"
|
||
|
|
# Use the test data we have on the repository
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
downloader = DOIDownloader()
|
||
|
|
outfile = os.path.join(local_store, "tiny-data.txt")
|
||
|
|
downloader(url + "tiny-data.txt", outfile, None)
|
||
|
|
check_tiny_data(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
def test_zenodo_downloader_with_slash_in_fname():
|
||
|
|
"""
|
||
|
|
Test the Zenodo downloader when the path contains a forward slash
|
||
|
|
|
||
|
|
Related to issue #336
|
||
|
|
"""
|
||
|
|
# Use the test data we have on the repository
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
base_url = ZENODOURL_W_SLASH + "santisoler/pooch-test-data-v1.zip"
|
||
|
|
downloader = DOIDownloader()
|
||
|
|
outfile = os.path.join(local_store, "test-data.zip")
|
||
|
|
downloader(base_url, outfile, None)
|
||
|
|
# unpack the downloaded zip file so we can check the integrity of
|
||
|
|
# tiny-data.txt
|
||
|
|
fnames = Unzip()(outfile, action="download", pooch=None)
|
||
|
|
(fname,) = [f for f in fnames if "tiny-data.txt" in f]
|
||
|
|
check_tiny_data(fname)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
def test_figshare_unspecified_version():
|
||
|
|
"""
|
||
|
|
Test if passing a Figshare url without a version warns about it, but still
|
||
|
|
downloads it.
|
||
|
|
"""
|
||
|
|
url = FIGSHAREURL
|
||
|
|
# Remove the last bits of the doi, where the version is specified and
|
||
|
|
url = url[: url.rindex(".")] + "/"
|
||
|
|
# Create expected warning message
|
||
|
|
doi = url[4:-1]
|
||
|
|
warning_msg = f"The Figshare DOI '{doi}' doesn't specify which version of "
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
downloader = DOIDownloader()
|
||
|
|
outfile = os.path.join(local_store, "tiny-data.txt")
|
||
|
|
with pytest.warns(UserWarning, match=warning_msg):
|
||
|
|
downloader(url + "tiny-data.txt", outfile, None)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"version, missing, present",
|
||
|
|
[
|
||
|
|
(
|
||
|
|
1,
|
||
|
|
"LC08_L2SP_218074_20190114_20200829_02_T1-cropped.tar.gz",
|
||
|
|
"cropped-before.tar.gz",
|
||
|
|
),
|
||
|
|
(
|
||
|
|
2,
|
||
|
|
"cropped-before.tar.gz",
|
||
|
|
"LC08_L2SP_218074_20190114_20200829_02_T1-cropped.tar.gz",
|
||
|
|
),
|
||
|
|
],
|
||
|
|
)
|
||
|
|
def test_figshare_data_repository_versions(version, missing, present):
|
||
|
|
"""
|
||
|
|
Test if setting the version in Figshare DOI works as expected
|
||
|
|
"""
|
||
|
|
# Use a Figshare repo as example (we won't download files from it since
|
||
|
|
# they are too big)
|
||
|
|
doi = f"10.6084/m9.figshare.21665630.v{version}"
|
||
|
|
url = f"https://doi.org/{doi}/"
|
||
|
|
figshare = FigshareRepository(doi, url)
|
||
|
|
filenames = [item["name"] for item in figshare.api_response]
|
||
|
|
assert present in filenames
|
||
|
|
assert missing not in filenames
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
def test_ftp_downloader(ftpserver):
|
||
|
|
"Test ftp downloader"
|
||
|
|
with data_over_ftp(ftpserver, "tiny-data.txt") as url:
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
downloader = FTPDownloader(port=ftpserver.server_port)
|
||
|
|
outfile = os.path.join(local_store, "tiny-data.txt")
|
||
|
|
downloader(url, outfile, None)
|
||
|
|
check_tiny_data(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
@pytest.mark.skipif(paramiko is None, reason="requires paramiko to run SFTP")
|
||
|
|
def test_sftp_downloader():
|
||
|
|
"Test sftp downloader"
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
downloader = SFTPDownloader(username="demo", password="password")
|
||
|
|
url = "sftp://test.rebex.net/pub/example/pocketftp.png"
|
||
|
|
outfile = os.path.join(local_store, "pocketftp.png")
|
||
|
|
downloader(url, outfile, None)
|
||
|
|
assert os.path.exists(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
@pytest.mark.skipif(paramiko is None, reason="requires paramiko to run SFTP")
|
||
|
|
def test_sftp_downloader_fail_if_file_object():
|
||
|
|
"Downloader should fail when a file object rather than string is passed"
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
downloader = SFTPDownloader(username="demo", password="password")
|
||
|
|
url = "sftp://test.rebex.net/pub/example/pocketftp.png"
|
||
|
|
outfile = os.path.join(local_store, "pocketftp.png")
|
||
|
|
with open(outfile, "wb") as outfile_obj:
|
||
|
|
with pytest.raises(TypeError):
|
||
|
|
downloader(url, outfile_obj, None)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.skipif(paramiko is not None, reason="paramiko must be missing")
|
||
|
|
def test_sftp_downloader_fail_if_paramiko_missing():
|
||
|
|
"test must fail if paramiko is not installed"
|
||
|
|
with pytest.raises(ValueError) as exc:
|
||
|
|
SFTPDownloader()
|
||
|
|
assert "'paramiko'" in str(exc.value)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.skipif(tqdm is not None, reason="tqdm must be missing")
|
||
|
|
@pytest.mark.parametrize("downloader", [HTTPDownloader, FTPDownloader, SFTPDownloader])
|
||
|
|
def test_downloader_progressbar_fails(downloader):
|
||
|
|
"Make sure an error is raised if trying to use progressbar without tqdm"
|
||
|
|
with pytest.raises(ValueError) as exc:
|
||
|
|
downloader(progressbar=True)
|
||
|
|
assert "'tqdm'" in str(exc.value)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
@pytest.mark.skipif(tqdm is None, reason="requires tqdm")
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"url,downloader",
|
||
|
|
[(BASEURL, HTTPDownloader), (FIGSHAREURL, DOIDownloader)],
|
||
|
|
ids=["http", "figshare"],
|
||
|
|
)
|
||
|
|
def test_downloader_progressbar(url, downloader, capsys):
|
||
|
|
"Setup a downloader function that prints a progress bar for fetch"
|
||
|
|
download = downloader(progressbar=True)
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
fname = "tiny-data.txt"
|
||
|
|
url = url + fname
|
||
|
|
outfile = os.path.join(local_store, fname)
|
||
|
|
download(url, outfile, None)
|
||
|
|
# Read stderr and make sure the progress bar is printed only when told
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
printed = captured.err.split("\r")[-1].strip()
|
||
|
|
assert len(printed) == 79
|
||
|
|
if sys.platform == "win32":
|
||
|
|
progress = "100%|####################"
|
||
|
|
else:
|
||
|
|
progress = "100%|████████████████████"
|
||
|
|
# Bar size is not always the same so can't reliably test the whole bar.
|
||
|
|
assert printed[:25] == progress
|
||
|
|
# Check that the downloaded file has the right content
|
||
|
|
check_tiny_data(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
@pytest.mark.skipif(tqdm is None, reason="requires tqdm")
|
||
|
|
def test_downloader_progressbar_ftp(capsys, ftpserver):
|
||
|
|
"Setup an FTP downloader function that prints a progress bar for fetch"
|
||
|
|
with data_over_ftp(ftpserver, "tiny-data.txt") as url:
|
||
|
|
download = FTPDownloader(progressbar=True, port=ftpserver.server_port)
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
outfile = os.path.join(local_store, "tiny-data.txt")
|
||
|
|
download(url, outfile, None)
|
||
|
|
# Read stderr and make sure the progress bar is printed only when
|
||
|
|
# told
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
printed = captured.err.split("\r")[-1].strip()
|
||
|
|
assert len(printed) == 79
|
||
|
|
if sys.platform == "win32":
|
||
|
|
progress = "100%|####################"
|
||
|
|
else:
|
||
|
|
progress = "100%|████████████████████"
|
||
|
|
# Bar size is not always the same so can't reliably test the whole
|
||
|
|
# bar.
|
||
|
|
assert printed[:25] == progress
|
||
|
|
# Check that the file was actually downloaded
|
||
|
|
check_tiny_data(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
@pytest.mark.skipif(tqdm is None, reason="requires tqdm")
|
||
|
|
@pytest.mark.skipif(paramiko is None, reason="requires paramiko")
|
||
|
|
def test_downloader_progressbar_sftp(capsys):
|
||
|
|
"Setup an SFTP downloader function that prints a progress bar for fetch"
|
||
|
|
downloader = SFTPDownloader(progressbar=True, username="demo", password="password")
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
url = "sftp://test.rebex.net/pub/example/pocketftp.png"
|
||
|
|
outfile = os.path.join(local_store, "pocketftp.png")
|
||
|
|
downloader(url, outfile, None)
|
||
|
|
# Read stderr and make sure the progress bar is printed only when told
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
printed = captured.err.split("\r")[-1].strip()
|
||
|
|
assert len(printed) == 79
|
||
|
|
if sys.platform == "win32":
|
||
|
|
progress = "100%|####################"
|
||
|
|
else:
|
||
|
|
progress = "100%|████████████████████"
|
||
|
|
# Bar size is not always the same so can't reliably test the whole bar.
|
||
|
|
assert printed[:25] == progress
|
||
|
|
# Check that the file was actually downloaded
|
||
|
|
assert os.path.exists(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.network
|
||
|
|
def test_downloader_arbitrary_progressbar(capsys):
|
||
|
|
"Setup a downloader function with an arbitrary progress bar class."
|
||
|
|
|
||
|
|
class MinimalProgressDisplay:
|
||
|
|
"""A minimalist replacement for tqdm.tqdm"""
|
||
|
|
|
||
|
|
def __init__(self, total):
|
||
|
|
self.count = 0
|
||
|
|
self.total = total
|
||
|
|
|
||
|
|
def __repr__(self):
|
||
|
|
"""represent current completion"""
|
||
|
|
return str(self.count) + "/" + str(self.total)
|
||
|
|
|
||
|
|
def render(self):
|
||
|
|
"""print self.__repr__ to stderr"""
|
||
|
|
print(f"\r{self}", file=sys.stderr, end="")
|
||
|
|
|
||
|
|
def update(self, i):
|
||
|
|
"""modify completion and render"""
|
||
|
|
self.count = i
|
||
|
|
self.render()
|
||
|
|
|
||
|
|
def reset(self):
|
||
|
|
"""set counter to 0"""
|
||
|
|
self.count = 0
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def close():
|
||
|
|
"""print a new empty line"""
|
||
|
|
print("", file=sys.stderr)
|
||
|
|
|
||
|
|
pbar = MinimalProgressDisplay(total=None)
|
||
|
|
download = HTTPDownloader(progressbar=pbar)
|
||
|
|
with TemporaryDirectory() as local_store:
|
||
|
|
fname = "large-data.txt"
|
||
|
|
url = BASEURL + fname
|
||
|
|
outfile = os.path.join(local_store, "large-data.txt")
|
||
|
|
download(url, outfile, None)
|
||
|
|
# Read stderr and make sure the progress bar is printed only when told
|
||
|
|
captured = capsys.readouterr()
|
||
|
|
printed = captured.err.split("\r")[-1].strip()
|
||
|
|
|
||
|
|
progress = "336/336"
|
||
|
|
assert printed == progress
|
||
|
|
|
||
|
|
# Check that the downloaded file has the right content
|
||
|
|
check_large_data(outfile)
|
||
|
|
|
||
|
|
|
||
|
|
class TestZenodoAPISupport:
|
||
|
|
"""
|
||
|
|
Test support for different Zenodo APIs
|
||
|
|
"""
|
||
|
|
|
||
|
|
article_id = 123456
|
||
|
|
doi = f"10.0001/zenodo.{article_id}"
|
||
|
|
doi_url = f"https://doi.org/{doi}"
|
||
|
|
file_name = "my-file.zip"
|
||
|
|
file_url = (
|
||
|
|
"https://zenodo.org/api/files/513d7033-93a2-4eeb-821c-2fb0bbab0012/my-file.zip"
|
||
|
|
)
|
||
|
|
file_checksum = "2942bfabb3d05332b66eb128e0842cff"
|
||
|
|
|
||
|
|
legacy_api_response = {
|
||
|
|
"created": "2021-20-19T08:00:00.000000+00:00",
|
||
|
|
"modified": "2021-20-19T08:00:00.000000+00:00",
|
||
|
|
"id": article_id,
|
||
|
|
"doi": doi,
|
||
|
|
"doi_url": doi_url,
|
||
|
|
"files": [
|
||
|
|
{
|
||
|
|
"id": "513d7033-93a2-4eeb-821c-2fb0bbab0012",
|
||
|
|
"key": file_name,
|
||
|
|
"checksum": f"md5:{file_checksum}",
|
||
|
|
"links": {
|
||
|
|
"self": file_url,
|
||
|
|
},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
}
|
||
|
|
|
||
|
|
new_api_response = {
|
||
|
|
"created": "2021-20-19T08:00:00.000000+00:00",
|
||
|
|
"modified": "2021-20-19T08:00:00.000000+00:00",
|
||
|
|
"id": article_id,
|
||
|
|
"doi": doi,
|
||
|
|
"doi_url": doi_url,
|
||
|
|
"files": [
|
||
|
|
{
|
||
|
|
"id": "513d7033-93a2-4eeb-821c-2fb0bbab0012",
|
||
|
|
"filename": file_name,
|
||
|
|
"checksum": file_checksum,
|
||
|
|
"links": {
|
||
|
|
"self": file_url,
|
||
|
|
},
|
||
|
|
}
|
||
|
|
],
|
||
|
|
}
|
||
|
|
|
||
|
|
invalid_api_response = {
|
||
|
|
"created": "2021-20-19T08:00:00.000000+00:00",
|
||
|
|
"modified": "2021-20-19T08:00:00.000000+00:00",
|
||
|
|
"id": article_id,
|
||
|
|
"doi": doi,
|
||
|
|
"doi_url": doi_url,
|
||
|
|
"files": [
|
||
|
|
{
|
||
|
|
"id": "513d7033-93a2-4eeb-821c-2fb0bbab0012",
|
||
|
|
"filename": file_name,
|
||
|
|
"checksum": file_checksum,
|
||
|
|
"links": {
|
||
|
|
"self": file_url,
|
||
|
|
},
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"id": "513d7033-93a2-4eeb-821c-2fb0bbab0012",
|
||
|
|
"key": file_name,
|
||
|
|
"checksum": f"md5:{file_checksum}",
|
||
|
|
"links": {
|
||
|
|
"self": file_url,
|
||
|
|
},
|
||
|
|
},
|
||
|
|
],
|
||
|
|
}
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"api_version, api_response",
|
||
|
|
[
|
||
|
|
("legacy", legacy_api_response),
|
||
|
|
("new", new_api_response),
|
||
|
|
("invalid", invalid_api_response),
|
||
|
|
],
|
||
|
|
)
|
||
|
|
def test_api_version(self, httpserver, api_version, api_response):
|
||
|
|
"""
|
||
|
|
Test if the API version is correctly detected.
|
||
|
|
"""
|
||
|
|
# Create a local http server
|
||
|
|
httpserver.expect_request(f"/zenodo.{self.article_id}").respond_with_json(
|
||
|
|
api_response
|
||
|
|
)
|
||
|
|
# Create Zenodo downloader
|
||
|
|
downloader = ZenodoRepository(doi=self.doi, archive_url=self.doi_url)
|
||
|
|
# Override base url for the API of the downloader
|
||
|
|
downloader.base_api_url = httpserver.url_for("")
|
||
|
|
# Check if the API version is correctly identified
|
||
|
|
if api_version != "invalid":
|
||
|
|
assert downloader.api_version == api_version
|
||
|
|
else:
|
||
|
|
msg = "Couldn't determine the version of the Zenodo API"
|
||
|
|
with pytest.raises(ValueError, match=msg):
|
||
|
|
api_version = downloader.api_version
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"api_version, api_response",
|
||
|
|
[("legacy", legacy_api_response), ("new", new_api_response)],
|
||
|
|
)
|
||
|
|
def test_download_url(self, httpserver, api_version, api_response):
|
||
|
|
"""
|
||
|
|
Test if the download url is correct for each API version.
|
||
|
|
"""
|
||
|
|
# Create a local http server
|
||
|
|
httpserver.expect_request(f"/zenodo.{self.article_id}").respond_with_json(
|
||
|
|
api_response
|
||
|
|
)
|
||
|
|
# Create Zenodo downloader
|
||
|
|
downloader = ZenodoRepository(doi=self.doi, archive_url=self.doi_url)
|
||
|
|
# Override base url for the API of the downloader
|
||
|
|
downloader.base_api_url = httpserver.url_for("")
|
||
|
|
# Check if the download url is correct
|
||
|
|
download_url = downloader.download_url(file_name=self.file_name)
|
||
|
|
if api_version == "legacy":
|
||
|
|
assert download_url == self.file_url
|
||
|
|
else:
|
||
|
|
expected_url = (
|
||
|
|
"https://zenodo.org/records/"
|
||
|
|
f"{self.article_id}/files/{self.file_name}?download=1"
|
||
|
|
)
|
||
|
|
assert download_url == expected_url
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"api_response",
|
||
|
|
[legacy_api_response, new_api_response],
|
||
|
|
)
|
||
|
|
def test_populate_registry(self, httpserver, tmp_path, api_response):
|
||
|
|
"""
|
||
|
|
Test if population of registry is correctly done for each API version.
|
||
|
|
"""
|
||
|
|
# Create a local http server
|
||
|
|
httpserver.expect_request(f"/zenodo.{self.article_id}").respond_with_json(
|
||
|
|
api_response
|
||
|
|
)
|
||
|
|
# Create sample pooch object
|
||
|
|
puppy = Pooch(base_url="", path=tmp_path)
|
||
|
|
# Create Zenodo downloader
|
||
|
|
downloader = ZenodoRepository(doi=self.doi, archive_url=self.doi_url)
|
||
|
|
# Override base url for the API of the downloader
|
||
|
|
downloader.base_api_url = httpserver.url_for("")
|
||
|
|
# Populate registry
|
||
|
|
downloader.populate_registry(puppy)
|
||
|
|
assert puppy.registry == {self.file_name: f"md5:{self.file_checksum}"}
|