Changeset View
Changeset View
Standalone 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(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) | |||||
assert actual_snapshot is not None | |||||
assert actual_snapshot == complete_snapshot | |||||
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) | |||||
assert actual_snapshot is not None | |||||
assert actual_snapshot == complete_snapshot | |||||
actual_snapshot = snapshot_get_latest(swh_storage, origin.url, branches_count=1) | |||||
assert actual_snapshot is not None | |||||
assert actual_snapshot.id == complete_snapshot.id | |||||
assert len(actual_snapshot.branches.values()) == 1 | |||||
with pytest.raises(ValueError, match="branches_count must be a positive integer"): | |||||
snapshot_get_latest(swh_storage, origin.url, branches_count="something-wrong") |