Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/git/tests/test_loader.py
| # Copyright (C) 2018-2021 The Software Heritage developers | # Copyright (C) 2018-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 | ||||
| from functools import partial | from functools import partial | ||||
| from http.server import HTTPServer, SimpleHTTPRequestHandler | from http.server import HTTPServer, SimpleHTTPRequestHandler | ||||
| import os | import os | ||||
| import subprocess | import subprocess | ||||
| from tempfile import SpooledTemporaryFile | from tempfile import SpooledTemporaryFile | ||||
| from threading import Thread | from threading import Thread | ||||
| from unittest.mock import MagicMock, call | |||||
| from dulwich.errors import GitProtocolError, NotGitRepository, ObjectFormatException | from dulwich.errors import GitProtocolError, NotGitRepository, ObjectFormatException | ||||
| from dulwich.porcelain import push | from dulwich.porcelain import push | ||||
| import dulwich.repo | import dulwich.repo | ||||
| import pytest | import pytest | ||||
| from swh.loader.git import dumb | from swh.loader.git import dumb | ||||
| from swh.loader.git.loader import GitLoader | from swh.loader.git.loader import GitLoader | ||||
| from swh.loader.git.tests.test_from_disk import FullGitLoaderTests | from swh.loader.git.tests.test_from_disk import FullGitLoaderTests | ||||
| from swh.loader.tests import ( | from swh.loader.tests import ( | ||||
| assert_last_visit_matches, | assert_last_visit_matches, | ||||
| get_stats, | get_stats, | ||||
| prepare_repository_from_archive, | prepare_repository_from_archive, | ||||
| ) | ) | ||||
| from swh.model.model import Origin | |||||
| class CommonGitLoaderNotFound: | class CommonGitLoaderNotFound: | ||||
| @pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
| def __inject_fixtures(self, mocker): | def __inject_fixtures(self, mocker): | ||||
| """Inject required fixtures in unittest.TestCase class""" | """Inject required fixtures in unittest.TestCase class""" | ||||
| self.mocker = mocker | self.mocker = mocker | ||||
| ▲ Show 20 Lines • Show All 72 Lines • ▼ Show 20 Lines | def init(self, swh_storage, datadir, tmp_path): | ||||
| archive_path, archive_name, tmp_path=tmp_path | archive_path, archive_name, tmp_path=tmp_path | ||||
| ) | ) | ||||
| self.destination_path = os.path.join(tmp_path, archive_name) | self.destination_path = os.path.join(tmp_path, archive_name) | ||||
| self.loader = GitLoader(swh_storage, self.repo_url) | self.loader = GitLoader(swh_storage, self.repo_url) | ||||
| self.repo = dulwich.repo.Repo(self.destination_path) | self.repo = dulwich.repo.Repo(self.destination_path) | ||||
| class TestGitLoader2(FullGitLoaderTests, CommonGitLoaderNotFound): | class TestGitLoader2(FullGitLoaderTests, CommonGitLoaderNotFound): | ||||
| """Mostly the same loading scenario but with a base-url different than the repo-url. | """Mostly the same loading scenario but with a ``parent_origin`` different from the | ||||
| ``origin``; as if the ``origin`` was a forge-fork of ``parent_origin``, detected | |||||
| by the metadata loader. | |||||
| To walk slightly different paths, the end result should stay the same. | To walk slightly different paths, the end result should stay the same. | ||||
| """ | """ | ||||
| @pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
| def init(self, swh_storage, datadir, tmp_path): | def init(self, swh_storage, datadir, tmp_path, mocker): | ||||
| archive_name = "testrepo" | archive_name = "testrepo" | ||||
| archive_path = os.path.join(datadir, f"{archive_name}.tgz") | archive_path = os.path.join(datadir, f"{archive_name}.tgz") | ||||
| tmp_path = str(tmp_path) | tmp_path = str(tmp_path) | ||||
| self.repo_url = prepare_repository_from_archive( | self.repo_url = prepare_repository_from_archive( | ||||
| archive_path, archive_name, tmp_path=tmp_path | archive_path, archive_name, tmp_path=tmp_path | ||||
| ) | ) | ||||
| self.destination_path = os.path.join(tmp_path, archive_name) | self.destination_path = os.path.join(tmp_path, archive_name) | ||||
| base_url = f"base://{self.repo_url}" | |||||
| self.loader = GitLoader(swh_storage, self.repo_url, base_url=base_url) | self.fetcher = MagicMock() | ||||
| self.fetcher.get_origin_metadata.return_value = [] | |||||
| self.fetcher.get_parent_origins.return_value = [ | |||||
| Origin(url=f"base://{self.repo_url}") | |||||
| ] | |||||
| self.fetcher_cls = MagicMock(return_value=self.fetcher) | |||||
| self.fetcher_cls.SUPPORTED_LISTERS = ["fake-lister"] | |||||
| mocker.patch( | |||||
| "swh.loader.core.metadata_fetchers._fetchers", | |||||
| return_value=[self.fetcher_cls], | |||||
| ) | |||||
| self.loader = GitLoader( | |||||
| MagicMock(wraps=swh_storage), | |||||
| self.repo_url, | |||||
| lister_name="fake-lister", | |||||
| lister_instance_name="", | |||||
| ) | |||||
| self.repo = dulwich.repo.Repo(self.destination_path) | self.repo = dulwich.repo.Repo(self.destination_path) | ||||
| def test_load_incremental(self): | |||||
| res = self.loader.load() | |||||
| assert res == {"status": "eventful"} | |||||
| self.fetcher_cls.assert_called_once_with( | |||||
| credentials={}, | |||||
| lister_name="fake-lister", | |||||
| lister_instance_name="", | |||||
| origin=Origin(url=self.repo_url), | |||||
| ) | |||||
| self.fetcher.get_parent_origins.assert_called_once_with() | |||||
| # First tries the same origin | |||||
| assert self.loader.storage.origin_visit_get_latest.mock_calls == [ | |||||
| call( | |||||
| self.repo_url, | |||||
| allowed_statuses=None, | |||||
| require_snapshot=True, | |||||
| type=None, | |||||
| ), | |||||
| # As it does not already have a snapshot, fall back to the parent origin | |||||
| call( | |||||
| f"base://{self.repo_url}", | |||||
| allowed_statuses=None, | |||||
| require_snapshot=True, | |||||
| type=None, | |||||
| ), | |||||
| ] | |||||
| self.fetcher.reset_mock() | |||||
| self.fetcher_cls.reset_mock() | |||||
| self.loader.storage.reset_mock() | |||||
| # Load again | |||||
| res = self.loader.load() | |||||
| assert res == {"status": "uneventful"} | |||||
| self.fetcher_cls.assert_called_once_with( | |||||
| credentials={}, | |||||
| lister_name="fake-lister", | |||||
| lister_instance_name="", | |||||
| origin=Origin(url=self.repo_url), | |||||
| ) | |||||
| self.fetcher.get_parent_origins.assert_not_called() | |||||
| assert self.loader.storage.origin_visit_get_latest.mock_calls == [ | |||||
| # Tries the same origin, and finds a snapshot | |||||
| call( | |||||
| self.repo_url, | |||||
| allowed_statuses=None, | |||||
| require_snapshot=True, | |||||
| type=None, | |||||
| ), | |||||
| # -> does not need to fall back to the parent | |||||
| ] | |||||
| class DumbGitLoaderTestBase(FullGitLoaderTests): | class DumbGitLoaderTestBase(FullGitLoaderTests): | ||||
| """Prepare a git repository to be loaded using the HTTP dumb transfer protocol.""" | """Prepare a git repository to be loaded using the HTTP dumb transfer protocol.""" | ||||
| @pytest.fixture(autouse=True) | @pytest.fixture(autouse=True) | ||||
| def init(self, swh_storage, datadir, tmp_path): | def init(self, swh_storage, datadir, tmp_path): | ||||
| # remove any proxy settings in order to successfully spawn a local HTTP server | # remove any proxy settings in order to successfully spawn a local HTTP server | ||||
| http_proxy = os.environ.get("http_proxy") | http_proxy = os.environ.get("http_proxy") | ||||
| ▲ Show 20 Lines • Show All 189 Lines • Show Last 20 Lines | |||||