Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show All 24 Lines | |||||
| from typing import ClassVar, Optional | from typing import ClassVar, Optional | ||||
| from swh.model import from_disk, identifiers | from swh.model import from_disk, identifiers | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| Content, OriginVisit, Release, Revision | Content, OriginVisit, Release, Revision | ||||
| ) | ) | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.storage import HashCollision, get_storage | from swh.model.hashutil import hash_to_hex | ||||
| from swh.storage import get_storage | |||||
| from swh.storage.converters import origin_url_to_sha1 as sha1 | from swh.storage.converters import origin_url_to_sha1 as sha1 | ||||
| from swh.storage.exc import StorageArgumentException | from swh.storage.exc import HashCollision, StorageArgumentException | ||||
| from swh.storage.interface import StorageInterface | from swh.storage.interface import StorageInterface | ||||
| from swh.storage.utils import content_hex_hashes | |||||
| from .storage_data import data | from .storage_data import data | ||||
| @contextmanager | @contextmanager | ||||
| def db_transaction(storage): | def db_transaction(storage): | ||||
| with storage.db() as db: | with storage.db() as db: | ||||
| with db.transaction() as cur: | with db.transaction() as cur: | ||||
| ▲ Show 20 Lines • Show All 264 Lines • ▼ Show 20 Lines | def test_content_add_collision(self, swh_storage): | ||||
| cont1b = cont1.copy() | cont1b = cont1.copy() | ||||
| sha256_array = bytearray(cont1b['sha256']) | sha256_array = bytearray(cont1b['sha256']) | ||||
| sha256_array[0] += 1 | sha256_array[0] += 1 | ||||
| cont1b['sha256'] = bytes(sha256_array) | cont1b['sha256'] = bytes(sha256_array) | ||||
| with pytest.raises(HashCollision) as cm: | with pytest.raises(HashCollision) as cm: | ||||
| swh_storage.content_add([cont1, cont1b]) | swh_storage.content_add([cont1, cont1b]) | ||||
| actual_algo = cm.value.args[0] | exc = cm.value | ||||
| actual_algo = exc.algo | |||||
| assert actual_algo in ['sha1', 'sha1_git', 'blake2s256'] | assert actual_algo in ['sha1', 'sha1_git', 'blake2s256'] | ||||
| actual_id = cm.value.args[1] | actual_id = exc.hash_id | ||||
| assert actual_id == cont1[actual_algo] | assert actual_id == hash_to_hex(cont1[actual_algo]) | ||||
| assert len(cm.value.args[2]) == 2 | collisions = exc.args[2] | ||||
| assert cm.value.args[2] == [ | assert len(collisions) == 2 | ||||
| assert collisions == [ | |||||
| content_hex_hashes(Content.from_dict(cont1).hashes()), | |||||
| content_hex_hashes(Content.from_dict(cont1b).hashes()) | |||||
| ] | |||||
| assert exc.colliding_content_hashes() == [ | |||||
| Content.from_dict(cont1).hashes(), | Content.from_dict(cont1).hashes(), | ||||
| Content.from_dict(cont1b).hashes() | Content.from_dict(cont1b).hashes() | ||||
| ] | ] | ||||
| def test_content_update(self, swh_storage): | def test_content_update(self, swh_storage): | ||||
| if hasattr(swh_storage, 'storage'): | if hasattr(swh_storage, 'storage'): | ||||
| swh_storage.journal_writer.journal = None # TODO, not supported | swh_storage.journal_writer.journal = None # TODO, not supported | ||||
| ▲ Show 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | def test_content_add_metadata_collision(self, swh_storage): | ||||
| cont1b = cont1.copy() | cont1b = cont1.copy() | ||||
| sha1_git_array = bytearray(cont1b['sha256']) | sha1_git_array = bytearray(cont1b['sha256']) | ||||
| sha1_git_array[0] += 1 | sha1_git_array[0] += 1 | ||||
| cont1b['sha256'] = bytes(sha1_git_array) | cont1b['sha256'] = bytes(sha1_git_array) | ||||
| with pytest.raises(HashCollision) as cm: | with pytest.raises(HashCollision) as cm: | ||||
| swh_storage.content_add_metadata([cont1, cont1b]) | swh_storage.content_add_metadata([cont1, cont1b]) | ||||
| actual_algo = cm.value.args[0] | exc = cm.value | ||||
| actual_algo = exc.algo | |||||
| assert actual_algo in ['sha1', 'sha1_git', 'blake2s256'] | assert actual_algo in ['sha1', 'sha1_git', 'blake2s256'] | ||||
| actual_id = cm.value.args[1] | actual_id = exc.hash_id | ||||
| assert actual_id == cont1[actual_algo] | assert actual_id == hash_to_hex(cont1[actual_algo]) | ||||
| assert len(cm.value.args[2]) == 2 | collisions = exc.args[2] | ||||
| assert cm.value.args[2] == [ | assert len(collisions) == 2 | ||||
| assert collisions == [ | |||||
| content_hex_hashes(Content.from_dict(cont1).hashes()), | |||||
| content_hex_hashes(Content.from_dict(cont1b).hashes()) | |||||
| ] | |||||
| assert exc.colliding_content_hashes() == [ | |||||
| Content.from_dict(cont1).hashes(), | Content.from_dict(cont1).hashes(), | ||||
| Content.from_dict(cont1b).hashes() | Content.from_dict(cont1b).hashes() | ||||
| ] | ] | ||||
| def test_skipped_content_add(self, swh_storage): | def test_skipped_content_add(self, swh_storage): | ||||
| cont = data.skipped_cont | cont = data.skipped_cont | ||||
| cont2 = data.skipped_cont2 | cont2 = data.skipped_cont2 | ||||
| cont2['blake2s256'] = None | cont2['blake2s256'] = None | ||||
| ▲ Show 20 Lines • Show All 3,436 Lines • Show Last 20 Lines | |||||