Differential D3314 Diff 11732 swh/storage/tests/algos/test_snapshot.py

Changeset View

Standalone View

swh/storage/tests/algos/test_snapshot.py

# Copyright (C) 2018 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.identifiers import snapshot_identifier, identifier_to_bytes
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.identifiers import snapshot_identifier, identifier_to_bytes
		from swh.model.model import Origin, OriginVisit, OriginVisitStatus, Snapshot

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

from swh.storage.algos.snapshot import snapshot_get_all_branches		from swh.storage.tests.storage_data import data


@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 All 16 Lines	snapshot = {
"branches": {b"%s%05d" % (branch_name, i): branch_target for i in range(10000)}		"branches": {b"%s%05d" % (branch_name, i): branch_target for i in range(10000)}
}		}
snapshot["id"] = identifier_to_bytes(snapshot_identifier(snapshot))		snapshot["id"] = identifier_to_bytes(snapshot_identifier(snapshot))

swh_storage.snapshot_add([snapshot])		swh_storage.snapshot_add([snapshot])

returned_snapshot = snapshot_get_all_branches(swh_storage, snapshot["id"])		returned_snapshot = snapshot_get_all_branches(swh_storage, snapshot["id"])
assert snapshot == returned_snapshot		assert snapshot == returned_snapshot


		def test_snapshot_get_latest_none(swh_storage):
		"""Retrieve latest snapshot on unknown origin or origin without snapshot should yield no
		result

		"""
		assert snapshot_get_latest(swh_storage, "unknown-origin") is None

		# no snapshot on origin visit then nothing is found
		origin = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(origin)
		swh_storage.origin_visit_add(
		[
		OriginVisit(
		origin=origin.url,
		date=data.date_visit1,
		type=data.type_visit1,
		status="ongoing",
		snapshot=None,
		)
		]
		)
		assert snapshot_get_latest(swh_storage, origin.url) is None


		def test_snapshot_get_latest_ok(swh_storage):
		origin = Origin.from_dict(data.origin)
		swh_storage.origin_add_one(origin)

		visit1 = OriginVisit(
		origin=origin.url,
		date=data.date_visit1,
		type=data.type_visit1,
		status="ongoing",
		snapshot=None,
		)
		ov1 = swh_storage.origin_visit_add([visit1])[0]

		# Add snapshot to visit1, latest snapshot = visit 1 snapshot
		complete_snapshot = Snapshot.from_dict(data.complete_snapshot)
		swh_storage.snapshot_add([complete_snapshot])

		swh_storage.origin_visit_status_add(
		[
		OriginVisitStatus(
		origin=origin.url,
		visit=ov1.visit,
		date=data.date_visit2,
		status="partial",
		snapshot=None,
		)
		]
		)
		assert data.date_visit1 < data.date_visit2

		# no snapshot associated to the visit, so None
		actual_snapshot = snapshot_get_latest(
		swh_storage, origin.url, allowed_statuses=["partial"]
		)
		assert actual_snapshot is None

		date_now = now()
		assert data.date_visit2 < date_now
		swh_storage.origin_visit_status_add(
		[
		OriginVisitStatus(
		origin=origin.url,
		visit=ov1.visit,
		date=date_now,
		status="full",
		snapshot=complete_snapshot.id,
		)
		]
		)

		actual_snapshot = snapshot_get_latest(
		swh_storage, origin.url, allowed_statuses=["full"]
		)
		assert actual_snapshot is not None
		assert actual_snapshot == complete_snapshot