Differential D4698 Diff 18696 swh/storage/tests/storage_tests.py

Changeset View

Standalone View

swh/storage/tests/storage_tests.py

# Copyright (C) 2015-2020 The Software Heritage developers		# Copyright (C) 2015-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 collections import defaultdict		from collections import defaultdict
import datetime		import datetime
from datetime import timedelta		from datetime import timedelta
import inspect		import inspect
import itertools		import itertools
import math		import math
import random		import random
from typing import Any, ClassVar, Dict, Iterator, Optional		from typing import Any, ClassVar, Dict, Iterator, Optional

import attr		import attr
from hypothesis import HealthCheck, given, settings, strategies		from hypothesis import HealthCheck, given, settings, strategies
import pytest		import pytest

from swh.model import from_disk		from swh.model import from_disk
from swh.model.hashutil import hash_to_bytes		from swh.model.hashutil import hash_to_bytes
from swh.model.hypothesis_strategies import objects		from swh.model.hypothesis_strategies import objects
		from swh.model.identifiers import ExtendedObjectType, ExtendedSWHID
from swh.model.model import (		from swh.model.model import (
Content,		Content,
Directory,		Directory,
		ExtID,
Origin,		Origin,
OriginVisit,		OriginVisit,
OriginVisitStatus,		OriginVisitStatus,
Person,		Person,
RawExtrinsicMetadata,		RawExtrinsicMetadata,
Revision,		Revision,
SkippedContent,		SkippedContent,
Snapshot,		Snapshot,
▲ Show 20 Lines • Show All 1,022 Lines • ▼ Show 20 Lines	def test_revision_get_random(self, swh_storage, sample_data):
swh_storage.revision_add([revision1, revision2, revision3])		swh_storage.revision_add([revision1, revision2, revision3])

assert swh_storage.revision_get_random() in {		assert swh_storage.revision_get_random() in {
revision1.id,		revision1.id,
revision2.id,		revision2.id,
revision3.id,		revision3.id,
}		}

		def test_extid_add_git(self, swh_storage, sample_data):

		gitids = [
		revision.id
		for revision in sample_data.revisions
		if revision.type.value == "git"
		]
		nullids = [None] * len(gitids)
		extids = [
		ExtID(
		extid=gitid,
		extid_type="git",
		target=ExtendedSWHID(
		object_id=gitid, object_type=ExtendedObjectType.REVISION,
		),
		)
		for gitid in gitids
		]

		assert swh_storage.extid_get_from_extid("git", gitids) == nullids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, gitids)
		== nullids
		)

		summary = swh_storage.extid_add(extids)
		assert summary == {"extid:add": len(gitids)}

		assert swh_storage.extid_get_from_extid("git", gitids) == extids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, gitids)
		== extids
		)

		assert swh_storage.extid_get_from_extid("hg", gitids) == nullids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.RELEASE, gitids)
		== nullids
		)

		def test_extid_add_hg(self, swh_storage, sample_data):
		def get_node(revision):
		node = None
		if revision.extra_headers:
		node = dict(revision.extra_headers).get(b"node")
		if node is None and revision.metadata:
		node = hash_to_bytes(revision.metadata.get("node"))
		return node

		swhids = [
		revision.id
		for revision in sample_data.revisions
		if revision.type.value == "hg"
		]
		extids = [
		get_node(revision)
		for revision in sample_data.revisions
		if revision.type.value == "hg"
		]
		nullids = [None] * len(swhids)

		assert swh_storage.extid_get_from_extid("hg", extids) == nullids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, swhids)
		== nullids
		)

		extid_objs = [
		ExtID(
		extid=hgid,
		extid_type="hg",
		target=ExtendedSWHID(
		object_id=swhid, object_type=ExtendedObjectType.REVISION,
		),
		)
		for hgid, swhid in zip(extids, swhids)
		]
		summary = swh_storage.extid_add(extid_objs)
		assert summary == {"extid:add": len(swhids)}

		assert swh_storage.extid_get_from_extid("hg", extids) == extid_objs
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, swhids)
		== extid_objs
		)

		assert swh_storage.extid_get_from_extid("git", extids) == nullids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.RELEASE, swhids)
		== nullids
		)

		def test_extid_add_twice(self, swh_storage, sample_data):

		gitids = [
		revision.id
		for revision in sample_data.revisions
		if revision.type.value == "git"
		]

		extids = [
		ExtID(
		extid=gitid,
		extid_type="git",
		target=ExtendedSWHID(
		object_id=gitid, object_type=ExtendedObjectType.REVISION,
		),
		)
		for gitid in gitids
		]
		summary = swh_storage.extid_add(extids)
		assert summary == {"extid:add": len(gitids)}

		# add them again, should be noop
		summary = swh_storage.extid_add(extids)
		# assert summary == {"extid:add": 0}
		assert swh_storage.extid_get_from_extid("git", gitids) == extids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, gitids)
		== extids
		)

		def test_extid_add_extid_unicity(self, swh_storage, sample_data):

		ids = [
		revision.id
		for revision in sample_data.revisions
		if revision.type.value == "git"
		]
		nullids = [None] * len(ids)

		extids = [
		ExtID(
		extid=extid,
		extid_type="git",
		target=ExtendedSWHID(
		object_id=extid, object_type=ExtendedObjectType.REVISION,
		),
		)
		for extid in ids
		]
		swh_storage.extid_add(extids)

		# try to add "modified-extid" versions, should be noops
		extids2 = [
		ExtID(
		extid=extid,
		extid_type="hg",
		target=ExtendedSWHID(
		object_id=extid, object_type=ExtendedObjectType.REVISION,
		),
		)
		for extid in ids
		]
		swh_storage.extid_add(extids2)

		assert swh_storage.extid_get_from_extid("git", ids) == extids
		assert swh_storage.extid_get_from_extid("hg", ids) == nullids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, ids)
		== extids
		)

		def test_extid_add_target_unicity(self, swh_storage, sample_data):

		ids = [
		revision.id
		for revision in sample_data.revisions
		if revision.type.value == "git"
		]
		nullids = [None] * len(ids)

		extids = [
		ExtID(
		extid=extid,
		extid_type="git",
		target=ExtendedSWHID(
		object_id=extid, object_type=ExtendedObjectType.REVISION,
		),
		)
		for extid in ids
		]
		swh_storage.extid_add(extids)

		# try to add "modified" versions, should be noops
		extids2 = [
		ExtID(
		extid=extid,
		extid_type="git",
		target=ExtendedSWHID(
		object_id=extid, object_type=ExtendedObjectType.RELEASE,
		),
		)
		for extid in ids
		]
		swh_storage.extid_add(extids2)

		assert swh_storage.extid_get_from_extid("git", ids) == extids
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.REVISION, ids)
		== extids
		)
		assert (
		swh_storage.extid_get_from_target(ExtendedObjectType.RELEASE, ids)
		== nullids
		)

def test_release_add(self, swh_storage, sample_data):		def test_release_add(self, swh_storage, sample_data):
release, release2 = sample_data.releases[:2]		release, release2 = sample_data.releases[:2]

init_missing = swh_storage.release_missing([release.id, release2.id])		init_missing = swh_storage.release_missing([release.id, release2.id])
assert list(init_missing) == [release.id, release2.id]		assert list(init_missing) == [release.id, release2.id]

actual_result = swh_storage.release_add([release, release2])		actual_result = swh_storage.release_add([release, release2])
assert actual_result == {"release:add": 2}		assert actual_result == {"release:add": 2}
▲ Show 20 Lines • Show All 2,904 Lines • Show Last 20 Lines