Differential D6316 Diff 23156 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 tempfile import mkdtemp
from unittest.mock import MagicMock		from unittest.mock import MagicMock

import pytest		import pytest

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

module_name = "swh.lister.opam.lister"		module_name = "swh.lister.opam.lister"


@pytest.fixture		@pytest.fixture
def mock_opam(mocker):		def mock_opam(mocker):
"""Fixture to bypass the actual opam calls within the test context.		"""Fixture to bypass the actual opam calls within the test context.

"""		"""
# inhibits the real `subprocess.call` which prepares the required internal opam		# inhibits the real `subprocess.call` which prepares the required internal opam
# state		# state
mock_init = mocker.patch(f"{module_name}.call", return_value=None)		mock_init = mocker.patch(f"{module_name}.call", return_value=None)
# replaces the real Popen with a fake one (list origins command)		# 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)		mock_open = mocker.patch(f"{module_name}.Popen", return_value=mocked_popen)
return mock_init, mock_open		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")

		os.makedirs(opam_root, exist_ok=True)
		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):		def test_lister_opam_optional_instance(swh_scheduler):
"""Instance name should be optional and default to be built out of the netloc."""		"""Instance name should be optional and default to be built out of the netloc."""
netloc = "opam.ocaml.org"		netloc = "opam.ocaml.org"
instance_url = f"https://{netloc}"		instance_url = f"https://{netloc}"

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


def test_urls(swh_scheduler, mock_opam):		def test_urls(swh_scheduler, mock_opam):
mock_init, mock_popen = mock_opam		mock_init, mock_popen = mock_opam

instance_url = "https://opam.ocaml.org"		instance_url = "https://opam.ocaml.org"
		tmp_folder = mkdtemp(prefix="swh_opam_lister")

lister = OpamLister(		lister = OpamLister(
swh_scheduler,		swh_scheduler, url=instance_url, instance="opam", opam_root=tmp_folder,
url=instance_url,
instance="opam",
opam_root=mkdtemp(prefix="swh_opam_lister"),
)		)
assert lister.instance == "opam"		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_init.called
assert mock_popen.called		assert mock_popen.called

assert stats.pages == 3		assert stats.pages == 3
Show All 34 Lines	expected_urls = [
f"opam+{instance_url}/packages/calculon/",		f"opam+{instance_url}/packages/calculon/",
f"opam+{instance_url}/packages/directories/",		f"opam+{instance_url}/packages/directories/",
f"opam+{instance_url}/packages/ocb/",		f"opam+{instance_url}/packages/ocb/",
]		]

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_multi_instance(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