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) |