Differential D2640 Diff 9443 swh/storage/tests/test_cassandra.py

Changeset View

Standalone View

Show First 20 Lines • Show All 143 Lines • ▼ Show 20 Lines	def keyspace(cassandra_cluster):
return keyspace		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, keyspace):		def swh_storage_backend_config(cassandra_cluster, keyspace):
(hosts, port) = cassandra_cluster		(hosts, port) = cassandra_cluster

storage = get_storage(		storage_config = dict(
'cassandra',		cls='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_config

		storage = get_storage(**storage_config)

for table in TABLES:		for table in TABLES:
storage._cql_runner._session.execute('TRUNCATE TABLE "%s"' % table)		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')
▲ Show 20 Lines • Show All 93 Lines • Show Last 20 Lines