Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/tests/test_utils.py
# Copyright (C) 2019-2021 The Software Heritage developers | # Copyright (C) 2019-2021 The Software Heritage developers | ||||
# See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
# License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
# See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
import json | import json | ||||
import os | import os | ||||
from unittest.mock import MagicMock | |||||
from urllib.error import URLError | |||||
import pytest | import pytest | ||||
from swh.loader.exception import NotFound | from swh.loader.exception import NotFound | ||||
import swh.loader.package | import swh.loader.package | ||||
from swh.loader.package.utils import api_info, download, release_name | from swh.loader.package.utils import api_info, download, release_name | ||||
▲ Show 20 Lines • Show All 108 Lines • ▼ Show 20 Lines | for hash_algo in good.keys(): | ||||
wrong_hash, | wrong_hash, | ||||
good[hash_algo], | good[hash_algo], | ||||
) | ) | ||||
with pytest.raises(ValueError, match=expected_msg): | with pytest.raises(ValueError, match=expected_msg): | ||||
download(url, dest=str(tmp_path), hashes=expected_hashes) | download(url, dest=str(tmp_path), hashes=expected_hashes) | ||||
@pytest.mark.fs | |||||
def test_ftp_download_ok(tmp_path, mocker): | |||||
"""Download without issue should provide filename and hashes""" | |||||
filename = "requests-0.0.1.tar.gz" | |||||
url = "ftp://pypi.org/pypi/requests/%s" % filename | |||||
data = b"this is something" | |||||
cm = MagicMock() | |||||
cm.getstatus.return_value = 200 | |||||
cm.read.side_effect = [data, b""] | |||||
cm.__enter__.return_value = cm | |||||
mocker.patch("swh.loader.package.utils.urlopen").return_value = cm | |||||
actual_filepath, actual_hashes = download(url, dest=str(tmp_path)) | |||||
actual_filename = os.path.basename(actual_filepath) | |||||
assert actual_filename == filename | |||||
assert actual_hashes["length"] == len(data) | |||||
assert ( | |||||
actual_hashes["checksums"]["sha1"] == "fdd1ce606a904b08c816ba84f3125f2af44d92b2" | |||||
) # noqa | |||||
assert ( | |||||
actual_hashes["checksums"]["sha256"] | |||||
== "1d9224378d77925d612c9f926eb9fb92850e6551def8328011b6a972323298d5" | |||||
) | |||||
@pytest.mark.fs | |||||
def test_ftp_download_ko(tmp_path, mocker): | |||||
"""Download without issue should provide filename and hashes""" | |||||
filename = "requests-0.0.1.tar.gz" | |||||
url = "ftp://pypi.org/pypi/requests/%s" % filename | |||||
mocker.patch("swh.loader.package.utils.urlopen").side_effect = URLError("FTP error") | |||||
with pytest.raises(URLError): | |||||
download(url, dest=str(tmp_path)) | |||||
def test_api_info_failure(requests_mock): | def test_api_info_failure(requests_mock): | ||||
"""Failure to fetch info/release information should raise""" | """Failure to fetch info/release information should raise""" | ||||
url = "https://pypi.org/pypi/requests/json" | url = "https://pypi.org/pypi/requests/json" | ||||
status_code = 400 | status_code = 400 | ||||
requests_mock.get(url, status_code=status_code) | requests_mock.get(url, status_code=status_code) | ||||
with pytest.raises(NotFound) as e0: | with pytest.raises(NotFound) as e0: | ||||
api_info(url) | api_info(url) | ||||
Show All 20 Lines |