Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/tests/__init__.py
| Show First 20 Lines • Show All 109 Lines • ▼ Show 20 Lines | |||||
| class InexistentObjectsError(AssertionError): | class InexistentObjectsError(AssertionError): | ||||
| """When a targeted branch reference does not exist in the storage""" | """When a targeted branch reference does not exist in the storage""" | ||||
| pass | pass | ||||
| def check_snapshot( | def check_snapshot( | ||||
| snapshot: Snapshot, | expected_snapshot: Snapshot, | ||||
| storage: StorageInterface, | storage: StorageInterface, | ||||
| allowed_empty: Iterable[Tuple[TargetType, bytes]] = [], | allowed_empty: Iterable[Tuple[TargetType, bytes]] = [], | ||||
| ) -> Snapshot: | ) -> Snapshot: | ||||
| """Check that: | """Check that: | ||||
| - snapshot exists in the storage and match | - snapshot exists in the storage and match | ||||
| - each object reference up to the revision/release targets exists | - each object reference up to the revision/release targets exists | ||||
| Args: | Args: | ||||
| snapshot: full snapshot to check for existence and consistency | expected_snapshot: full snapshot to check for existence and consistency | ||||
| storage: storage to lookup information into | storage: storage to lookup information into | ||||
| allowed_empty: Iterable of branch we allow to be empty (some edge case loaders | allowed_empty: Iterable of branch we allow to be empty (some edge case loaders | ||||
| allows this case to happen, nixguix for example allows the branch evaluation" | allows this case to happen, nixguix for example allows the branch evaluation" | ||||
| 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 not isinstance(snapshot, Snapshot): | if not isinstance(expected_snapshot, Snapshot): | ||||
| raise AssertionError(f"variable 'snapshot' must be a snapshot: {snapshot!r}") | raise AssertionError( | ||||
| f"argument 'expected_snapshot' must be a snapshot: {expected_snapshot!r}" | |||||
| ) | |||||
| expected_snapshot = snapshot_get_all_branches(storage, snapshot.id) | snapshot = snapshot_get_all_branches(storage, expected_snapshot.id) | ||||
| if expected_snapshot is None: | if snapshot is None: | ||||
| raise AssertionError(f"Snapshot {snapshot.id.hex()} is not found") | raise AssertionError(f"Snapshot {expected_snapshot.id.hex()} is not found") | ||||
| assert snapshot == expected_snapshot | assert snapshot == expected_snapshot | ||||
| objects_by_target_type = defaultdict(list) | objects_by_target_type = defaultdict(list) | ||||
| object_to_branch = {} | object_to_branch = {} | ||||
| for branch, target in expected_snapshot.branches.items(): | for branch, target in expected_snapshot.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 | ||||
| ▲ Show 20 Lines • Show All 109 Lines • Show Last 20 Lines | |||||