Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/validate.py
| Show All 39 Lines | def _check_hashes(self, objects: Iterable): | ||||
| for obj in objects: | for obj in objects: | ||||
| id_ = hash_to_bytes(obj.compute_hash()) | id_ = hash_to_bytes(obj.compute_hash()) | ||||
| if id_ != obj.id: | if id_ != obj.id: | ||||
| raise StorageArgumentException( | raise StorageArgumentException( | ||||
| f"Object has id {hash_to_hex(obj.id)}, " | f"Object has id {hash_to_hex(obj.id)}, " | ||||
| f"but it should be {hash_to_hex(id_)}: {obj}" | f"but it should be {hash_to_hex(id_)}: {obj}" | ||||
| ) | ) | ||||
| def content_add(self, content: List[Content]) -> Dict: | def content_add(self, content: List[Content]) -> Dict[str, int]: | ||||
| for cont in content: | for cont in content: | ||||
| hashes = MultiHash.from_data(cont.data).digest() | hashes = MultiHash.from_data(cont.data).digest() | ||||
| if hashes != cont.hashes(): | if hashes != cont.hashes(): | ||||
| raise StorageArgumentException( | raise StorageArgumentException( | ||||
| f"Object has hashes {cont.hashes()}, but they should be {hashes}" | f"Object has hashes {cont.hashes()}, but they should be {hashes}" | ||||
| ) | ) | ||||
| return self.storage.content_add(content) | return self.storage.content_add(content) | ||||
| def directory_add(self, directories: List[Directory]) -> Dict: | def directory_add(self, directories: List[Directory]) -> Dict[str, int]: | ||||
| self._check_hashes(directories) | self._check_hashes(directories) | ||||
| return self.storage.directory_add(directories) | return self.storage.directory_add(directories) | ||||
| def revision_add(self, revisions: List[Revision]) -> Dict: | def revision_add(self, revisions: List[Revision]) -> Dict[str, int]: | ||||
| self._check_hashes(revisions) | self._check_hashes(revisions) | ||||
| return self.storage.revision_add(revisions) | return self.storage.revision_add(revisions) | ||||
| def release_add(self, releases: List[Release]) -> Dict: | def release_add(self, releases: List[Release]) -> Dict[str, int]: | ||||
| self._check_hashes(releases) | self._check_hashes(releases) | ||||
| return self.storage.release_add(releases) | return self.storage.release_add(releases) | ||||
| def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: | def snapshot_add(self, snapshots: List[Snapshot]) -> Dict[str, int]: | ||||
| self._check_hashes(snapshots) | self._check_hashes(snapshots) | ||||
| return self.storage.snapshot_add(snapshots) | return self.storage.snapshot_add(snapshots) | ||||