Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/tests/loader/common.py
| # Copyright (C) 2017-2019 The Software Heritage developers | # Copyright (C) 2017-2019 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 json | import json | ||||
| from typing import Dict | from typing import Dict, Optional | ||||
| from swh.deposit.client import PrivateApiDepositClient | from swh.deposit.client import PrivateApiDepositClient | ||||
| from swh.model.hashutil import hash_to_bytes, hash_to_hex | from swh.model.hashutil import hash_to_bytes, hash_to_hex | ||||
| from swh.model.model import SnapshotBranch, TargetType | |||||
| from swh.storage.algos.snapshot import snapshot_get_all_branches | |||||
| CLIENT_TEST_CONFIG = { | CLIENT_TEST_CONFIG = { | ||||
| "url": "http://nowhere:9000/", | "url": "http://nowhere:9000/", | ||||
| "auth": {}, # no authentication in test scenario | "auth": {}, # no authentication in test scenario | ||||
| } | } | ||||
| class SWHDepositTestClient(PrivateApiDepositClient): | class SWHDepositTestClient(PrivateApiDepositClient): | ||||
| ▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines | keys = [ | ||||
| "release", | "release", | ||||
| "revision", | "revision", | ||||
| "skipped_content", | "skipped_content", | ||||
| "snapshot", | "snapshot", | ||||
| ] | ] | ||||
| return {k: stats.get(k) for k in keys} | return {k: stats.get(k) for k in keys} | ||||
| def decode_target(target): | def decode_target(branch: Optional[SnapshotBranch]) -> Optional[Dict]: | ||||
| """Test helper to ease readability in test | """Test helper to ease readability in test | ||||
| """ | """ | ||||
| if not target: | if not branch: | ||||
| return target | return None | ||||
| target_type = target["target_type"] | target_type = branch.target_type | ||||
| if target_type == "alias": | if target_type == TargetType.ALIAS: | ||||
| decoded_target = target["target"].decode("utf-8") | decoded_target = branch.target.decode("utf-8") | ||||
| else: | else: | ||||
| decoded_target = hash_to_hex(target["target"]) | decoded_target = hash_to_hex(branch.target) | ||||
| return {"target": decoded_target, "target_type": target_type} | return {"target": decoded_target, "target_type": target_type} | ||||
| def check_snapshot(expected_snapshot, storage): | def check_snapshot(expected_snapshot, storage): | ||||
| """Check for snapshot match. | """Check for snapshot match. | ||||
| Provide the hashes as hexadecimal, the conversion is done | Provide the hashes as hexadecimal, the conversion is done | ||||
| within the method. | within the method. | ||||
| Args: | Args: | ||||
| expected_snapshot (dict): full snapshot with hex ids | expected_snapshot (dict): full snapshot with hex ids | ||||
| storage (Storage): expected storage | storage (Storage): expected storage | ||||
| """ | """ | ||||
| expected_snapshot_id = expected_snapshot["id"] | expected_snapshot_id = expected_snapshot["id"] | ||||
| expected_branches = expected_snapshot["branches"] | expected_branches = expected_snapshot["branches"] | ||||
| snap = storage.snapshot_get(hash_to_bytes(expected_snapshot_id)) | snap = snapshot_get_all_branches(hash_to_bytes(expected_snapshot_id)) | ||||
| if snap is None: | if snap is None: | ||||
| # display known snapshots instead if possible | # display known snapshots instead if possible | ||||
| if hasattr(storage, "_snapshots"): # in-mem storage | if hasattr(storage, "_snapshots"): # in-mem storage | ||||
| from pprint import pprint | from pprint import pprint | ||||
| for snap_id, (_snap, _) in storage._snapshots.items(): | for snap_id, (_snap, _) in storage._snapshots.items(): | ||||
| snapd = _snap.to_dict() | snapd = _snap.to_dict() | ||||
| snapd["id"] = hash_to_hex(snapd["id"]) | snapd["id"] = hash_to_hex(snapd["id"]) | ||||
| branches = { | branches = { | ||||
| branch.decode("utf-8"): decode_target(target) | branch.decode("utf-8"): decode_target(target) | ||||
| for branch, target in snapd["branches"].items() | for branch, target in snapd["branches"].items() | ||||
| } | } | ||||
| snapd["branches"] = branches | snapd["branches"] = branches | ||||
| pprint(snapd) | pprint(snapd) | ||||
| raise AssertionError("Snapshot is not found") | raise AssertionError("Snapshot is not found") | ||||
| branches = { | branches = { | ||||
| branch.decode("utf-8"): decode_target(target) | branch.decode("utf-8"): decode_target(branch) | ||||
| for branch, target in snap["branches"].items() | for branch_name, branch in snap["branches"].items() | ||||
| } | } | ||||
| assert expected_branches == branches | assert expected_branches == branches | ||||