Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/tests/test_init.py
| # Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 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 | ||||
| import datetime | import datetime | ||||
| import pytest | import pytest | ||||
| import os | import os | ||||
| import subprocess | import subprocess | ||||
| from swh.storage import get_storage | |||||
| from swh.loader.tests import prepare_repository_from_archive, assert_last_visit_matches | from swh.loader.tests import prepare_repository_from_archive, assert_last_visit_matches | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| OriginVisit, | OriginVisit, | ||||
| OriginVisitStatus, | OriginVisitStatus, | ||||
| Snapshot, | Snapshot, | ||||
| SnapshotBranch, | SnapshotBranch, | ||||
| TargetType, | TargetType, | ||||
| ) | ) | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.loader.tests import ( | from swh.loader.tests import ( | ||||
| decode_target, | decode_target, | ||||
| check_snapshot, | check_snapshot, | ||||
| ) | ) | ||||
| hash_hex = "43e45d56f88993aae6a0198013efa80716fd8920" | hash_hex = "43e45d56f88993aae6a0198013efa80716fd8920" | ||||
| storage_config = {"cls": "pipeline", "steps": [{"cls": "memory",}]} | |||||
| ORIGIN_VISIT = OriginVisit( | ORIGIN_VISIT = OriginVisit( | ||||
| origin="some-url", | origin="some-url", | ||||
| visit=1, | visit=1, | ||||
| date=datetime.datetime.now(tz=datetime.timezone.utc), | date=datetime.datetime.now(tz=datetime.timezone.utc), | ||||
| type="archive", | type="archive", | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 152 Lines • ▼ Show 20 Lines | def test_decode_target(): | ||||
| ) | ) | ||||
| assert actual_decode_target == { | assert actual_decode_target == { | ||||
| "target_type": "revision", | "target_type": "revision", | ||||
| "target": hash_hex, | "target": hash_hex, | ||||
| } | } | ||||
| def test_check_snapshot(): | def test_check_snapshot(swh_storage): | ||||
| storage = get_storage(**storage_config) | |||||
| snap_id = "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7" | snap_id = "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7" | ||||
| snapshot = Snapshot( | snapshot = Snapshot( | ||||
| id=hash_to_bytes(snap_id), | id=hash_to_bytes(snap_id), | ||||
| branches={ | branches={ | ||||
| b"master": SnapshotBranch( | b"master": SnapshotBranch( | ||||
| target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION, | target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| s = storage.snapshot_add([snapshot]) | s = swh_storage.snapshot_add([snapshot]) | ||||
| assert s == { | assert s == { | ||||
| "snapshot:add": 1, | "snapshot:add": 1, | ||||
| } | } | ||||
| expected_snapshot = { | expected_snapshot = { | ||||
| "id": snap_id, | "id": snap_id, | ||||
| "branches": {"master": {"target": hash_hex, "target_type": "revision",}}, | "branches": {"master": {"target": hash_hex, "target_type": "revision",}}, | ||||
| } | } | ||||
| check_snapshot(expected_snapshot, storage) | check_snapshot(expected_snapshot, swh_storage) | ||||
| def test_check_snapshot_failure(): | |||||
| storage = get_storage(**storage_config) | |||||
| def test_check_snapshot_failure(swh_storage): | |||||
| snapshot = Snapshot( | snapshot = Snapshot( | ||||
| id=hash_to_bytes("2498dbf535f882bc7f9a18fb16c9ad27fda7bab7"), | id=hash_to_bytes("2498dbf535f882bc7f9a18fb16c9ad27fda7bab7"), | ||||
| branches={ | branches={ | ||||
| b"master": SnapshotBranch( | b"master": SnapshotBranch( | ||||
| target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION, | target=hash_to_bytes(hash_hex), target_type=TargetType.REVISION, | ||||
| ), | ), | ||||
| }, | }, | ||||
| ) | ) | ||||
| s = storage.snapshot_add([snapshot]) | s = swh_storage.snapshot_add([snapshot]) | ||||
| assert s == { | assert s == { | ||||
| "snapshot:add": 1, | "snapshot:add": 1, | ||||
| } | } | ||||
| unexpected_snapshot = { | unexpected_snapshot = { | ||||
| "id": "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7", # id is correct | "id": "2498dbf535f882bc7f9a18fb16c9ad27fda7bab7", # id is correct | ||||
| "branches": { | "branches": { | ||||
| "master": {"target": hash_hex, "target_type": "release",} # wrong branch | "master": {"target": hash_hex, "target_type": "release",} # wrong branch | ||||
| }, | }, | ||||
| } | } | ||||
| with pytest.raises(AssertionError, match="Differing items"): | with pytest.raises(AssertionError, match="Differing items"): | ||||
| check_snapshot(unexpected_snapshot, storage) | check_snapshot(unexpected_snapshot, swh_storage) | ||||
| # snapshot id which does not exist | # snapshot id which does not exist | ||||
| unexpected_snapshot["id"] = "999666f535f882bc7f9a18fb16c9ad27fda7bab7" | unexpected_snapshot["id"] = "999666f535f882bc7f9a18fb16c9ad27fda7bab7" | ||||
| with pytest.raises(AssertionError, match="is not found"): | with pytest.raises(AssertionError, match="is not found"): | ||||
| check_snapshot(unexpected_snapshot, storage) | check_snapshot(unexpected_snapshot, swh_storage) | ||||