Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/cpan/tests/test_lister.py
# Copyright (C) 2022 The Software Heritage developers | # Copyright (C) 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 collections import defaultdict | from collections import defaultdict | ||||
from itertools import chain | from itertools import chain | ||||
import json | import json | ||||
from pathlib import Path | from pathlib import Path | ||||
import pytest | import pytest | ||||
from swh.lister.cpan.lister import CpanLister | from swh.lister.cpan.lister import CpanLister, get_module_version | ||||
@pytest.fixture | @pytest.fixture | ||||
def release_search_response(datadir): | def release_search_response(datadir): | ||||
return json.loads( | return json.loads( | ||||
Path(datadir, "https_fastapi.metacpan.org", "v1_release__search").read_bytes() | Path(datadir, "https_fastapi.metacpan.org", "v1_release__search").read_bytes() | ||||
) | ) | ||||
@pytest.fixture | def release_scroll_response(datadir, page): | ||||
def release_scroll_first_response(datadir): | |||||
return json.loads( | return json.loads( | ||||
Path(datadir, "https_fastapi.metacpan.org", "v1__search_scroll").read_bytes() | Path( | ||||
datadir, "https_fastapi.metacpan.org", f"v1__search_scroll_page{page}" | |||||
).read_bytes() | |||||
) | ) | ||||
@pytest.fixture | |||||
def release_scroll_first_response(datadir): | |||||
return release_scroll_response(datadir, page=1) | |||||
@pytest.fixture | |||||
def release_scroll_second_response(datadir): | |||||
return release_scroll_response(datadir, page=2) | |||||
@pytest.fixture | |||||
def release_scroll_third_response(datadir): | |||||
return release_scroll_response(datadir, page=3) | |||||
@pytest.fixture | |||||
def release_scroll_fourth_response(datadir): | |||||
return release_scroll_response(datadir, page=4) | |||||
@pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
def mock_network_requests( | def mock_network_requests( | ||||
requests_mock, release_search_response, release_scroll_first_response | requests_mock, | ||||
release_search_response, | |||||
release_scroll_first_response, | |||||
release_scroll_second_response, | |||||
release_scroll_third_response, | |||||
release_scroll_fourth_response, | |||||
): | ): | ||||
requests_mock.get( | requests_mock.get( | ||||
"https://fastapi.metacpan.org/v1/release/_search", | "https://fastapi.metacpan.org/v1/release/_search", | ||||
json=release_search_response, | json=release_search_response, | ||||
) | ) | ||||
requests_mock.get( | requests_mock.get( | ||||
"https://fastapi.metacpan.org/v1/_search/scroll", | "https://fastapi.metacpan.org/v1/_search/scroll", | ||||
[ | [ | ||||
{ | { | ||||
"json": release_scroll_first_response, | "json": release_scroll_first_response, | ||||
}, | }, | ||||
{ | |||||
"json": release_scroll_second_response, | |||||
}, | |||||
{ | |||||
"json": release_scroll_third_response, | |||||
}, | |||||
{ | |||||
"json": release_scroll_fourth_response, | |||||
}, | |||||
{"json": {"hits": {"hits": []}, "_scroll_id": ""}}, | {"json": {"hits": {"hits": []}, "_scroll_id": ""}}, | ||||
], | ], | ||||
) | ) | ||||
@pytest.mark.parametrize( | |||||
"module_name,module_version,release_name,expected_version", | |||||
[ | |||||
("Validator-Custom", "0.1207", "Validator-Custom-0.1207", "0.1207"), | |||||
("UDPServersAndClients", 0, "UDPServersAndClients", "0"), | |||||
("Compiler", 0, "Compiler-a1", "a1"), | |||||
("Call-Context", 0.01, "Call-Context-0.01", "0.01"), | |||||
], | |||||
) | |||||
def test_get_module_version( | |||||
module_name, module_version, release_name, expected_version | |||||
): | |||||
assert ( | |||||
get_module_version(module_name, module_version, release_name) | |||||
== expected_version | |||||
) | |||||
def test_cpan_lister( | def test_cpan_lister( | ||||
swh_scheduler, release_search_response, release_scroll_first_response | swh_scheduler, | ||||
release_search_response, | |||||
release_scroll_first_response, | |||||
release_scroll_second_response, | |||||
release_scroll_third_response, | |||||
release_scroll_fourth_response, | |||||
): | ): | ||||
lister = CpanLister(scheduler=swh_scheduler) | lister = CpanLister(scheduler=swh_scheduler) | ||||
res = lister.run() | res = lister.run() | ||||
expected_origins = set() | expected_origins = set() | ||||
expected_artifacts = defaultdict(list) | expected_artifacts = defaultdict(list) | ||||
expected_module_metadata = defaultdict(list) | expected_module_metadata = defaultdict(list) | ||||
for release in chain( | for release in chain( | ||||
release_search_response["hits"]["hits"], | release_search_response["hits"]["hits"], | ||||
release_scroll_first_response["hits"]["hits"], | release_scroll_first_response["hits"]["hits"], | ||||
release_scroll_second_response["hits"]["hits"], | |||||
release_scroll_third_response["hits"]["hits"], | |||||
release_scroll_fourth_response["hits"]["hits"], | |||||
): | ): | ||||
distribution = release["_source"]["distribution"] | distribution = release["_source"]["distribution"] | ||||
release_name = release["_source"]["name"] | release_name = release["_source"]["name"] | ||||
checksum_sha256 = release["_source"]["checksum_sha256"] | checksum_sha256 = release["_source"]["checksum_sha256"] | ||||
download_url = release["_source"]["download_url"] | download_url = release["_source"]["download_url"] | ||||
version = release["_source"]["version"] | version = release["_source"]["version"] | ||||
size = release["_source"]["stat"]["size"] | size = release["_source"]["stat"]["size"] | ||||
author = release["_source"]["author"] | author = release["_source"]["author"] | ||||
author_fullname = release["_source"]["metadata"]["author"][0] | author_fullname = release["_source"]["metadata"]["author"][0] | ||||
date = release["_source"]["date"] | date = release["_source"]["date"] | ||||
origin_url = f"https://metacpan.org/dist/{distribution}" | origin_url = f"https://metacpan.org/dist/{distribution}" | ||||
version = get_module_version(distribution, version, release_name) | |||||
expected_origins.add(origin_url) | expected_origins.add(origin_url) | ||||
expected_artifacts[origin_url].append( | expected_artifacts[origin_url].append( | ||||
{ | { | ||||
"url": download_url, | "url": download_url, | ||||
"filename": download_url.split("/")[-1], | "filename": download_url.split("/")[-1], | ||||
"version": version, | "version": version, | ||||
"length": size, | "length": size, | ||||
"checksums": {"sha256": checksum_sha256}, | "checksums": {"sha256": checksum_sha256}, | ||||
Show All 29 Lines |