Differential D3648 Diff 12847 swh/storage/tests/algos/test_snapshot.py

Changeset View

Standalone View

swh/storage/tests/algos/test_snapshot.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

from hypothesis import given		from hypothesis import given
import pytest		import pytest

from swh.model.collections import ImmutableDict		from swh.model.collections import ImmutableDict
from swh.model.hypothesis_strategies import snapshots, branch_names, branch_targets		from swh.model.hypothesis_strategies import snapshots, branch_names, branch_targets
from swh.model.model import OriginVisit, OriginVisitStatus, Snapshot		from swh.model.model import OriginVisit, OriginVisitStatus, Snapshot

from swh.storage.algos.snapshot import snapshot_get_all_branches, snapshot_get_latest		from swh.storage.algos.snapshot import (
		snapshot_get_all_branches,
		snapshot_get_latest,
		snapshot_id_get_from_revision,
		)
from swh.storage.utils import now		from swh.storage.utils import now


@pytest.fixture		@pytest.fixture
def swh_storage_backend_config():		def swh_storage_backend_config():
yield {		yield {
"cls": "memory",		"cls": "memory",
"journal_writer": None,		"journal_writer": None,
▲ Show 20 Lines • Show All 118 Lines • ▼ Show 20 Lines	def test_snapshot_get_latest(swh_storage, sample_data):

actual_snapshot = snapshot_get_latest(swh_storage, origin.url, branches_count=1)		actual_snapshot = snapshot_get_latest(swh_storage, origin.url, branches_count=1)
assert actual_snapshot is not None		assert actual_snapshot is not None
assert actual_snapshot.id == complete_snapshot.id		assert actual_snapshot.id == complete_snapshot.id
assert len(actual_snapshot.branches.values()) == 1		assert len(actual_snapshot.branches.values()) == 1

with pytest.raises(ValueError, match="branches_count must be a positive integer"):		with pytest.raises(ValueError, match="branches_count must be a positive integer"):
snapshot_get_latest(swh_storage, origin.url, branches_count="something-wrong")		snapshot_get_latest(swh_storage, origin.url, branches_count="something-wrong")


		def test_snapshot_get_id_from_revision(swh_storage, sample_data):
		origin = sample_data.origin
		swh_storage.origin_add([origin])

		date_visit2 = now()
		visit1, visit2 = sample_data.origin_visits[:2]
		assert visit1.origin == origin.url

		ov1, ov2 = swh_storage.origin_visit_add([visit1, visit2])

		revision1, revision2, revision3 = sample_data.revisions[:3]
		swh_storage.revision_add([revision1, revision2])

		empty_snapshot, complete_snapshot = sample_data.snapshots[1:3]
		swh_storage.snapshot_add([complete_snapshot])

		# Add complete_snapshot to visit1 which targets revision1
		ovs1, ovs2 = [
		OriginVisitStatus(
		origin=origin.url,
		visit=ov1.visit,
		date=date_visit2,
		status="partial",
		snapshot=complete_snapshot.id,
		),
		OriginVisitStatus(
		origin=origin.url,
		visit=ov2.visit,
		date=now(),
		status="full",
		snapshot=empty_snapshot.id,
		),
		]

		swh_storage.origin_visit_status_add([ovs1, ovs2])
		assert ov1.date < ov2.date
		assert ov2.date < ovs1.date
		assert ovs1.date < ovs2.date

		# revision3 does not exist so result is None
		actual_snapshot_id = snapshot_id_get_from_revision(
		swh_storage, origin.url, revision3.id
		)
		assert actual_snapshot_id is None

		# no snapshot targets revision2 for origin.url so result is None
		actual_snapshot_id = snapshot_id_get_from_revision(
		swh_storage, origin.url, revision2.id
		)
		assert actual_snapshot_id is None

		# complete_snapshot targets at least revision1
		actual_snapshot_id = snapshot_id_get_from_revision(
		swh_storage, origin.url, revision1.id
		)
		assert actual_snapshot_id == complete_snapshot.id