Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_winery.py
- This file was added.
| # Copyright (C) 2021 The Software Heritage developers | |||||
| # See the AUTHORS file at the top-level directory of this distribution | |||||
| # License: GNU General Public License version 3, or any later version | |||||
| # See top-level LICENSE file for more information | |||||
| import concurrent.futures | |||||
| import pytest | |||||
| from swh.objstorage import exc | |||||
| from swh.objstorage.backends.winery.database import Database | |||||
| from swh.objstorage.factory import get_objstorage | |||||
| @pytest.fixture | |||||
| def dsn(postgresql): | |||||
| dsn = ( | |||||
| f"postgres://{postgresql.info.user}" | |||||
| f":@{postgresql.info.host}:{postgresql.info.port}" | |||||
| ) | |||||
| yield dsn | |||||
| # pytest-postgresql will not remove databases that it did not create | |||||
| d = Database(dsn) | |||||
| for database in d.list_databases(): | |||||
| if database != postgresql.info.dbname and database != "tests_tmpl": | |||||
| d.drop_database(database) | |||||
| def test_winery_sharedbase(dsn): | |||||
| base = get_objstorage( | |||||
| cls="winery", base_dsn=dsn, shard_dsn=dsn, shard_max_size=1 | |||||
| ).base | |||||
| shard1 = base.whoami | |||||
| assert shard1 is not None | |||||
| assert shard1 == base.whoami | |||||
| id1 = base.id | |||||
| assert id1 is not None | |||||
| assert id1 == base.id | |||||
| def test_winery_add_get(dsn): | |||||
| winery = get_objstorage( | |||||
| cls="winery", base_dsn=dsn, shard_dsn=dsn, shard_max_size=1024 | |||||
| ) | |||||
| shard = winery.base.whoami | |||||
| content = b"SOMETHING" | |||||
| obj_id = winery.add(content=content) | |||||
| assert obj_id.hex() == "0c8c841f7d9fd4874d841506d3ffc16808b1d579" | |||||
| assert winery.add(content=content, obj_id=obj_id) == obj_id | |||||
| assert winery.add(content=content, obj_id=obj_id, check_presence=False) == obj_id | |||||
| assert winery.base.whoami == shard | |||||
| assert winery.get(obj_id) == content | |||||
| with pytest.raises(exc.ObjNotFoundError): | |||||
| winery.get(b"unknown") | |||||
| def test_winery_add_and_pack(dsn): | |||||
| winery = get_objstorage(cls="winery", base_dsn=dsn, shard_dsn=dsn, shard_max_size=1) | |||||
| shard = winery.base.whoami | |||||
| content = b"SOMETHING" | |||||
| obj_id = winery.add(content=content) | |||||
| assert obj_id.hex() == "0c8c841f7d9fd4874d841506d3ffc16808b1d579" | |||||
| assert winery.base.whoami != shard | |||||
| assert len(winery.packers) == 1 | |||||
| done, not_done = concurrent.futures.wait(winery.packers) | |||||
| assert len(not_done) == 0 | |||||
| assert len(done) == 1 | |||||
| packer = list(done)[0] | |||||
| assert packer.exception(timeout=0) is None | |||||
| assert packer.result() is True | |||||
| winery.executor.shutdown() | |||||
| def test_winery_delete(dsn): | |||||
| winery = get_objstorage(cls="winery", base_dsn=dsn, shard_dsn=dsn, shard_max_size=1) | |||||
| with pytest.raises(PermissionError): | |||||
| winery.delete(None) | |||||