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 copy | import copy | ||||
| import datetime | import datetime | ||||
| import os.path | import os.path | ||||
| import dulwich.repo | import dulwich.repo | ||||
| import pytest | import pytest | ||||
| from unittest import TestCase | from unittest import TestCase | ||||
| from swh.model.model import Snapshot, SnapshotBranch, TargetType | from swh.model.model import Snapshot, SnapshotBranch, TargetType | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.loader.git.from_disk import GitLoaderFromDisk, GitLoaderFromArchive | from swh.storage.algos.snapshot import snapshot_get_all_branches | ||||
| from swh.loader.tests import ( | from swh.loader.tests import ( | ||||
| assert_last_visit_matches, | assert_last_visit_matches, | ||||
| check_snapshot, | check_snapshot, | ||||
| get_stats, | get_stats, | ||||
| prepare_repository_from_archive, | prepare_repository_from_archive, | ||||
| ) | ) | ||||
| from swh.loader.git.from_disk import GitLoaderFromDisk, GitLoaderFromArchive | |||||
| SNAPSHOT1 = Snapshot( | SNAPSHOT1 = Snapshot( | ||||
| id=hash_to_bytes("a23699280a82a043f8c0994cf1631b568f716f95"), | id=hash_to_bytes("a23699280a82a043f8c0994cf1631b568f716f95"), | ||||
| branches={ | branches={ | ||||
| b"HEAD": SnapshotBranch( | b"HEAD": SnapshotBranch( | ||||
| target=b"refs/heads/master", target_type=TargetType.ALIAS, | target=b"refs/heads/master", target_type=TargetType.ALIAS, | ||||
| ), | ), | ||||
| b"refs/heads/master": SnapshotBranch( | b"refs/heads/master": SnapshotBranch( | ||||
| ▲ Show 20 Lines • Show All 196 Lines • ▼ Show 20 Lines | def test_load_changed(self): | ||||
| assert stats1 == expected_stats | assert stats1 == expected_stats | ||||
| visit_status = assert_last_visit_matches( | visit_status = assert_last_visit_matches( | ||||
| self.loader.storage, self.repo_url, status="full", type="git" | self.loader.storage, self.repo_url, status="full", type="git" | ||||
| ) | ) | ||||
| assert visit_status.snapshot is not None | assert visit_status.snapshot is not None | ||||
| snapshot_id = visit_status.snapshot | snapshot_id = visit_status.snapshot | ||||
| snapshot = self.loader.storage.snapshot_get(snapshot_id) | snapshot = snapshot_get_all_branches(self.loader.storage, snapshot_id) | ||||
| branches = snapshot["branches"] | branches = snapshot.branches | ||||
| assert branches[b"HEAD"] == { | assert branches[b"HEAD"] == SnapshotBranch( | ||||
| "target": b"refs/heads/master", | target=b"refs/heads/master", target_type=TargetType.ALIAS, | ||||
| "target_type": "alias", | ) | ||||
| } | assert branches[b"refs/heads/master"] == SnapshotBranch( | ||||
| assert branches[b"refs/heads/master"] == { | target=hash_to_bytes(new_revision), target_type=TargetType.REVISION, | ||||
| "target": hash_to_bytes(new_revision), | ) | ||||
| "target_type": "revision", | |||||
| } | |||||
| # Merge branch1 into HEAD. | # Merge branch1 into HEAD. | ||||
| current = self.repo[b"HEAD"] | current = self.repo[b"HEAD"] | ||||
| branch1 = self.repo[b"refs/heads/branch1"] | branch1 = self.repo[b"refs/heads/branch1"] | ||||
| merged_tree = dulwich.objects.Tree() | merged_tree = dulwich.objects.Tree() | ||||
| for item in self.repo[current.tree].items(): | for item in self.repo[current.tree].items(): | ||||
| Show All 30 Lines | def test_load_changed(self): | ||||
| visit_status = assert_last_visit_matches( | visit_status = assert_last_visit_matches( | ||||
| self.loader.storage, self.repo_url, status="full", type="git" | self.loader.storage, self.repo_url, status="full", type="git" | ||||
| ) | ) | ||||
| assert visit_status.snapshot is not None | assert visit_status.snapshot is not None | ||||
| merge_snapshot_id = visit_status.snapshot | merge_snapshot_id = visit_status.snapshot | ||||
| assert merge_snapshot_id != snapshot_id | assert merge_snapshot_id != snapshot_id | ||||
| merge_snapshot = self.loader.storage.snapshot_get(merge_snapshot_id) | merge_snapshot = snapshot_get_all_branches( | ||||
| merge_branches = merge_snapshot["branches"] | self.loader.storage, merge_snapshot_id | ||||
| assert merge_branches[b"HEAD"] == { | ) | ||||
| "target": b"refs/heads/master", | merge_branches = merge_snapshot.branches | ||||
| "target_type": "alias", | assert merge_branches[b"HEAD"] == SnapshotBranch( | ||||
| } | target=b"refs/heads/master", target_type=TargetType.ALIAS, | ||||
| assert merge_branches[b"refs/heads/master"] == { | ) | ||||
| "target": hash_to_bytes(merge_commit.decode()), | assert merge_branches[b"refs/heads/master"] == SnapshotBranch( | ||||
| "target_type": "revision", | target=hash_to_bytes(merge_commit.decode()), | ||||
| } | target_type=TargetType.REVISION, | ||||
| ) | |||||
| def test_load_filter_branches(self): | def test_load_filter_branches(self): | ||||
| filtered_branches = {b"refs/pull/42/merge"} | filtered_branches = {b"refs/pull/42/merge"} | ||||
| unfiltered_branches = {b"refs/pull/42/head"} | unfiltered_branches = {b"refs/pull/42/head"} | ||||
| # Add branches to the repository on disk; some should be filtered by | # Add branches to the repository on disk; some should be filtered by | ||||
| # the loader, some should not. | # the loader, some should not. | ||||
| for branch_name in filtered_branches | unfiltered_branches: | for branch_name in filtered_branches | unfiltered_branches: | ||||
| Show All 31 Lines | def test_load_dangling_symref(self): | ||||
| assert res == {"status": "eventful"} | assert res == {"status": "eventful"} | ||||
| visit_status = assert_last_visit_matches( | visit_status = assert_last_visit_matches( | ||||
| self.loader.storage, self.repo_url, status="full", type="git" | self.loader.storage, self.repo_url, status="full", type="git" | ||||
| ) | ) | ||||
| snapshot_id = visit_status.snapshot | snapshot_id = visit_status.snapshot | ||||
| assert snapshot_id is not None | assert snapshot_id is not None | ||||
| snapshot = self.loader.storage.snapshot_get(snapshot_id) | snapshot = snapshot_get_all_branches(self.loader.storage, snapshot_id) | ||||
| branches = snapshot["branches"] | branches = snapshot.branches | ||||
| assert branches[b"HEAD"] == { | assert branches[b"HEAD"] == SnapshotBranch( | ||||
| "target": b"refs/heads/dangling-branch", | target=b"refs/heads/dangling-branch", target_type=TargetType.ALIAS, | ||||
| "target_type": "alias", | ) | ||||
| } | |||||
| assert branches[b"refs/heads/dangling-branch"] is None | assert branches[b"refs/heads/dangling-branch"] is None | ||||
| stats = get_stats(self.loader.storage) | stats = get_stats(self.loader.storage) | ||||
| assert stats == { | assert stats == { | ||||
| "content": 4, | "content": 4, | ||||
| "directory": 7, | "directory": 7, | ||||
| "origin": 1, | "origin": 1, | ||||
| "origin_visit": 1, | "origin_visit": 1, | ||||
| ▲ Show 20 Lines • Show All 48 Lines • Show Last 20 Lines | |||||