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 os.path | import os.path | ||||
import dulwich.repo | import dulwich.repo | ||||
from swh.model.model import Snapshot, SnapshotBranch, TargetType | |||||
from swh.model.hashutil import hash_to_bytes | |||||
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 swh.loader.core.tests import BaseLoaderTest | |||||
from swh.model.hashutil import hash_to_bytes | |||||
from . import TEST_LOADER_CONFIG | from . import TEST_LOADER_CONFIG | ||||
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" | ||||
▲ Show 20 Lines • Show All 287 Lines • ▼ Show 20 Lines | def test_load_changed(self): | ||||
"target": b"refs/heads/master", | "target": b"refs/heads/master", | ||||
"target_type": "alias", | "target_type": "alias", | ||||
} | } | ||||
assert merge_branches[b"refs/heads/master"] == { | assert merge_branches[b"refs/heads/master"] == { | ||||
"target": hash_to_bytes(merge_commit.decode()), | "target": hash_to_bytes(merge_commit.decode()), | ||||
"target_type": "revision", | "target_type": "revision", | ||||
} | } | ||||
def test_load_filter_branches(self): | |||||
filtered_branches = {b"refs/pull/42/merge"} | |||||
unfiltered_branches = {b"refs/pull/42/head"} | |||||
# Add branches to the repository on disk; some should be filtered by | |||||
# the loader, some should not. | |||||
for branch_name in filtered_branches | unfiltered_branches: | |||||
self.repo[branch_name] = self.repo[b"refs/heads/master"] | |||||
# Generate the expected snapshot from SNAPSHOT1 (which is the original | |||||
# state of the git repo)... | |||||
branches = {} | |||||
for branch_name, branch_dict in SNAPSHOT1["branches"].items(): | |||||
target_type_name = branch_dict["target_type"] | |||||
target_obj = branch_dict["target"] | |||||
if target_type_name != "alias": | |||||
target = bytes.fromhex(target_obj) | |||||
else: | |||||
target = target_obj.encode() | |||||
branch = SnapshotBranch( | |||||
target=target, target_type=TargetType(target_type_name) | |||||
) | |||||
branches[branch_name.encode()] = branch | |||||
# ... and the unfiltered_branches, which are all pointing to the same | |||||
# commit as "refs/heads/master". | |||||
for branch_name in unfiltered_branches: | |||||
branches[branch_name] = branches[b"refs/heads/master"] | |||||
expected_snapshot = Snapshot(branches=branches) | |||||
# Load the modified repository | |||||
res = self.load() | |||||
assert res["status"] == "eventful" | |||||
assert self.loader.load_status() == {"status": "eventful"} | |||||
assert self.loader.visit_status() == "full" | |||||
visit = self.storage.origin_visit_get_latest(self.repo_url) | |||||
assert visit["snapshot"] == expected_snapshot.id | |||||
assert visit["status"] == "full" | |||||
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) | ||||
Show All 25 Lines |