Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/git/tests/test_from_disk.py
| # Copyright (C) 2018-2020 The Software Heritage developers | # Copyright (C) 2018-2020 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 datetime | import datetime | ||||
| import os.path | import os.path | ||||
| import dulwich.repo | import dulwich.repo | ||||
| from swh.model.model import Snapshot, SnapshotBranch, TargetType | from typing import Optional | ||||
| from swh.model.model import OriginVisitStatus, Snapshot, SnapshotBranch, TargetType | |||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.loader.core.tests import BaseLoaderTest | from swh.loader.core.tests import BaseLoaderTest | ||||
| from swh.loader.git.from_disk import GitLoaderFromDisk as OrigGitLoaderFromDisk | from swh.loader.git.from_disk import GitLoaderFromDisk as OrigGitLoaderFromDisk | ||||
| from swh.loader.git.from_disk import GitLoaderFromArchive as OrigGitLoaderFromArchive | from swh.loader.git.from_disk import GitLoaderFromArchive as OrigGitLoaderFromArchive | ||||
| from . import TEST_LOADER_CONFIG | from . import TEST_LOADER_CONFIG | ||||
| def assert_last_visit_ok( | |||||
| storage, | |||||
| url: str, | |||||
| status: str, | |||||
| type: Optional[str] = None, | |||||
| snapshot: Optional[bytes] = None, | |||||
| ) -> OriginVisitStatus: | |||||
| """Ensure a given visit/visit-status is expectedly with status, (optional) type and | |||||
| (optional) snapshot. | |||||
| This returns the last visit_status for that given origin | |||||
| Args: | |||||
| url: Origin url | |||||
| status: expected status | |||||
| type: expected_type | |||||
| Raises: | |||||
| AssertionError in case visit or visit status is not found | |||||
| Returns: | |||||
| the visit status for further check during the remaining part of the test. | |||||
| """ | |||||
| visit = storage.origin_visit_get_latest(url) | |||||
| assert visit is not None, f"Visit should exist for origin {url}" | |||||
| if type: | |||||
| assert visit["type"] == type | |||||
| visit_id = visit["visit"] | |||||
| visit_status = storage.origin_visit_status_get_latest(url, visit_id) | |||||
| assert ( | |||||
| visit_status is not None | |||||
| ), f"Visit status should exist for origin {url}, visit {visit_id}" | |||||
| assert visit_status.status == status | |||||
| if snapshot: | |||||
| assert visit_status.snapshot == snapshot | |||||
| return visit_status | |||||
| class GitLoaderFromArchive(OrigGitLoaderFromArchive): | class GitLoaderFromArchive(OrigGitLoaderFromArchive): | ||||
| def project_name_from_archive(self, archive_path): | def project_name_from_archive(self, archive_path): | ||||
| # We don't want the project name to be 'resources'. | # We don't want the project name to be 'resources'. | ||||
| return "testrepo" | return "testrepo" | ||||
| def parse_config_file(self, *args, **kwargs): | def parse_config_file(self, *args, **kwargs): | ||||
| return TEST_LOADER_CONFIG | return TEST_LOADER_CONFIG | ||||
| ▲ Show 20 Lines • Show All 154 Lines • ▼ Show 20 Lines | def test_load(self): | ||||
| self.assertRevisionsContain(REVISIONS1) | self.assertRevisionsContain(REVISIONS1) | ||||
| self.assertSnapshotEqual(SNAPSHOT1) | self.assertSnapshotEqual(SNAPSHOT1) | ||||
| self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | ||||
| self.assertEqual(self.loader.visit_status(), "full") | self.assertEqual(self.loader.visit_status(), "full") | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | assert_last_visit_ok( | ||||
| self.assertEqual(visit["snapshot"], hash_to_bytes(SNAPSHOT1["id"])) | self.storage, | ||||
| self.assertEqual(visit["status"], "full") | self.repo_url, | ||||
| status="full", | |||||
| type="git", | |||||
| snapshot=hash_to_bytes(SNAPSHOT1["id"]), | |||||
| ) | |||||
| def test_load_unchanged(self): | def test_load_unchanged(self): | ||||
| """Checks loading a repository a second time does not add | """Checks loading a repository a second time does not add | ||||
| any extra data.""" | any extra data.""" | ||||
| res = self.load() | res = self.load() | ||||
| self.assertEqual(res["status"], "eventful") | self.assertEqual(res["status"], "eventful") | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | assert_last_visit_ok( | ||||
| self.assertEqual(visit["snapshot"], hash_to_bytes(SNAPSHOT1["id"])) | self.storage, | ||||
| self.assertEqual(visit["status"], "full") | self.repo_url, | ||||
| status="full", | |||||
| type="git", | |||||
| snapshot=hash_to_bytes(SNAPSHOT1["id"]), | |||||
| ) | |||||
| res = self.load() | res = self.load() | ||||
| self.assertEqual(res["status"], "uneventful") | self.assertEqual(res["status"], "uneventful") | ||||
| self.assertCountSnapshots(1) | self.assertCountSnapshots(1) | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | assert_last_visit_ok( | ||||
| self.assertEqual(visit["snapshot"], hash_to_bytes(SNAPSHOT1["id"])) | self.storage, | ||||
| self.assertEqual(visit["status"], "full") | self.repo_url, | ||||
| status="full", | |||||
| type="git", | |||||
| snapshot=hash_to_bytes(SNAPSHOT1["id"]), | |||||
| ) | |||||
| class DirGitLoaderTest(BaseDirGitLoaderFromDiskTest, GitLoaderFromDiskTests): | class DirGitLoaderTest(BaseDirGitLoaderFromDiskTest, GitLoaderFromDiskTests): | ||||
| """Tests for the GitLoaderFromDisk. Includes the common ones, and | """Tests for the GitLoaderFromDisk. Includes the common ones, and | ||||
| add others that only work with a local dir.""" | add others that only work with a local dir.""" | ||||
| def test_load_changed(self): | def test_load_changed(self): | ||||
| """Loads a repository, makes some changes by adding files, commits, | """Loads a repository, makes some changes by adding files, commits, | ||||
| Show All 26 Lines | def test_load_changed(self): | ||||
| self.assertCountRevisions(7 + 1) | self.assertCountRevisions(7 + 1) | ||||
| self.assertCountSnapshots(1 + 1) | self.assertCountSnapshots(1 + 1) | ||||
| self.assertRevisionsContain(revisions) | self.assertRevisionsContain(revisions) | ||||
| self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | ||||
| self.assertEqual(self.loader.visit_status(), "full") | self.assertEqual(self.loader.visit_status(), "full") | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | visit_status = assert_last_visit_ok( | ||||
| self.assertIsNotNone(visit["snapshot"]) | self.storage, self.repo_url, status="full", type="git" | ||||
| self.assertEqual(visit["status"], "full") | ) | ||||
| self.assertIsNotNone(visit_status.snapshot) | |||||
| snapshot_id = visit["snapshot"] | snapshot_id = visit_status.snapshot | ||||
| snapshot = self.storage.snapshot_get(snapshot_id) | snapshot = self.storage.snapshot_get(snapshot_id) | ||||
| branches = snapshot["branches"] | branches = snapshot["branches"] | ||||
| assert branches[b"HEAD"] == { | assert branches[b"HEAD"] == { | ||||
| "target": b"refs/heads/master", | "target": b"refs/heads/master", | ||||
| "target_type": "alias", | "target_type": "alias", | ||||
| } | } | ||||
| assert branches[b"refs/heads/master"] == { | assert branches[b"refs/heads/master"] == { | ||||
| "target": hash_to_bytes(new_revision), | "target": hash_to_bytes(new_revision), | ||||
| Show All 31 Lines | def test_load_changed(self): | ||||
| self.assertCountRevisions(7 + 2) | self.assertCountRevisions(7 + 2) | ||||
| self.assertCountSnapshots(1 + 1 + 1) | self.assertCountSnapshots(1 + 1 + 1) | ||||
| self.assertRevisionsContain(revisions) | self.assertRevisionsContain(revisions) | ||||
| self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | self.assertEqual(self.loader.load_status(), {"status": "eventful"}) | ||||
| self.assertEqual(self.loader.visit_status(), "full") | self.assertEqual(self.loader.visit_status(), "full") | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | visit_status = assert_last_visit_ok( | ||||
| self.assertIsNotNone(visit["snapshot"]) | self.storage, self.repo_url, status="full", type="git" | ||||
| self.assertEqual(visit["status"], "full") | ) | ||||
| self.assertIsNotNone(visit_status.snapshot) | |||||
| merge_snapshot_id = visit["snapshot"] | merge_snapshot_id = visit_status.snapshot | ||||
| assert merge_snapshot_id != snapshot_id | assert merge_snapshot_id != snapshot_id | ||||
| merge_snapshot = self.storage.snapshot_get(merge_snapshot_id) | merge_snapshot = self.storage.snapshot_get(merge_snapshot_id) | ||||
| merge_branches = merge_snapshot["branches"] | merge_branches = merge_snapshot["branches"] | ||||
| assert merge_branches[b"HEAD"] == { | assert merge_branches[b"HEAD"] == { | ||||
| "target": b"refs/heads/master", | "target": b"refs/heads/master", | ||||
| "target_type": "alias", | "target_type": "alias", | ||||
| } | } | ||||
| Show All 38 Lines | def test_load_filter_branches(self): | ||||
| # Load the modified repository | # Load the modified repository | ||||
| res = self.load() | res = self.load() | ||||
| assert res["status"] == "eventful" | assert res["status"] == "eventful" | ||||
| assert self.loader.load_status() == {"status": "eventful"} | assert self.loader.load_status() == {"status": "eventful"} | ||||
| assert self.loader.visit_status() == "full" | assert self.loader.visit_status() == "full" | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | assert_last_visit_ok( | ||||
| assert visit["snapshot"] == expected_snapshot.id | self.storage, | ||||
| assert visit["status"] == "full" | self.repo_url, | ||||
| status="full", | |||||
| type="git", | |||||
| snapshot=expected_snapshot.id, | |||||
| ) | |||||
| def test_load_dangling_symref(self): | def test_load_dangling_symref(self): | ||||
| with open(os.path.join(self.destination_path, ".git/HEAD"), "wb") as f: | with open(os.path.join(self.destination_path, ".git/HEAD"), "wb") as f: | ||||
| f.write(b"ref: refs/heads/dangling-branch\n") | f.write(b"ref: refs/heads/dangling-branch\n") | ||||
| res = self.load() | res = self.load() | ||||
| self.assertEqual(res["status"], "eventful", res) | self.assertEqual(res["status"], "eventful", res) | ||||
| self.assertContentsContain(CONTENT1) | self.assertContentsContain(CONTENT1) | ||||
| self.assertCountDirectories(7) | self.assertCountDirectories(7) | ||||
| self.assertCountReleases(0) # FIXME: should be 2 after T2059 | self.assertCountReleases(0) # FIXME: should be 2 after T2059 | ||||
| self.assertCountRevisions(7) | self.assertCountRevisions(7) | ||||
| self.assertCountSnapshots(1) | self.assertCountSnapshots(1) | ||||
| visit = self.storage.origin_visit_get_latest(self.repo_url) | visit_status = assert_last_visit_ok( | ||||
| snapshot_id = visit["snapshot"] | self.storage, self.repo_url, status="full", type="git" | ||||
| ) | |||||
| snapshot_id = visit_status.snapshot | |||||
| assert snapshot_id is not None | assert snapshot_id is not None | ||||
| assert visit["status"] == "full" | |||||
| snapshot = self.storage.snapshot_get(snapshot_id) | snapshot = self.storage.snapshot_get(snapshot_id) | ||||
| branches = snapshot["branches"] | branches = snapshot["branches"] | ||||
| assert branches[b"HEAD"] == { | assert branches[b"HEAD"] == { | ||||
| "target": b"refs/heads/dangling-branch", | "target": b"refs/heads/dangling-branch", | ||||
| "target_type": "alias", | "target_type": "alias", | ||||
| } | } | ||||
| assert branches[b"refs/heads/dangling-branch"] is None | assert branches[b"refs/heads/dangling-branch"] is None | ||||
| class GitLoaderFromArchiveTest(BaseGitLoaderFromArchiveTest, GitLoaderFromDiskTests): | class GitLoaderFromArchiveTest(BaseGitLoaderFromArchiveTest, GitLoaderFromDiskTests): | ||||
| """Tests for GitLoaderFromArchive. Imports the common ones | """Tests for GitLoaderFromArchive. Imports the common ones | ||||
| from GitLoaderTests.""" | from GitLoaderTests.""" | ||||
| pass | pass | ||||