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 |