Differential D2615 Diff 9354 swh/storage/tests/test_cassandra.py

Changeset View

Standalone View

swh/storage/tests/test_cassandra.py

	# Copyright (C) 2018-2019 The Software Heritage developers			# Copyright (C) 2018-2019 The Software Heritage developers
	# See the AUTHORS file at the top-level directory of this distribution			# See the AUTHORS file at the top-level directory of this distribution
	# License: GNU General Public License version 3, or any later version			# License: GNU General Public License version 3, or any later version
	# See top-level LICENSE file for more information			# See top-level LICENSE file for more information

	import os			import os
	import signal			import signal
	import socket			import socket
	import subprocess			import subprocess
	import time			import time

	import pytest			import pytest

	from swh.storage import get_storage			from swh.storage import get_storage
	from swh.storage.cassandra import create_keyspace			from swh.storage.cassandra import create_keyspace
				from swh.storage.cassandra.schema import TABLES

	from swh.storage.tests.test_storage import TestStorage as _TestStorage			from swh.storage.tests.test_storage import TestStorage as _TestStorage
	from swh.storage.tests.test_storage import TestStorageGeneratedData \			from swh.storage.tests.test_storage import TestStorageGeneratedData \
	as _TestStorageGeneratedData			as _TestStorageGeneratedData


	CONFIG_TEMPLATE = '''			CONFIG_TEMPLATE = '''
	data_file_directories:			data_file_directories:
	▲ Show 20 Lines • Show All 98 Lines • ▼ Show 20 Lines

	class RequestHandler:			class RequestHandler:
	def on_request(self, rf):			def on_request(self, rf):
	if hasattr(rf.message, 'query'):			if hasattr(rf.message, 'query'):
	print()			print()
	print(rf.message.query)			print(rf.message.query)


				@pytest.fixture(scope='session')
				def keyspace(cassandra_cluster):
				(hosts, port) = cassandra_cluster
				keyspace = os.urandom(10).hex()

				create_keyspace(hosts, keyspace, port)

				return keyspace


	# tests are executed using imported classes (TestStorage and			# tests are executed using imported classes (TestStorage and
	# TestStorageGeneratedData) using overloaded swh_storage fixture			# TestStorageGeneratedData) using overloaded swh_storage fixture
	# below			# below

	@pytest.fixture			@pytest.fixture
	def swh_storage(cassandra_cluster):			def swh_storage(cassandra_cluster, keyspace):
	(hosts, port) = cassandra_cluster			(hosts, port) = cassandra_cluster
	keyspace = os.urandom(10).hex()

	create_keyspace(hosts, keyspace, port)

	storage = get_storage(			storage = get_storage(
	'cassandra',			'cassandra',
	hosts=hosts, port=port,			hosts=hosts, port=port,
	keyspace=keyspace,			keyspace=keyspace,
	journal_writer={			journal_writer={
	'cls': 'memory',			'cls': 'memory',
	},			},
	objstorage={			objstorage={
	'cls': 'memory',			'cls': 'memory',
	'args': {},			'args': {},
	},			},
	)			)

	yield storage			yield storage

	storage._cql_runner._session.execute(			for table in TABLES:
	'DROP KEYSPACE "%s"' % keyspace)			storage._cql_runner._session.execute('TRUNCATE TABLE "%s"' % table)


	@pytest.mark.cassandra			@pytest.mark.cassandra
	class TestCassandraStorage(_TestStorage):			class TestCassandraStorage(_TestStorage):
	@pytest.mark.skip('content_update is not yet implemented for Cassandra')			@pytest.mark.skip('content_update is not yet implemented for Cassandra')
	def test_content_update(self):			def test_content_update(self):
	pass			pass

	▲ Show 20 Lines • Show All 95 Lines • Show Last 20 Lines