Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show All 17 Lines | |||||
| from hypothesis import given, strategies, settings, HealthCheck | from hypothesis import given, strategies, settings, HealthCheck | ||||
| 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.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.storage import HashCollision | from swh.storage import HashCollision | ||||
| from swh.storage.converters import origin_url_to_sha1 as sha1 | |||||
| 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 896 Lines • ▼ Show 20 Lines | def test_origin_get(self, swh_storage): | ||||
| assert swh_storage.origin_get(data.origin) is None | assert swh_storage.origin_get(data.origin) is None | ||||
| swh_storage.origin_add_one(data.origin) | swh_storage.origin_add_one(data.origin) | ||||
| actual_origin0 = swh_storage.origin_get( | actual_origin0 = swh_storage.origin_get( | ||||
| [{'url': data.origin['url']}]) | [{'url': data.origin['url']}]) | ||||
| assert len(actual_origin0) == 1 | assert len(actual_origin0) == 1 | ||||
| assert actual_origin0[0]['url'] == data.origin['url'] | assert actual_origin0[0]['url'] == data.origin['url'] | ||||
| def test_origin_get_by_sha1(self, swh_storage): | |||||
| assert swh_storage.origin_get(data.origin) is None | |||||
| swh_storage.origin_add_one(data.origin) | |||||
| origins = list(swh_storage.origin_get_by_sha1([ | |||||
| sha1(data.origin['url']) | |||||
| ])) | |||||
| assert len(origins) == 1 | |||||
| assert origins[0]['url'] == data.origin['url'] | |||||
| def test_origin_get_by_sha1_not_found(self, swh_storage): | |||||
| assert swh_storage.origin_get(data.origin) is None | |||||
| origins = list(swh_storage.origin_get_by_sha1([ | |||||
| sha1(data.origin['url']) | |||||
| ])) | |||||
| assert len(origins) == 1 | |||||
| assert origins[0] is None | |||||
| def test_origin_search_single_result(self, swh_storage): | def test_origin_search_single_result(self, swh_storage): | ||||
| found_origins = list(swh_storage.origin_search(data.origin['url'])) | found_origins = list(swh_storage.origin_search(data.origin['url'])) | ||||
| assert len(found_origins) == 0 | assert len(found_origins) == 0 | ||||
| found_origins = list(swh_storage.origin_search(data.origin['url'], | found_origins = list(swh_storage.origin_search(data.origin['url'], | ||||
| regexp=True)) | regexp=True)) | ||||
| assert len(found_origins) == 0 | assert len(found_origins) == 0 | ||||
| ▲ Show 20 Lines • Show All 2,478 Lines • Show Last 20 Lines | |||||