Changeset View
Changeset View
Standalone 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.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.model.model import Origin, 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 | ||||
| from swh.storage.utils import now | from swh.storage.utils import now | ||||
| from swh.storage.tests.storage_data import data | from swh.storage.tests.storage_data import data | ||||
| Show All 11 Lines | def test_snapshot_small(swh_storage, snapshot): # noqa | ||||
| 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 | ||||
| @given(branch_name=branch_names(), branch_target=branch_targets(only_objects=True)) | @given(branch_name=branch_names(), branch_target=branch_targets(only_objects=True)) | ||||
| def test_snapshot_large(swh_storage, branch_name, branch_target): # noqa | def test_snapshot_large(swh_storage, branch_name, branch_target): # noqa | ||||
| branch_target = branch_target.to_dict() | snapshot = Snapshot( | ||||
| branches=ImmutableDict( | |||||
| snapshot = { | (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)) | |||||
| 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.to_dict() == returned_snapshot | ||||
| def test_snapshot_get_latest_none(swh_storage): | def test_snapshot_get_latest_none(swh_storage): | ||||
| """Retrieve latest snapshot on unknown origin or origin without snapshot should | """Retrieve latest snapshot on unknown origin or origin without snapshot should | ||||
| yield no result | yield no result | ||||
| """ | """ | ||||
| # unknown origin so None | # unknown origin so None | ||||
| ▲ Show 20 Lines • Show All 96 Lines • Show Last 20 Lines | |||||