Differential D6393 Diff 23250 swh/lister/opam/tests/test_lister.py

Changeset View

Standalone View

swh/lister/opam/tests/test_lister.py

	# Copyright (C) 2021 The Software Heritage developers			# Copyright (C) 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 io			import io
				import os
				from tempfile import mkdtemp
	from unittest.mock import MagicMock			from unittest.mock import MagicMock

	from swh.lister.opam.lister import OpamLister			import pytest

				from swh.lister.opam.lister import OpamLister, opam_init

	def test_urls(swh_scheduler, mocker):			module_name = "swh.lister.opam.lister"

	instance_url = "https://opam.ocaml.org"

	lister = OpamLister(swh_scheduler, url=instance_url, instance="opam")			@pytest.fixture
				def mock_opam(mocker):
				"""Fixture to bypass the actual opam calls within the test context.

				"""
				# inhibits the real `subprocess.call` which prepares the required internal opam
				# state
				mock_init = mocker.patch(f"{module_name}.call", return_value=None)
				# replaces the real Popen with a fake one (list origins command)
	mocked_popen = MagicMock()			mocked_popen = MagicMock()
	mocked_popen.stdout = io.BytesIO(b"bar\nbaz\nfoo\n")			mocked_popen.stdout = io.BytesIO(b"bar\nbaz\nfoo\n")
				mock_open = mocker.patch(f"{module_name}.Popen", return_value=mocked_popen)
				return mock_init, mock_open


				def test_mock_init_repository_init(mock_opam, tmp_path, datadir):
				"""Initializing opam root directory with an instance should be ok

				"""
				mock_init, mock_popen = mock_opam

				instance = "fake"
				instance_url = f"file://{datadir}/{instance}"
				opam_root = str(tmp_path / "test-opam")
				assert not os.path.exists(opam_root)

				# This will initialize an opam directory with the instance
				opam_init(opam_root, instance, instance_url, {})

				assert mock_init.called


				def test_mock_init_repository_update(mock_opam, tmp_path, datadir):
				"""Updating opam root directory with another instance should be ok

				"""
				mock_init, mock_popen = mock_opam

				instance = "fake_opam_repo"
				instance_url = f"file://{datadir}/{instance}"
				opam_root = str(tmp_path / "test-opam")

	# replaces the real Popen with a fake one			os.makedirs(opam_root, exist_ok=True)
	mocker.patch("swh.lister.opam.lister.Popen", return_value=mocked_popen)			with open(os.path.join(opam_root, "opam"), "w") as f:
				f.write("one file to avoid empty folder")

				assert os.path.exists(opam_root)
				assert os.listdir(opam_root) == ["opam"] # not empty
				# This will update the repository opam with another instance
				opam_init(opam_root, instance, instance_url, {})

				assert mock_init.called


				def test_lister_opam_optional_instance(swh_scheduler):
				"""Instance name should be optional and default to be built out of the netloc."""
				netloc = "opam.ocaml.org"
				instance_url = f"https://{netloc}"

				lister = OpamLister(swh_scheduler, url=instance_url,)
				assert lister.instance == netloc
				assert lister.opam_root == "/tmp/opam/"


				def test_urls(swh_scheduler, mock_opam):
				mock_init, mock_popen = mock_opam
				instance_url = "https://opam.ocaml.org"
				tmp_folder = mkdtemp(prefix="swh_opam_lister")

				lister = OpamLister(
				swh_scheduler, url=instance_url, instance="opam", opam_root=tmp_folder,
				)
				assert lister.instance == "opam"
				assert lister.opam_root == tmp_folder

	# call the lister and get all listed origins urls			# call the lister and get all listed origins urls
	stats = lister.run()			stats = lister.run()

				assert mock_init.called
				assert mock_popen.called

	assert stats.pages == 3			assert stats.pages == 3
	assert stats.origins == 3			assert stats.origins == 3

	scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results			scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results

	expected_urls = [			expected_urls = [
	f"opam+{instance_url}/packages/bar/",			f"opam+{instance_url}/packages/bar/",
	f"opam+{instance_url}/packages/baz/",			f"opam+{instance_url}/packages/baz/",
	f"opam+{instance_url}/packages/foo/",			f"opam+{instance_url}/packages/foo/",
	]			]

	result_urls = [origin.url for origin in scheduler_origins]			result_urls = [origin.url for origin in scheduler_origins]

	assert expected_urls == result_urls			assert expected_urls == result_urls


	def test_opam_binary(datadir, swh_scheduler):			def test_opam_binary(datadir, swh_scheduler):
				instance_url = f"file://{datadir}/fake_opam_repo"

				lister = OpamLister(
				swh_scheduler,
				url=instance_url,
				instance="fake",
				opam_root=mkdtemp(prefix="swh_opam_lister"),
				)

				stats = lister.run()

				assert stats.pages == 4
				assert stats.origins == 4

				scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results

				expected_urls = [
				f"opam+{instance_url}/packages/agrid/",
				f"opam+{instance_url}/packages/calculon/",
				f"opam+{instance_url}/packages/directories/",
				f"opam+{instance_url}/packages/ocb/",
				]

				result_urls = [origin.url for origin in scheduler_origins]

				assert expected_urls == result_urls


				def test_opam_multi_instance(datadir, swh_scheduler):
	instance_url = f"file://{datadir}/fake_opam_repo"			instance_url = f"file://{datadir}/fake_opam_repo"

	lister = OpamLister(swh_scheduler, url=instance_url, instance="fake")			lister = OpamLister(
				swh_scheduler,
				url=instance_url,
				instance="fake",
				opam_root=mkdtemp(prefix="swh_opam_lister"),
				)

	stats = lister.run()			stats = lister.run()

	assert stats.pages == 4			assert stats.pages == 4
	assert stats.origins == 4			assert stats.origins == 4

	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 10 Lines