Page MenuHomeSoftware Heritage
Paste P993

swh/loader/mercurial/tests/test_partial_storage.py
ActivePublic
Actions

Authored by vlorentz on Mar 29 2021, 1:56 PM.
from datetime import datetime
from hashlib import sha1
import os
import pytest
from swh.loader.mercurial.utils import parse_visit_date
from swh.loader.tests import (
assert_last_visit_matches,
check_snapshot,
get_stats,
prepare_repository_from_archive,
)
from swh.model.from_disk import Content, DentryPerms
from swh.model.hashutil import hash_to_bytes
from swh.model.identifiers import ObjectType
from swh.model.model import RevisionType, Snapshot, SnapshotBranch, TargetType
from swh.storage import get_storage
from swh.storage.algos.snapshot import snapshot_get_latest
from ..from_disk import HgDirectory, HgLoaderFromDisk
from .loader_checker import ExpectedSwhids, LoaderChecker
def _partial_copy_storage(
old_storage, origin_url: str,
):
"""Create a new storage, and only copy origin/visit/status/snapshot to it."""
new_storage = get_storage(cls="memory")
snapshot = snapshot_get_latest(old_storage, origin_url)
assert snapshot
# copy origin, visit, status
new_storage.origin_add(old_storage.origin_get([origin_url]))
visit = old_storage.origin_visit_get_latest(origin_url)
new_storage.origin_visit_add([visit])
statuses = old_storage.origin_visit_status_get(origin_url, visit.visit).results
new_storage.origin_visit_status_add(statuses)
new_storage.snapshot_add([snapshot])
return new_storage
def test_load_unchanged_repo__dangling(swh_storage, datadir, tmp_path):
"""Checks the loader will load revisions targeted by an ExtID if the
revisions are missing from the storage"""
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://", "")
loader = HgLoaderFromDisk(swh_storage, repo_path)
assert loader.load() == {"status": "eventful"}
assert get_stats(loader.storage) == {
"content": 3,
"directory": 3,
"origin": 1,
"origin_visit": 1,
"release": 1,
"revision": 3,
"skipped_content": 0,
"snapshot": 1,
}
old_storage = swh_storage
# Create a new storage, and only copy some objects to it.
new_storage = _partial_copy_storage(old_storage, repo_path,)
# Replace the loader's storage with the new, partial, storage
loader.storage = new_storage
assert get_stats(loader.storage) == {
"content": 0,
"directory": 0,
"origin": 1,
"origin_visit": 1,
"release": 0,
"revision": 0,
"skipped_content": 0,
"snapshot": 1,
}
assert loader.load() == {"status": "eventful"}
assert get_stats(loader.storage) == {
"content": 0, # <--------- not loaded again
"directory": 3,
"origin": 1,
"origin_visit": 2,
"release": 1,
"revision": 3,
"skipped_content": 0,
"snapshot": 1,
}
About · Developer information · Legal