Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/npm/tests/test_lister.py
# Copyright (C) 2018-2021 The Software Heritage developers | # Copyright (C) 2018-2022 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 | ||||
from itertools import chain | from itertools import chain | ||||
import json | import json | ||||
from pathlib import Path | from pathlib import Path | ||||
Show All 22 Lines | |||||
@pytest.fixture | @pytest.fixture | ||||
def npm_incremental_listing_page2(datadir): | def npm_incremental_listing_page2(datadir): | ||||
return json.loads(Path(datadir, "npm_incremental_page2.json").read_text()) | return json.loads(Path(datadir, "npm_incremental_page2.json").read_text()) | ||||
@pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
def retry_sleep_mock(mocker): | def retry_sleep_mock(mocker): | ||||
mocker.patch.object(NpmLister.page_request.retry, "sleep") | mocker.patch.object(NpmLister.http_request.retry, "sleep") | ||||
def _check_listed_npm_packages(lister, packages, scheduler_origins): | def _check_listed_npm_packages(lister, packages, scheduler_origins): | ||||
for package in packages: | for package in packages: | ||||
package_name = package["doc"]["name"] | package_name = package["doc"]["name"] | ||||
latest_version = package["doc"]["dist-tags"]["latest"] | latest_version = package["doc"]["dist-tags"]["latest"] | ||||
package_last_update = iso8601.parse_date(package["doc"]["time"][latest_version]) | package_last_update = iso8601.parse_date(package["doc"]["time"][latest_version]) | ||||
origin_url = lister.PACKAGE_URL_TEMPLATE.format(package_name=package_name) | origin_url = lister.PACKAGE_URL_TEMPLATE.format(package_name=package_name) | ||||
Show All 24 Lines | requests_mock.get( | ||||
lister.API_FULL_LISTING_URL, | lister.API_FULL_LISTING_URL, | ||||
[ | [ | ||||
{"json": npm_full_listing_page1}, | {"json": npm_full_listing_page1}, | ||||
{"json": npm_full_listing_page2}, | {"json": npm_full_listing_page2}, | ||||
], | ], | ||||
additional_matcher=_match_request, | additional_matcher=_match_request, | ||||
) | ) | ||||
spy_get = mocker.spy(lister.session, "get") | spy_request = mocker.spy(lister.session, "request") | ||||
stats = lister.run() | stats = lister.run() | ||||
assert stats.pages == 2 | assert stats.pages == 2 | ||||
assert stats.origins == page_size * stats.pages | assert stats.origins == page_size * stats.pages | ||||
spy_get.assert_has_calls( | spy_request.assert_has_calls( | ||||
[ | [ | ||||
mocker.call( | mocker.call( | ||||
"GET", | |||||
lister.API_FULL_LISTING_URL, | lister.API_FULL_LISTING_URL, | ||||
params=_url_params(page_size + 1, startkey='""'), | params=_url_params(page_size + 1, startkey='""'), | ||||
), | ), | ||||
mocker.call( | mocker.call( | ||||
"GET", | |||||
lister.API_FULL_LISTING_URL, | lister.API_FULL_LISTING_URL, | ||||
params=_url_params( | params=_url_params( | ||||
page_size + 1, | page_size + 1, | ||||
startkey=f'"{npm_full_listing_page1["rows"][-1]["id"]}"', | startkey=f'"{npm_full_listing_page1["rows"][-1]["id"]}"', | ||||
), | ), | ||||
), | ), | ||||
] | ] | ||||
) | ) | ||||
Show All 25 Lines | requests_mock.get( | ||||
[ | [ | ||||
{"json": npm_incremental_listing_page1}, | {"json": npm_incremental_listing_page1}, | ||||
{"json": npm_incremental_listing_page2}, | {"json": npm_incremental_listing_page2}, | ||||
{"json": {"results": []}}, | {"json": {"results": []}}, | ||||
], | ], | ||||
additional_matcher=_match_request, | additional_matcher=_match_request, | ||||
) | ) | ||||
spy_get = mocker.spy(lister.session, "get") | spy_request = mocker.spy(lister.session, "request") | ||||
assert lister.get_state_from_scheduler() == NpmListerState() | assert lister.get_state_from_scheduler() == NpmListerState() | ||||
stats = lister.run() | stats = lister.run() | ||||
assert stats.pages == 2 | assert stats.pages == 2 | ||||
assert stats.origins == page_size * stats.pages | assert stats.origins == page_size * stats.pages | ||||
last_seq = npm_incremental_listing_page2["results"][-1]["seq"] | last_seq = npm_incremental_listing_page2["results"][-1]["seq"] | ||||
spy_get.assert_has_calls( | spy_request.assert_has_calls( | ||||
[ | [ | ||||
mocker.call( | mocker.call( | ||||
"GET", | |||||
lister.API_INCREMENTAL_LISTING_URL, | lister.API_INCREMENTAL_LISTING_URL, | ||||
params=_url_params(page_size, since="0"), | params=_url_params(page_size, since="0"), | ||||
), | ), | ||||
mocker.call( | mocker.call( | ||||
"GET", | |||||
lister.API_INCREMENTAL_LISTING_URL, | lister.API_INCREMENTAL_LISTING_URL, | ||||
params=_url_params( | params=_url_params( | ||||
page_size, | page_size, | ||||
since=str(npm_incremental_listing_page1["results"][-1]["seq"]), | since=str(npm_incremental_listing_page1["results"][-1]["seq"]), | ||||
), | ), | ||||
), | ), | ||||
mocker.call( | mocker.call( | ||||
"GET", | |||||
lister.API_INCREMENTAL_LISTING_URL, | lister.API_INCREMENTAL_LISTING_URL, | ||||
params=_url_params(page_size, since=str(last_seq)), | params=_url_params(page_size, since=str(last_seq)), | ||||
), | ), | ||||
] | ] | ||||
) | ) | ||||
scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results | scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results | ||||
Show All 17 Lines | ): | ||||
"""Check incremental npm listing will restart from saved state""" | """Check incremental npm listing will restart from saved state""" | ||||
page_size = 2 | page_size = 2 | ||||
last_seq = 67 | last_seq = 67 | ||||
lister = NpmLister(scheduler=swh_scheduler, page_size=page_size, incremental=True) | lister = NpmLister(scheduler=swh_scheduler, page_size=page_size, incremental=True) | ||||
lister.state = NpmListerState(last_seq=last_seq) | lister.state = NpmListerState(last_seq=last_seq) | ||||
requests_mock.get(lister.API_INCREMENTAL_LISTING_URL, json={"results": []}) | requests_mock.get(lister.API_INCREMENTAL_LISTING_URL, json={"results": []}) | ||||
spy_get = mocker.spy(lister.session, "get") | spy_request = mocker.spy(lister.session, "request") | ||||
lister.run() | lister.run() | ||||
spy_get.assert_called_with( | spy_request.assert_called_with( | ||||
"GET", | |||||
lister.API_INCREMENTAL_LISTING_URL, | lister.API_INCREMENTAL_LISTING_URL, | ||||
params=_url_params(page_size, since=str(last_seq)), | params=_url_params(page_size, since=str(last_seq)), | ||||
) | ) | ||||
def test_npm_lister_http_error( | def test_npm_lister_http_error( | ||||
swh_scheduler, | swh_scheduler, | ||||
requests_mock, | requests_mock, | ||||
mocker, | mocker, | ||||
): | ): | ||||
lister = NpmLister(scheduler=swh_scheduler) | lister = NpmLister(scheduler=swh_scheduler) | ||||
requests_mock.get(lister.API_FULL_LISTING_URL, status_code=500) | requests_mock.get(lister.API_FULL_LISTING_URL, status_code=500) | ||||
with pytest.raises(HTTPError): | with pytest.raises(HTTPError): | ||||
lister.run() | lister.run() |