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 unittest.mock import MagicMock | ||||
from urllib.error import URLError | from urllib.error import URLError | ||||
from urllib.parse import quote | from urllib.parse import quote | ||||
import pytest | import pytest | ||||
from requests.exceptions import HTTPError | from requests.exceptions import HTTPError | ||||
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 | ||||
def test_version_generation(): | def test_version_generation(): | ||||
assert ( | assert ( | ||||
swh.loader.package.__version__ != "devel" | swh.loader.package.__version__ != "devel" | ||||
), "Make sure swh.loader.core is installed (e.g. pip install -e .)" | ), "Make sure swh.loader.core is installed (e.g. pip install -e .)" | ||||
▲ Show 20 Lines • Show All 183 Lines • ▼ Show 20 Lines | |||||
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( | ||||
HTTPError, match=f"{status_code} Client Error: None for url: {url}" | |||||
): | |||||
api_info(url) | api_info(url) | ||||
assert e0.value.args[0] == "Fail to query '%s'. Reason: %s" % (url, status_code) | |||||
def test_api_info(requests_mock): | def test_api_info(requests_mock): | ||||
"""Fetching json info from pypi project should be ok""" | """Fetching json info from pypi project should be ok""" | ||||
url = "https://pypi.org/pypi/requests/json" | url = "https://pypi.org/pypi/requests/json" | ||||
requests_mock.get(url, text='{"version": "0.0.1"}') | requests_mock.get(url, text='{"version": "0.0.1"}') | ||||
actual_info = json.loads(api_info(url)) | actual_info = json.loads(api_info(url)) | ||||
assert actual_info == { | assert actual_info == { | ||||
"version": "0.0.1", | "version": "0.0.1", | ||||
Show All 37 Lines | def test_download_retry_reraise(mocker, requests_mock, tmp_path): | ||||
requests_mock.get( | requests_mock.get( | ||||
url, | url, | ||||
[{"status_code": 429}] * 5, | [{"status_code": 429}] * 5, | ||||
) | ) | ||||
with pytest.raises(HTTPError): | with pytest.raises(HTTPError): | ||||
_check_download_ok(url, dest=str(tmp_path)) | _check_download_ok(url, dest=str(tmp_path)) | ||||
@pytest.fixture(autouse=True) | |||||
def mock_api_info_retry_sleep(mocker): | |||||
mocker.patch.object(api_info.retry, "sleep") | |||||
def test_api_info_retry(mocker, requests_mock, tmp_path): | |||||
url = "https://example.org/api/endpoint" | |||||
json_data = {"foo": "bar"} | |||||
requests_mock.get( | |||||
url, | |||||
[ | |||||
{"status_code": 429}, | |||||
{"status_code": 429}, | |||||
{ | |||||
"json": json_data, | |||||
"status_code": 200, | |||||
}, | |||||
], | |||||
) | |||||
assert json.loads(api_info(url)) == json_data | |||||
def test_api_info_retry_reraise(mocker, requests_mock, tmp_path): | |||||
url = "https://example.org/api/endpoint" | |||||
requests_mock.get( | |||||
url, | |||||
[{"status_code": 429}] * 5, | |||||
) | |||||
with pytest.raises(HTTPError, match=f"429 Client Error: None for url: {url}"): | |||||
api_info(url) |