Differential D6036 Diff 21860 swh/loader/mercurial/tests/test_from_disk.py

Changeset View

Standalone View

swh/loader/mercurial/tests/test_from_disk.py

# Copyright (C) 2020-2021 The Software Heritage developers		# Copyright (C) 2020-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 datetime import datetime		from datetime import datetime
from hashlib import sha1		from hashlib import sha1
import os		import os
from pathlib import Path		from pathlib import Path
import subprocess		import subprocess
		import unittest

import attr		import attr
import pytest		import pytest

from swh.loader.mercurial.loader import HgBundle20Loader		from swh.loader.mercurial.loader import HgBundle20Loader
from swh.loader.mercurial.utils import parse_visit_date		from swh.loader.mercurial.utils import parse_visit_date
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.model.from_disk import Content, DentryPerms		from swh.model.from_disk import Content, DentryPerms
from swh.model.hashutil import hash_to_bytes, hash_to_hex		from swh.model.hashutil import hash_to_bytes, hash_to_hex
from swh.model.identifiers import ObjectType		from swh.model.identifiers import ObjectType
from swh.model.model import RevisionType, Snapshot, SnapshotBranch, TargetType		from swh.model.model import RevisionType, Snapshot, SnapshotBranch, TargetType
from swh.storage import get_storage		from swh.storage import get_storage
from swh.storage.algos.snapshot import snapshot_get_latest		from swh.storage.algos.snapshot import snapshot_get_latest

from ..from_disk import HgDirectory, HgLoaderFromDisk		from ..from_disk import EXTID_VERSION, HgDirectory, HgLoaderFromDisk
from .loader_checker import ExpectedSwhids, LoaderChecker		from .loader_checker import ExpectedSwhids, LoaderChecker

VISIT_DATE = parse_visit_date("2016-05-03 15:16:32+00")		VISIT_DATE = parse_visit_date("2016-05-03 15:16:32+00")
assert VISIT_DATE is not None		assert VISIT_DATE is not None


def random_content() -> Content:		def random_content() -> Content:
"""Create minimal content object."""		"""Create minimal content object."""
▲ Show 20 Lines • Show All 570 Lines • ▼ Show 20 Lines	assert get_stats(loader.storage) == {
"directory": 3,		"directory": 3,
"origin": 1,		"origin": 1,
"origin_visit": 2,		"origin_visit": 2,
"release": 1,		"release": 1,
"revision": 3,		"revision": 3,
"skipped_content": 0,		"skipped_content": 0,
"snapshot": 2,		"snapshot": 2,
}		}


		def test_load_repo_check_extids_write_version(swh_storage, datadir, tmp_path):
		"""ExtIDs should be stored with a given version when loading is done"""
		archive_name = "hello"
		archive_path = Path(datadir, f"{archive_name}.tgz")
		repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)

		hg_strip(repo_url.replace("file://", ""), "tip")
		loader = HgLoaderFromDisk(swh_storage, repo_url)
		assert loader.load() == {"status": "eventful"}

		# Ensure we write ExtIDs to a specific version.
		snapshot = snapshot_get_latest(swh_storage, repo_url)

		# First, filter out revisions from that snapshot
		revision_ids = [
		branch.target
		for branch in snapshot.branches.values()
		if branch.target_type == TargetType.REVISION
		]

		assert len(revision_ids) > 0

		# Those revisions should have their associated ExtID version set to EXTID_VERSION
		extids = swh_storage.extid_get_from_target(ObjectType.REVISION, revision_ids)

		assert len(extids) == len(revision_ids)
		for extid in extids:
		assert extid.extid_version == EXTID_VERSION


		def test_load_new_extid_should_be_eventful(swh_storage, datadir, tmp_path):
		"""Changing the extid version should make loaders ignore existing extids,
		and load the repo again."""
		archive_name = "hello"
		archive_path = os.path.join(datadir, f"{archive_name}.tgz")
		repo_url = prepare_repository_from_archive(archive_path, archive_name, tmp_path)
		repo_path = repo_url.replace("file://", "")

		with unittest.mock.patch("swh.loader.mercurial.from_disk.EXTID_VERSION", 0):
		loader = HgLoaderFromDisk(swh_storage, repo_path)
		assert loader.load() == {"status": "eventful"}

		loader = HgLoaderFromDisk(swh_storage, repo_path)
		assert loader.load() == {"status": "eventful"}

		loader = HgLoaderFromDisk(swh_storage, repo_path)
		assert loader.load() == {"status": "uneventful"}

		with unittest.mock.patch("swh.loader.mercurial.from_disk.EXTID_VERSION", 10000):
		loader = HgLoaderFromDisk(swh_storage, repo_path)
		assert loader.load() == {"status": "eventful"}

		loader = HgLoaderFromDisk(swh_storage, repo_path)
		assert loader.load() == {"status": "uneventful"}