Differential D6796 Diff 24766 swh/objstorage/tests/test_objstorage_winery.py

Changeset 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 pytest
				import sh

				from swh.objstorage import exc
				from swh.objstorage.backends.winery.database import Database
				from swh.objstorage.backends.winery.objstorage import Packer, pack
				from swh.objstorage.backends.winery.roshard import Pool
				from swh.objstorage.factory import get_objstorage


				@pytest.fixture
				def needs_ceph():
				try:
				sh.ceph("--version")
				except sh.CommandNotFound:
				pytest.skip("the ceph CLI was not found")


				@pytest.fixture
				def ceph_pool(needs_ceph):
				pool = Pool(shard_max_size=10 * 1024 * 1024)
				pool.init()
				pool.clobber()
				pool.pool_create()

				yield pool

				pool.images_clobber()
				pool.clobber()


				@pytest.fixture
				def winery(request, postgresql):
				marker = request.node.get_closest_marker("shard_max_size")
				if marker is None:
				shard_max_size = 1024
				else:
				shard_max_size = marker.args[0]
				dsn = (
				f"postgres://{postgresql.info.user}"
				f":@{postgresql.info.host}:{postgresql.info.port}"
				)
				winery = get_objstorage(
				cls="winery", base_dsn=dsn, shard_dsn=dsn, shard_max_size=shard_max_size
				)
				yield winery
				winery.uninit()
				#
				# pytest-postgresql will not remove databases that it did not
				# create between tests (only at the very end).
				#
				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(winery):
				base = winery.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(winery):
				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")
				winery.shard.drop()


				@pytest.mark.shard_max_size(1)
				def test_winery_add_and_pack(winery, mocker):
				mocker.patch("swh.objstorage.backends.winery.objstorage.pack", return_value=True)
				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
				packer = winery.packers[0]
				packer.join()
				assert packer.exitcode == 0


				def test_winery_delete(winery):
				with pytest.raises(PermissionError):
				winery.delete(None)


				def test_winery_get_shard_info(winery):
				assert winery.base.get_shard_info(1234) is None
				assert winery.base.get_shard_info_by_name("nothing") is None


				@pytest.mark.shard_max_size(10 * 1024 * 1024)
				def test_winery_packer(winery, ceph_pool):
				shard = winery.base.whoami
				content = b"SOMETHING"
				winery.add(content=content)
				winery.base.shard_packing_starts()
				packer = Packer(shard, **winery.args)
				try:
				assert packer.run() is True
				finally:
				packer.uninit()

				readonly, packing = winery.base.get_shard_info_by_name(shard)
				assert readonly is True
				assert packing is False


				@pytest.mark.shard_max_size(10 * 1024 * 1024)
				def test_winery_get_object(winery, ceph_pool):
				shard = winery.base.whoami
				content = b"SOMETHING"
				obj_id = winery.add(content=content)
				winery.base.shard_packing_starts()
				assert pack(shard, **winery.args) is True
				winery.base.get_shard_info_by_name(shard)
				assert winery.get(obj_id) == content


				def test_winery_ceph_pool(needs_ceph):
				name = "IMAGE"
				pool = Pool(shard_max_size=10 * 1024 * 1024)
				pool.init()
				pool.clobber()
				pool.pool_create()
				pool.image_create(name)
				p = pool.image_path(name)
				assert p.endswith(name)
				something = "SOMETHING"
				open(p, "w").write(something)
				assert open(p).read(len(something)) == something
				assert pool.image_list() == [name]
				pool.image_remap_ro(name)
				pool.images_clobber()
				assert pool.image_list() == [name]
				pool.clobber()
				assert pool.image_list() == []