diff --git a/swh/lister/opam/lister.py b/swh/lister/opam/lister.py index 486bdc2..4ad510e 100644 --- a/swh/lister/opam/lister.py +++ b/swh/lister/opam/lister.py @@ -1,111 +1,141 @@ # Copyright (C) 2021 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import io import logging import os from subprocess import PIPE, Popen, call -from typing import Iterator, Optional +from typing import Any, Dict, Iterator, Optional from swh.lister.pattern import StatelessLister from swh.scheduler.interface import SchedulerInterface from swh.scheduler.model import ListedOrigin from ..pattern import CredentialsType logger = logging.getLogger(__name__) PageType = str class OpamLister(StatelessLister[PageType]): """ List all repositories hosted on an opam repository. On initialisation, we create an opam root, with no ocaml compiler (no switch) as we won't need it and it's costly. In this opam root, we add a single opam repository (url) and give it a name (instance). Then, to get pages, we just ask opam to list all the packages for our opam repository in our opam root. Args: url: base URL of an opam repository (for instance https://opam.ocaml.org) instance: string identifier for the listed repository """ # Part of the lister API, that identifies this lister LISTER_NAME = "opam" def __init__( self, scheduler: SchedulerInterface, url: str, instance: Optional[str] = None, credentials: CredentialsType = None, opam_root: str = "/tmp/opam/", ): super().__init__( scheduler=scheduler, credentials=credentials, url=url, instance=instance, ) self.env = os.environ.copy() # Opam root folder is initialized in the :meth:`get_pages` method as no # side-effect should happen in the constructor to ease instantiation - self.opamroot = os.path.join(opam_root, self.instance) + self.opam_root = opam_root def get_pages(self) -> Iterator[PageType]: - # Initialize the opam root directory with the opam instance data to list. - call( - [ - "opam", - "init", - "--reinit", - "--bare", - "--no-setup", - "--root", - self.opamroot, - self.instance, - self.url, - ], - env=self.env, - ) + # Initialize the opam root directory + opam_init(self.opam_root, self.instance, self.url, self.env) + # Actually list opam instance data proc = Popen( [ "opam", "list", "--all", "--no-switch", + "--safe", "--repos", self.instance, "--root", - self.opamroot, + self.opam_root, "--normalise", "--short", ], env=self.env, stdout=PIPE, ) if proc.stdout is not None: for line in io.TextIOWrapper(proc.stdout): yield line.rstrip("\n") def get_origins_from_page(self, page: PageType) -> Iterator[ListedOrigin]: """Convert a page of OpamLister repositories into a list of ListedOrigins""" assert self.lister_obj.id is not None # a page is just a package name url = f"opam+{self.url}/packages/{page}/" yield ListedOrigin( lister_id=self.lister_obj.id, visit_type="opam", url=url, last_update=None, extra_loader_arguments={ - "opam_root": self.opamroot, + "opam_root": self.opam_root, "opam_instance": self.instance, "opam_url": self.url, "opam_package": page, }, ) + + +def opam_init(opam_root: str, instance: str, url: str, env: Dict[str, Any]) -> None: + """Initialize an opam_root folder. + + Args: + opam_root: The opam root folder to initialize + instance: Name of the opam repository to add or initialize + url: The associated url of the opam repository to add or initialize + env: The global environment to use for the opam command. + + Returns: + None. + + """ + if not os.path.exists(opam_root) or not os.listdir(opam_root): + command = [ + "opam", + "init", + "--reinit", + "--bare", + "--no-setup", + "--root", + opam_root, + instance, + url, + ] + else: + # The repository exists and is populated, we just add another instance in the + # repository. If it's already setup, it's a noop + command = [ + "opam", + "repository", + "add", + "--root", + opam_root, + instance, + url, + ] + # Actually execute the command + call(command, env=env) diff --git a/swh/lister/opam/tests/test_lister.py b/swh/lister/opam/tests/test_lister.py index ffa281a..0079d7a 100644 --- a/swh/lister/opam/tests/test_lister.py +++ b/swh/lister/opam/tests/test_lister.py @@ -1,103 +1,170 @@ # Copyright (C) 2021 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import io +import os from tempfile import mkdtemp from unittest.mock import MagicMock import pytest -from swh.lister.opam.lister import OpamLister +from swh.lister.opam.lister import OpamLister, opam_init module_name = "swh.lister.opam.lister" @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.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") + + 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): """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.opamroot.endswith(lister.instance) + 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=mkdtemp(prefix="swh_opam_lister"), + 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 stats = lister.run() assert mock_init.called assert mock_popen.called assert stats.pages == 3 assert stats.origins == 3 scheduler_origins = swh_scheduler.get_listed_origins(lister.lister_obj.id).results expected_urls = [ f"opam+{instance_url}/packages/bar/", f"opam+{instance_url}/packages/baz/", f"opam+{instance_url}/packages/foo/", ] result_urls = [origin.url for origin in scheduler_origins] assert expected_urls == result_urls 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" + + 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