Differential D5974 Diff 21548 swh/storage/tests/test_cassandra.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_cassandra.py

# 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 datetime

import itertools

import os

import resource

import signal

import socket

import subprocess

import time

from typing import Any, Dict

import attr

from cassandra.cluster import NoHostAvailable

import pytest

from swh.core.api.classes import stream_results

from swh.model.model import Directory, DirectoryEntry, Snapshot, SnapshotBranch

from swh.storage import get_storage

from swh.storage.cassandra import create_keyspace

from swh.storage.cassandra.model import ContentRow, ExtIDRow

from swh.storage.cassandra.schema import HASH_ALGORITHMS, TABLES

▲ Show 20 Lines • Show All 207 Lines • ▼ Show 20 Lines

def swh_storage_backend_config(cassandra_cluster, keyspace):

for table in TABLES:

storage._cql_runner._session.execute('TRUNCATE TABLE "%s"' % table)

storage._cql_runner._cluster.shutdown()

@pytest.mark.cassandra

class TestCassandraStorage(_TestStorage):

def test_config_wrong_consistency_should_raise(self):

storage_config = dict(

cls="cassandra",

hosts=["first"],

port=9999,

keyspace="any",

consistency_level="fake",

journal_writer={"cls": "memory"},

objstorage={"cls": "memory"},

)

ardumontUnsubmitted

Not Done

objstorage={"cls": "memory"},

)

- with pytest.raises(ValueError):

+ with pytest.raises(ValueError, match="unknown"):

get_storage(**storage_config)

Not to check the full log of the exception but to check the at the very least, it raises for the right reason.

ardumont: Not to check the full log of the exception but to check the at the very least, it raises for…

with pytest.raises(ValueError, match="Unknown consistency"):

get_storage(**storage_config)

def test_config_consistency_used(self, swh_storage_backend_config):

config_with_consistency = dict(

swh_storage_backend_config, **{"consistency_level": "THREE"}

)

storage = get_storage(**config_with_consistency)

with pytest.raises(NoHostAvailable):

storage.content_get_random()

def test_content_add_murmur3_collision(self, swh_storage, mocker, sample_data):

"""The Murmur3 token is used as link from index tables to the main

table; and non-matching contents with colliding murmur3-hash

are filtered-out when reading the main table.

This test checks the content methods do filter out these collision.

"""

called = 0

▲ Show 20 Lines • Show All 431 Lines • Show Last 20 Lines