Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/tests/__init__.py
| Show First 20 Lines • Show All 129 Lines • ▼ Show 20 Lines | Args: | ||||
| to target the nixpkgs git commit reference, which may not yet be resolvable at | to target the nixpkgs git commit reference, which may not yet be resolvable at | ||||
| loading time) | loading time) | ||||
| Returns: | Returns: | ||||
| the snapshot stored in the storage for further test assertion if any is | the snapshot stored in the storage for further test assertion if any is | ||||
| needed. | needed. | ||||
| """ | """ | ||||
| if isinstance(snapshot, Snapshot): | if isinstance(snapshot, dict): | ||||
| expected_snapshot = snapshot | snapshot = Snapshot.from_dict(snapshot) | ||||
| elif isinstance(snapshot, dict): | if not isinstance(snapshot, Snapshot): | ||||
| # dict must be snapshot compliant | |||||
| snapshot_dict = {"id": hash_to_bytes(snapshot["id"])} | |||||
| branches = {} | |||||
| for branch, target in snapshot["branches"].items(): | |||||
| if isinstance(branch, str): | |||||
| branch = branch.encode("utf-8") | |||||
| branches[branch] = encode_target(target) | |||||
| snapshot_dict["branches"] = branches | |||||
| expected_snapshot = Snapshot.from_dict(snapshot_dict) | |||||
| else: | |||||
| raise AssertionError(f"variable 'snapshot' must be a snapshot: {snapshot!r}") | raise AssertionError(f"variable 'snapshot' must be a snapshot: {snapshot!r}") | ||||
| snapshot_dict = storage.snapshot_get(expected_snapshot.id) | snapshot_dict = storage.snapshot_get(snapshot.id) | ||||
| if snapshot_dict is None: | if snapshot_dict is None: | ||||
| raise AssertionError(f"Snapshot {expected_snapshot.id.hex()} is not found") | raise AssertionError(f"Snapshot {snapshot.id.hex()} is not found") | ||||
| snapshot_dict.pop("next_branch") | snapshot_dict.pop("next_branch") | ||||
| actual_snaphot = Snapshot.from_dict(snapshot_dict) | actual_snaphot = Snapshot.from_dict(snapshot_dict) | ||||
| assert isinstance(actual_snaphot, Snapshot) | assert isinstance(actual_snaphot, Snapshot) | ||||
| assert expected_snapshot == actual_snaphot | assert snapshot == actual_snaphot | ||||
| branches_by_target_type = defaultdict(list) | branches_by_target_type = defaultdict(list) | ||||
| object_to_branch = {} | object_to_branch = {} | ||||
| for branch, target in actual_snaphot.branches.items(): | for branch, target in actual_snaphot.branches.items(): | ||||
| if (target.target_type, branch) in allowed_empty: | if (target.target_type, branch) in allowed_empty: | ||||
| # safe for those elements to not be checked for existence | # safe for those elements to not be checked for existence | ||||
| continue | continue | ||||
| branches_by_target_type[target.target_type].append(target.target) | branches_by_target_type[target.target_type].append(target.target) | ||||
| ▲ Show 20 Lines • Show All 59 Lines • Show Last 20 Lines | |||||