Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show All 35 Lines | from swh.model.model import ( | ||||
| Release, | Release, | ||||
| Revision, | Revision, | ||||
| Snapshot, | Snapshot, | ||||
| ) | ) | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.storage import get_storage | 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 HashCollision, StorageArgumentException | from swh.storage.exc import HashCollision, StorageArgumentException | ||||
| from swh.storage.interface import StorageInterface, PagedResult # noqa | from swh.storage.interface import ListOrder, PagedResult, StorageInterface | ||||
| from swh.storage.utils import content_hex_hashes, now | from swh.storage.utils import content_hex_hashes, now | ||||
| @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: | ||||
| yield db, cur | yield db, cur | ||||
| ▲ Show 20 Lines • Show All 1,210 Lines • ▼ Show 20 Lines | def test_origin_visit_get__validation_failure(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| with pytest.raises( | with pytest.raises( | ||||
| StorageArgumentException, match="page_token must be a string" | StorageArgumentException, match="page_token must be a string" | ||||
| ): | ): | ||||
| swh_storage.origin_visit_get(origin.url, page_token=10) # not bytes | swh_storage.origin_visit_get(origin.url, page_token=10) # not bytes | ||||
| with pytest.raises( | with pytest.raises( | ||||
| StorageArgumentException, match="order must be one of asc, desc" | StorageArgumentException, match="order must be a ListOrder value" | ||||
| ): | ): | ||||
| swh_storage.origin_visit_get(origin.url, order="foobar") # wrong order | swh_storage.origin_visit_get(origin.url, order="foobar") # wrong order | ||||
| def test_origin_visit_get_all(self, swh_storage, sample_data): | def test_origin_visit_get_all(self, swh_storage, sample_data): | ||||
| origin = sample_data.origin | origin = sample_data.origin | ||||
| swh_storage.origin_add([origin]) | swh_storage.origin_add([origin]) | ||||
| ov1, ov2, ov3 = swh_storage.origin_visit_add( | ov1, ov2, ov3 = swh_storage.origin_visit_add( | ||||
| [ | [ | ||||
| ▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines | def test_origin_visit_get_all(self, swh_storage, sample_data): | ||||
| next_page_token = str(ov2.visit) | next_page_token = str(ov2.visit) | ||||
| actual_page = swh_storage.origin_visit_get( | actual_page = swh_storage.origin_visit_get( | ||||
| origin.url, page_token=next_page_token, limit=1 | origin.url, page_token=next_page_token, limit=1 | ||||
| ) | ) | ||||
| assert actual_page.next_page_token is None | assert actual_page.next_page_token is None | ||||
| assert actual_page == PagedResult(results=[ov3]) | assert actual_page == PagedResult(results=[ov3]) | ||||
| # order desc, no pagination, no limit | # order desc, no pagination, no limit | ||||
| actual_page = swh_storage.origin_visit_get(origin.url, order="desc") | actual_page = swh_storage.origin_visit_get(origin.url, order=ListOrder.DESC) | ||||
| assert actual_page.next_page_token is None | assert actual_page.next_page_token is None | ||||
| assert actual_page == PagedResult(results=[ov3, ov2, ov1]) | assert actual_page == PagedResult(results=[ov3, ov2, ov1]) | ||||
| # order desc, no pagination, limit | # order desc, no pagination, limit | ||||
| actual_page = swh_storage.origin_visit_get(origin.url, limit=2, order="desc") | actual_page = swh_storage.origin_visit_get( | ||||
| origin.url, limit=2, order=ListOrder.DESC | |||||
| ) | |||||
| next_page_token = actual_page.next_page_token | next_page_token = actual_page.next_page_token | ||||
| assert next_page_token is not None | assert next_page_token is not None | ||||
| assert actual_page.results == [ov3, ov2] | assert actual_page.results == [ov3, ov2] | ||||
| actual_page = swh_storage.origin_visit_get( | actual_page = swh_storage.origin_visit_get( | ||||
| origin.url, page_token=next_page_token, order="desc" | origin.url, page_token=next_page_token, order=ListOrder.DESC | ||||
| ) | ) | ||||
| assert actual_page.next_page_token is None | assert actual_page.next_page_token is None | ||||
| assert actual_page.results == [ov1] | assert actual_page.results == [ov1] | ||||
| assert actual_page == PagedResult(results=[ov1]) | assert actual_page == PagedResult(results=[ov1]) | ||||
| # order desc, pagination, no limit | # order desc, pagination, no limit | ||||
| next_page_token = str(ov3.visit) | next_page_token = str(ov3.visit) | ||||
| actual_page = swh_storage.origin_visit_get( | actual_page = swh_storage.origin_visit_get( | ||||
| origin.url, page_token=next_page_token, order="desc" | origin.url, page_token=next_page_token, order=ListOrder.DESC | ||||
| ) | ) | ||||
| assert actual_page.next_page_token is None | assert actual_page.next_page_token is None | ||||
| assert actual_page == PagedResult(results=[ov2, ov1]) | assert actual_page == PagedResult(results=[ov2, ov1]) | ||||
| # order desc, pagination, limit | # order desc, pagination, limit | ||||
| actual_page = swh_storage.origin_visit_get( | actual_page = swh_storage.origin_visit_get( | ||||
| origin.url, page_token=next_page_token, order="desc", limit=1 | origin.url, page_token=next_page_token, order=ListOrder.DESC, limit=1 | ||||
| ) | ) | ||||
| next_page_token = actual_page.next_page_token | next_page_token = actual_page.next_page_token | ||||
| assert next_page_token is not None | assert next_page_token is not None | ||||
| assert actual_page.results == [ov2] | assert actual_page.results == [ov2] | ||||
| actual_page = swh_storage.origin_visit_get( | actual_page = swh_storage.origin_visit_get( | ||||
| origin.url, page_token=next_page_token, order="desc" | origin.url, page_token=next_page_token, order=ListOrder.DESC | ||||
| ) | ) | ||||
| assert actual_page == PagedResult(results=[ov1]) | assert actual_page == PagedResult(results=[ov1]) | ||||
| def test_origin_visit_status_get_random(self, swh_storage, sample_data): | def test_origin_visit_status_get_random(self, swh_storage, sample_data): | ||||
| origins = sample_data.origins[:2] | origins = sample_data.origins[:2] | ||||
| swh_storage.origin_add(origins) | swh_storage.origin_add(origins) | ||||
| # Add some random visits within the selection range | # Add some random visits within the selection range | ||||
| ▲ Show 20 Lines • Show All 2,868 Lines • Show Last 20 Lines | |||||