Differential D8648 Diff 31259 swh/lister/cpan/tests/test_lister.py

Changeset 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 28 Lines