openstack/deb-python-cassandra-driver
Code Issues Proposed changes

Merged into master support for testing with remote cluster

Browse Source
This commit is contained in:
bjmb
2017-03-15 14:05:20 -04:00
parent 8a2bf3fb9c 8252196a0d
commit 79e6a43c95
18 changed files with 143 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/__init__.py
View File

@@ -12,9 +12,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.
try:
import unittest2 as unittest
except ImportError:
import unittest # noqa
import logging
import sys
import socket
import platform
log = logging.getLogger()
log.setLevel('DEBUG')
@@ -41,3 +46,5 @@ def is_gevent_monkey_patched():
def is_monkey_patched():
return is_gevent_monkey_patched() or is_eventlet_monkey_patched()
notwindows = unittest.skipUnless(not "Windows" in platform.system(), "This test is not adequate for windows")

16
tests/integration/__init__.py
View File

@@ -115,6 +115,7 @@ def _get_cass_version_from_dse(dse_version):
return cass_ver
CASSANDRA_IP = os.getenv('CASSANDRA_IP', '127.0.0.1')
CASSANDRA_DIR = os.getenv('CASSANDRA_DIR', None)
DSE_VERSION = os.getenv('DSE_VERSION', None)
DSE_CRED = os.getenv('DSE_CREDS', None)
@@ -141,6 +142,18 @@ if DSE_VERSION:
CCM_KWARGS['dse_credentials_file'] = DSE_CRED
#This changes the default contact_point parameter in Cluster
def set_default_cass_ip():
if CASSANDRA_IP.startswith("127.0.0."):
return
defaults = list(Cluster.__init__.__defaults__)
defaults = [[CASSANDRA_IP]] + defaults[1:]
try:
Cluster.__init__.__defaults__ = tuple(defaults)
except:
Cluster.__init__.__func__.__defaults__ = tuple(defaults)
def get_default_protocol():
if Version(CASSANDRA_VERSION) >= Version('2.2'):
@@ -208,6 +221,7 @@ default_protocol_version = get_default_protocol()
PROTOCOL_VERSION = int(os.getenv('PROTOCOL_VERSION', default_protocol_version))
local = unittest.skipUnless(CASSANDRA_IP.startswith("127.0.0."), 'Tests only runs against local C*')
notprotocolv1 = unittest.skipUnless(PROTOCOL_VERSION > 1, 'Protocol v1 not supported')
lessthenprotocolv4 = unittest.skipUnless(PROTOCOL_VERSION < 4, 'Protocol versions 4 or greater not supported')
greaterthanprotocolv3 = unittest.skipUnless(PROTOCOL_VERSION >= 4, 'Protocol versions less than 4 are not supported')
@@ -299,6 +313,8 @@ def is_current_cluster(cluster_name, node_counts):
def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=[]):
set_default_cass_ip()
global CCM_CLUSTER
if USE_CASS_EXTERNAL:
if CCM_CLUSTER:

5
tests/integration/cqlengine/__init__.py
View File

@@ -24,7 +24,7 @@ from cassandra.cqlengine import connection
from cassandra.cqlengine.management import create_keyspace_simple, CQLENG_ALLOW_SCHEMA_MANAGEMENT
import cassandra
from tests.integration import get_server_versions, use_single_node, PROTOCOL_VERSION
from tests.integration import get_server_versions, use_single_node, PROTOCOL_VERSION, CASSANDRA_IP, set_default_cass_ip
DEFAULT_KEYSPACE = 'cqlengine_test'
@@ -35,6 +35,7 @@ def setup_package():
warnings.simplefilter('always') # for testing warnings, make sure all are let through
os.environ[CQLENG_ALLOW_SCHEMA_MANAGEMENT] = '1'
set_default_cass_ip()
use_single_node()
setup_connection(DEFAULT_KEYSPACE)
@@ -52,7 +53,7 @@ def is_prepend_reversed():
def setup_connection(keyspace_name):
connection.setup(['127.0.0.1'],
connection.setup([CASSANDRA_IP],
consistency=ConsistencyLevel.ONE,
protocol_version=PROTOCOL_VERSION,
default_keyspace=keyspace_name)

9
tests/integration/cqlengine/columns/test_container_columns.py
View File

@@ -20,11 +20,13 @@ import sys
import traceback
from uuid import uuid4
from cassandra import WriteTimeout
from cassandra import WriteTimeout, OperationTimedOut
import cassandra.cqlengine.columns as columns
from cassandra.cqlengine.functions import get_total_seconds
from cassandra.cqlengine.models import Model, ValidationError
from cassandra.cqlengine.management import sync_table, drop_table
from tests.integration import CASSANDRA_IP
from tests.integration.cqlengine import is_prepend_reversed
from tests.integration.cqlengine.base import BaseCassEngTestCase
from tests.integration import greaterthancass20, CASSANDRA_VERSION
@@ -134,8 +136,11 @@ class TestSetColumn(BaseCassEngTestCase):
break
except WriteTimeout:
ex_type, ex, tb = sys.exc_info()
log.warn("{0}: {1} Backtrace: {2}".format(ex_type.__name__, ex, traceback.extract_tb(tb)))
log.warning("{0}: {1} Backtrace: {2}".format(ex_type.__name__, ex, traceback.extract_tb(tb)))
del tb
except OperationTimedOut:
#This will happen if the host is remote
self.assertFalse(CASSANDRA_IP.startswith("127.0.0."))
self.assertRaises(ValidationError, TestSetModel.create, **{'text_set': set(str(uuid4()) for i in range(65536))})
def test_partial_updates(self):

4
tests/integration/cqlengine/connections/test_connection.py
View File

@@ -24,7 +24,7 @@ from cassandra.cqlengine.management import sync_table
from cassandra.cluster import Cluster
from cassandra.query import dict_factory
from tests.integration import PROTOCOL_VERSION, execute_with_long_wait_retry
from tests.integration import PROTOCOL_VERSION, execute_with_long_wait_retry, local
from tests.integration.cqlengine.base import BaseCassEngTestCase
from tests.integration.cqlengine import DEFAULT_KEYSPACE, setup_connection
from cassandra.cqlengine import models
@@ -95,10 +95,12 @@ class ConnectionTest(BaseCassEngTestCase):
self.assertEqual(1, TestConnectModel.objects.count())
self.assertEqual(TestConnectModel.objects.first(), TCM2)
@local
def test_connection_setup_with_setup(self):
connection.setup(hosts=None, default_keyspace=None)
self.assertIsNotNone(connection.get_connection("default").cluster.metadata.get_host("127.0.0.1"))
@local
def test_connection_setup_with_default(self):
connection.default()
self.assertIsNotNone(connection.get_connection("default").cluster.metadata.get_host("127.0.0.1"))

28
tests/integration/cqlengine/test_connections.py
View File

@@ -23,6 +23,7 @@ from cassandra.cqlengine.query import ContextQuery, BatchQuery, ModelQuerySet
from tests.integration.cqlengine import setup_connection, DEFAULT_KEYSPACE
from tests.integration.cqlengine.base import BaseCassEngTestCase
from tests.integration.cqlengine.query import test_queryset
from tests.integration import local, CASSANDRA_IP
class TestModel(Model):
@@ -44,7 +45,6 @@ class AnotherTestModel(Model):
count = columns.Integer()
text = columns.Text()
class ContextQueryConnectionTests(BaseCassEngTestCase):
@classmethod
@@ -53,8 +53,8 @@ class ContextQueryConnectionTests(BaseCassEngTestCase):
create_keyspace_simple('ks1', 1)
conn.unregister_connection('default')
conn.register_connection('fake_cluster', ['127.0.0.100'], lazy_connect=True, retry_connect=True, default=True)
conn.register_connection('cluster', ['127.0.0.1'])
conn.register_connection('fake_cluster', ['1.2.3.4'], lazy_connect=True, retry_connect=True, default=True)
conn.register_connection('cluster', [CASSANDRA_IP])
with ContextQuery(TestModel, connection='cluster') as tm:
sync_table(tm)
@@ -141,7 +141,7 @@ class ManagementConnectionTests(BaseCassEngTestCase):
super(ManagementConnectionTests, cls).setUpClass()
conn.unregister_connection('default')
conn.register_connection('fake_cluster', ['127.0.0.100'], lazy_connect=True, retry_connect=True, default=True)
conn.register_connection('cluster', ['127.0.0.1'])
conn.register_connection('cluster', [CASSANDRA_IP])
@classmethod
def tearDownClass(cls):
@@ -227,11 +227,11 @@ class ManagementConnectionTests(BaseCassEngTestCase):
@test_category object_mapper
"""
cluster = Cluster(['127.0.0.1'])
cluster = Cluster([CASSANDRA_IP])
session = cluster.connect()
connection_name = 'from_session'
conn.register_connection(connection_name, session=session)
self.assertIsNotNone(conn.get_connection(connection_name).cluster.metadata.get_host("127.0.0.1"))
self.assertIsNotNone(conn.get_connection(connection_name).cluster.metadata.get_host(CASSANDRA_IP))
self.addCleanup(conn.unregister_connection, connection_name)
cluster.shutdown()
@@ -245,8 +245,8 @@ class ManagementConnectionTests(BaseCassEngTestCase):
@test_category object_mapper
"""
connection_name = 'from_hosts'
conn.register_connection(connection_name, hosts=['127.0.0.1'])
self.assertIsNotNone(conn.get_connection(connection_name).cluster.metadata.get_host("127.0.0.1"))
conn.register_connection(connection_name, hosts=[CASSANDRA_IP])
self.assertIsNotNone(conn.get_connection(connection_name).cluster.metadata.get_host(CASSANDRA_IP))
self.addCleanup(conn.unregister_connection, connection_name)
def test_connection_param_validation(self):
@@ -258,7 +258,7 @@ class ManagementConnectionTests(BaseCassEngTestCase):
@test_category object_mapper
"""
cluster = Cluster(['127.0.0.1'])
cluster = Cluster([CASSANDRA_IP])
session = cluster.connect()
with self.assertRaises(CQLEngineException):
conn.register_connection("bad_coonection1", session=session, consistency="not_null")
@@ -275,6 +275,8 @@ class ManagementConnectionTests(BaseCassEngTestCase):
cluster.shutdown()
cluster.shutdown()
class BatchQueryConnectionTests(BaseCassEngTestCase):
conns = ['cluster']
@@ -289,7 +291,7 @@ class BatchQueryConnectionTests(BaseCassEngTestCase):
conn.unregister_connection('default')
conn.register_connection('fake_cluster', ['127.0.0.100'], lazy_connect=True, retry_connect=True, default=True)
conn.register_connection('cluster', ['127.0.0.1'])
conn.register_connection('cluster', [CASSANDRA_IP])
@classmethod
def tearDownClass(cls):
@@ -415,7 +417,6 @@ class BatchQueryConnectionTests(BaseCassEngTestCase):
with BatchQuery(connection='cluster') as b:
obj1.batch(b).using(connection='test').save()
class UsingDescriptorTests(BaseCassEngTestCase):
conns = ['cluster']
@@ -427,7 +428,7 @@ class UsingDescriptorTests(BaseCassEngTestCase):
conn.unregister_connection('default')
conn.register_connection('fake_cluster', ['127.0.0.100'], lazy_connect=True, retry_connect=True, default=True)
conn.register_connection('cluster', ['127.0.0.1'])
conn.register_connection('cluster', [CASSANDRA_IP])
@classmethod
def tearDownClass(cls):
@@ -527,13 +528,12 @@ class ModelQuerySetNew(ModelQuerySet):
super(ModelQuerySetNew, self).__init__(*args, **kwargs)
self._connection = "cluster"
class BaseConnectionTestNoDefault(object):
conns = ['cluster']
@classmethod
def setUpClass(cls):
conn.register_connection('cluster', ['127.0.0.1'])
conn.register_connection('cluster', [CASSANDRA_IP])
test_queryset.TestModel.__queryset__ = ModelQuerySetNew
test_queryset.IndexedTestModel.__queryset__ = ModelQuerySetNew
test_queryset.IndexedCollectionsTestModel.__queryset__ = ModelQuerySetNew

32
tests/integration/standard/test_authentication.py
View File

@@ -18,7 +18,8 @@ import time
from cassandra.cluster import Cluster, NoHostAvailable
from cassandra.auth import PlainTextAuthProvider, SASLClient, SaslAuthProvider
from tests.integration import use_singledc, get_cluster, remove_cluster, PROTOCOL_VERSION
from tests.integration import use_singledc, get_cluster, remove_cluster, PROTOCOL_VERSION, CASSANDRA_IP, \
set_default_cass_ip
from tests.integration.util import assert_quiescent_pool_state
try:
@@ -29,18 +30,25 @@ except ImportError:
log = logging.getLogger(__name__)
#This can be tested for remote hosts, but the cluster has to be configured accordingly
#@local
def setup_module():
use_singledc(start=False)
ccm_cluster = get_cluster()
ccm_cluster.stop()
config_options = {'authenticator': 'PasswordAuthenticator',
'authorizer': 'CassandraAuthorizer'}
ccm_cluster.set_configuration_options(config_options)
log.debug("Starting ccm test cluster with %s", config_options)
ccm_cluster.start(wait_for_binary_proto=True, wait_other_notice=True)
# there seems to be some race, with some versions of C* taking longer to
# get the auth (and default user) setup. Sleep here to give it a chance
time.sleep(10)
if CASSANDRA_IP.startswith("127.0.0."):
use_singledc(start=False)
ccm_cluster = get_cluster()
ccm_cluster.stop()
config_options = {'authenticator': 'PasswordAuthenticator',
'authorizer': 'CassandraAuthorizer'}
ccm_cluster.set_configuration_options(config_options)
log.debug("Starting ccm test cluster with %s", config_options)
ccm_cluster.start(wait_for_binary_proto=True, wait_other_notice=True)
# there seems to be some race, with some versions of C* taking longer to
# get the auth (and default user) setup. Sleep here to give it a chance
time.sleep(10)
else:
set_default_cass_ip()
def teardown_module():

4
tests/integration/standard/test_client_warnings.py
View File

@@ -21,7 +21,7 @@ except ImportError:
from cassandra.query import BatchStatement
from cassandra.cluster import Cluster
from tests.integration import use_singledc, PROTOCOL_VERSION
from tests.integration import use_singledc, PROTOCOL_VERSION, local
def setup_module():
@@ -93,6 +93,7 @@ class ClientWarningTests(unittest.TestCase):
self.assertRegexpMatches(future.warnings[0], 'Batch.*exceeding.*')
self.assertIsNotNone(future.get_query_trace())
@local
def test_warning_with_custom_payload(self):
"""
Test to validate client warning with custom payload
@@ -111,6 +112,7 @@ class ClientWarningTests(unittest.TestCase):
self.assertRegexpMatches(future.warnings[0], 'Batch.*exceeding.*')
self.assertDictEqual(future.custom_payload, payload)
@local
def test_warning_with_trace_and_custom_payload(self):
"""
Test to validate client warning with tracing and client warning

40
tests/integration/standard/test_cluster.py
View File

@@ -32,8 +32,9 @@ from cassandra.policies import (RoundRobinPolicy, ExponentialReconnectionPolicy,
WhiteListRoundRobinPolicy, AddressTranslator)
from cassandra.query import SimpleStatement, TraceUnavailable, tuple_factory
from tests.integration import use_singledc, PROTOCOL_VERSION, get_server_versions, CASSANDRA_VERSION, execute_until_pass, execute_with_long_wait_retry, get_node,\
MockLoggingHandler, get_unsupported_lower_protocol, get_unsupported_upper_protocol, protocolv5
from tests.integration import use_singledc, PROTOCOL_VERSION, get_server_versions, CASSANDRA_VERSION, DSE_VERSION, execute_until_pass, execute_with_long_wait_retry, get_node,\
MockLoggingHandler, get_unsupported_lower_protocol, get_unsupported_upper_protocol, protocolv5, local, CASSANDRA_IP
from tests.integration.util import assert_quiescent_pool_state
import sys
@@ -56,7 +57,7 @@ class IgnoredHostPolicy(RoundRobinPolicy):
class ClusterTests(unittest.TestCase):
@local
def test_ignored_host_up(self):
"""
Test to ensure that is_up is not set by default on ignored hosts
@@ -77,6 +78,7 @@ class ClusterTests(unittest.TestCase):
self.assertIsNone(host.is_up)
cluster.shutdown()
@local
def test_host_resolution(self):
"""
Test to insure A records are resolved appropriately.
@@ -90,6 +92,7 @@ class ClusterTests(unittest.TestCase):
cluster = Cluster(contact_points=["localhost"], protocol_version=PROTOCOL_VERSION, connect_timeout=1)
self.assertTrue('127.0.0.1' in cluster.contact_points_resolved)
@local
def test_host_duplication(self):
"""
Ensure that duplicate hosts in the contact points are surfaced in the cluster metadata
@@ -109,6 +112,7 @@ class ClusterTests(unittest.TestCase):
self.assertEqual(len(cluster.metadata.all_hosts()), 3)
cluster.shutdown()
@local
def test_raise_error_on_control_connection_timeout(self):
"""
Test for initial control connection timeout
@@ -436,9 +440,9 @@ class ClusterTests(unittest.TestCase):
self.assertEqual(original_type_meta.as_cql_query(), current_type_meta.as_cql_query())
cluster.shutdown()
@local
def test_refresh_schema_no_wait(self):
contact_points = ['127.0.0.1']
contact_points = [CASSANDRA_IP]
cluster = Cluster(protocol_version=PROTOCOL_VERSION, max_schema_agreement_wait=10,
contact_points=contact_points, load_balancing_policy=WhiteListRoundRobinPolicy(contact_points))
session = cluster.connect()
@@ -447,6 +451,7 @@ class ClusterTests(unittest.TestCase):
new_schema_ver = uuid4()
session.execute("UPDATE system.local SET schema_version=%s WHERE key='local'", (new_schema_ver,))
try:
agreement_timeout = 1
@@ -689,6 +694,7 @@ class ClusterTests(unittest.TestCase):
cluster.shutdown()
@local
def test_profile_load_balancing(self):
"""
Tests that profile load balancing policies are honored.
@@ -700,7 +706,7 @@ class ClusterTests(unittest.TestCase):
@test_category config_profiles
"""
query = "select release_version from system.local"
node1 = ExecutionProfile(load_balancing_policy=WhiteListRoundRobinPolicy(['127.0.0.1']))
node1 = ExecutionProfile(load_balancing_policy=WhiteListRoundRobinPolicy([CASSANDRA_IP]))
with Cluster(execution_profiles={'node1': node1}) as cluster:
session = cluster.connect(wait_for_all_pools=True)
@@ -713,7 +719,7 @@ class ClusterTests(unittest.TestCase):
self.assertEqual(queried_hosts, expected_hosts)
# by name we should only hit the one
expected_hosts = set(h for h in cluster.metadata.all_hosts() if h.address == '127.0.0.1')
expected_hosts = set(h for h in cluster.metadata.all_hosts() if h.address == CASSANDRA_IP)
queried_hosts = set()
for _ in cluster.metadata.all_hosts():
rs = session.execute(query, execution_profile='node1')
@@ -837,6 +843,7 @@ class ClusterTests(unittest.TestCase):
with self.assertRaises(ValueError):
session.execute(query, execution_profile='rr3')
@local
def test_profile_pool_management(self):
"""
Tests that changes to execution profiles correctly impact our cluster's pooling
@@ -863,6 +870,7 @@ class ClusterTests(unittest.TestCase):
pools = session.get_pool_state()
self.assertEqual(set(h.address for h in pools), set(('127.0.0.1', '127.0.0.2', '127.0.0.3')))
@local
def test_add_profile_timeout(self):
"""
Tests that EP Timeouts are honored.
@@ -889,10 +897,12 @@ class ClusterTests(unittest.TestCase):
try:
self.assertRaises(cassandra.OperationTimedOut, cluster.add_execution_profile, 'node2',
node2, pool_wait_timeout=sys.float_info.min)
break
except Exception:
end = time.time()
self.assertAlmostEqual(start, end, 1)
break
else:
raise Exception("add_execution_profile didn't timeout after {0} retries".format(max_retry_count))
class LocalHostAdressTranslator(AddressTranslator):
@@ -904,7 +914,7 @@ class LocalHostAdressTranslator(AddressTranslator):
new_addr = self.addr_map.get(addr)
return new_addr
@local
class TestAddressTranslation(unittest.TestCase):
def test_address_translator_basic(self):
@@ -948,11 +958,11 @@ class TestAddressTranslation(unittest.TestCase):
self.assertEqual(adder_map.get(str(host)), host.broadcast_address)
c.shutdown()
@local
class ContextManagementTest(unittest.TestCase):
load_balancing_policy = WhiteListRoundRobinPolicy(['127.0.0.1'])
cluster_kwargs = {'load_balancing_policy': load_balancing_policy,
load_balancing_policy = WhiteListRoundRobinPolicy([CASSANDRA_IP])
cluster_kwargs = {'execution_profiles': {EXEC_PROFILE_DEFAULT: ExecutionProfile(load_balancing_policy=
load_balancing_policy)},
'schema_metadata_enabled': False,
'token_metadata_enabled': False}
@@ -1068,7 +1078,7 @@ class HostStateTest(unittest.TestCase):
time.sleep(.01)
self.assertTrue(was_marked_down)
@local
class DontPrepareOnIgnoredHostsTest(unittest.TestCase):
ignored_addresses = ['127.0.0.3']
@@ -1106,7 +1116,7 @@ class DontPrepareOnIgnoredHostsTest(unittest.TestCase):
self.assertEqual(call(unignored_address), c)
cluster.shutdown()
@local
class DuplicateRpcTest(unittest.TestCase):
load_balancing_policy = WhiteListRoundRobinPolicy(['127.0.0.1'])

9
tests/integration/standard/test_connection.py
View File

@@ -32,7 +32,7 @@ from cassandra.policies import WhiteListRoundRobinPolicy, HostStateListener
from cassandra.pool import HostConnectionPool
from tests import is_monkey_patched
from tests.integration import use_singledc, PROTOCOL_VERSION, get_node
from tests.integration import use_singledc, PROTOCOL_VERSION, get_node, CASSANDRA_IP, local
try:
from cassandra.io.libevreactor import LibevConnection
@@ -49,7 +49,8 @@ class ConnectionTimeoutTest(unittest.TestCase):
def setUp(self):
self.defaultInFlight = Connection.max_in_flight
Connection.max_in_flight = 2
self.cluster = Cluster(protocol_version=PROTOCOL_VERSION, load_balancing_policy=WhiteListRoundRobinPolicy(['127.0.0.1']))
self.cluster = Cluster(protocol_version=PROTOCOL_VERSION, load_balancing_policy=
WhiteListRoundRobinPolicy([CASSANDRA_IP]))
self.session = self.cluster.connect()
def tearDown(self):
@@ -104,6 +105,7 @@ class HeartbeatTest(unittest.TestCase):
def tearDown(self):
self.cluster.shutdown()
@local
def test_heart_beat_timeout(self):
# Setup a host listener to ensure the nodes don't go down
test_listener = TestHostListener()
@@ -190,7 +192,8 @@ class ConnectionTests(object):
e = None
for i in range(5):
try:
conn = self.klass.factory(host='127.0.0.1', timeout=timeout, protocol_version=PROTOCOL_VERSION)
contact_point = CASSANDRA_IP
conn = self.klass.factory(host=contact_point, timeout=timeout, protocol_version=PROTOCOL_VERSION)
break
except (OperationTimedOut, NoHostAvailable, ConnectionShutdown) as e:
continue

5
tests/integration/standard/test_custom_payload.py
View File

@@ -23,11 +23,14 @@ import six
from cassandra.query import (SimpleStatement, BatchStatement, BatchType)
from cassandra.cluster import Cluster
from tests.integration import use_singledc, PROTOCOL_VERSION
from tests.integration import use_singledc, PROTOCOL_VERSION, local
def setup_module():
use_singledc()
#These test rely on the custom payload being returned but by default C*
#ignores all the payloads.
@local
class CustomPayloadTests(unittest.TestCase):
def setUp(self):

1
tests/integration/standard/test_custom_protocol_handler.py
View File

@@ -69,6 +69,7 @@ class CustomProtocolHandlerTest(unittest.TestCase):
cluster = Cluster(protocol_version=PROTOCOL_VERSION)
session = cluster.connect(keyspace="custserdes")
session.row_factory = tuple_factory
result = session.execute("SELECT schema_version FROM system.local")
uuid_type = result[0][0]
self.assertEqual(type(uuid_type), uuid.UUID)

14
tests/integration/standard/test_metadata.py
View File

@@ -37,7 +37,7 @@ from tests.integration import (get_cluster, use_singledc, PROTOCOL_VERSION, get_
BasicSegregatedKeyspaceUnitTestCase, BasicSharedKeyspaceUnitTestCase,
BasicExistingKeyspaceUnitTestCase, drop_keyspace_shutdown_cluster, CASSANDRA_VERSION,
BasicExistingSegregatedKeyspaceUnitTestCase, dseonly, DSE_VERSION,
get_supported_protocol_versions, greaterthanorequalcass30, lessthancass30)
get_supported_protocol_versions, greaterthanorequalcass30, lessthancass30, local)
from tests.integration import greaterthancass21
@@ -48,13 +48,14 @@ def setup_module():
class HostMetatDataTests(BasicExistingKeyspaceUnitTestCase):
@local
def test_broadcast_listen_address(self):
"""
Check to ensure that the broadcast and listen adresss is populated correctly
@since 3.3
@jira_ticket PYTHON-332
@expected_result They are populated for C*> 2.0.16, 2.1.6, 2.2.0
@expected_result They are populated for C*> 2.1.6, 2.2.0
@test_category metadata
"""
@@ -81,7 +82,7 @@ class HostMetatDataTests(BasicExistingKeyspaceUnitTestCase):
for host in self.cluster.metadata.all_hosts():
self.assertTrue(host.release_version.startswith(CASSANDRA_VERSION))
@local
class MetaDataRemovalTest(unittest.TestCase):
def setUp(self):
@@ -1090,6 +1091,7 @@ CREATE TABLE export_udts.users (
ksname = 'AnInterestingKeyspace'
cfname = 'AnInterestingTable'
session.execute("DROP KEYSPACE IF EXISTS {0}".format(ksname))
session.execute("""
CREATE KEYSPACE "%s"
WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}
@@ -1146,6 +1148,7 @@ CREATE TABLE export_udts.users (
self.assertRaises(AlreadyExists, session.execute, ddl % (ksname, cfname))
cluster.shutdown()
@local
def test_replicas(self):
"""
Ensure cluster.metadata.get_replicas return correctly when not attached to keyspace
@@ -1184,6 +1187,7 @@ CREATE TABLE export_udts.users (
self.assertEqual(set(get_replicas('test1rf', token)), set([owners[(i + 1) % 3]]))
cluster.shutdown()
@local
def test_legacy_tables(self):
if CASS_SERVER_VERSION < (2, 1, 0):
@@ -1450,7 +1454,7 @@ class TokenMetadataTest(unittest.TestCase):
"""
Test of TokenMap creation and other behavior.
"""
@local
def test_token(self):
expected_node_count = len(get_cluster().nodes)
@@ -2117,8 +2121,6 @@ class BadMetaTest(unittest.TestCase):
cls.parser_class = get_schema_parser(connection, str(CASS_SERVER_VERSION[0]), timeout=20).__class__
cls.cluster.control_connection.reconnect = Mock()
@classmethod
def teardown_class(cls):
drop_keyspace_shutdown_cluster(cls.keyspace_name, cls.session, cls.cluster)

6
tests/integration/standard/test_metrics.py
View File

@@ -28,12 +28,12 @@ from cassandra.protocol import SyntaxException
from cassandra.cluster import Cluster, NoHostAvailable
from tests.integration import get_cluster, get_node, use_singledc, PROTOCOL_VERSION, execute_until_pass
from greplin import scales
from tests.integration import BasicSharedKeyspaceUnitTestCaseWTable, BasicExistingKeyspaceUnitTestCase
from tests.integration import BasicSharedKeyspaceUnitTestCaseWTable, BasicExistingKeyspaceUnitTestCase, local
def setup_module():
use_singledc()
@local
class MetricsTests(unittest.TestCase):
def setUp(self):
@@ -178,7 +178,7 @@ class MetricsTests(unittest.TestCase):
class MetricsNamespaceTest(BasicSharedKeyspaceUnitTestCaseWTable):
@local
def test_metrics_per_cluster(self):
"""
Test to validate that metrics can be scopped to invdividual clusters

4
tests/integration/standard/test_policies.py
View File

@@ -18,11 +18,13 @@ try:
import unittest2 as unittest
except ImportError:
import unittest # noqa
from cassandra import OperationTimedOut
from cassandra.cluster import ExecutionProfile
from cassandra.query import SimpleStatement
from cassandra.policies import ConstantSpeculativeExecutionPolicy, RoundRobinPolicy
from tests.integration import BasicSharedKeyspaceUnitTestCase, greaterthancass21
from tests import notwindows
def setup_module():
@@ -52,6 +54,8 @@ class SpecExecTest(BasicSharedKeyspaceUnitTestCase):
self.cluster.add_execution_profile("spec_ep_rr", spec_ep_rr)
self.cluster.add_execution_profile("spec_ep_rr_lim", spec_ep_rr_lim)
#This doesn't work well with Windows clock granularity
@notwindows
@greaterthancass21
def test_speculative_execution(self):
"""

6
tests/integration/standard/test_prepared_statements.py
View File

@@ -48,7 +48,11 @@ class PreparedStatementTests(unittest.TestCase):
"""
Test basic PreparedStatement usage
"""
self.session.execute(
"""
DROP KEYSPACE IF EXISTS preparedtests
"""
)
self.session.execute(
"""
CREATE KEYSPACE preparedtests

10
tests/integration/standard/test_query.py
View File

@@ -25,7 +25,9 @@ from cassandra.query import (PreparedStatement, BoundStatement, SimpleStatement,
BatchStatement, BatchType, dict_factory, TraceUnavailable)
from cassandra.cluster import Cluster, NoHostAvailable
from cassandra.policies import HostDistance, RoundRobinPolicy
from tests.integration import use_singledc, PROTOCOL_VERSION, BasicSharedKeyspaceUnitTestCase, get_server_versions, greaterthanprotocolv3, MockLoggingHandler, get_supported_protocol_versions
from tests.integration import use_singledc, PROTOCOL_VERSION, BasicSharedKeyspaceUnitTestCase, get_server_versions, \
greaterthanprotocolv3, MockLoggingHandler, get_supported_protocol_versions, local
from tests import notwindows
import time
import re
@@ -115,6 +117,7 @@ class QueryTests(BasicSharedKeyspaceUnitTestCase):
for event in trace.events:
str(event)
@local
@greaterthanprotocolv3
def test_client_ip_in_trace(self):
"""
@@ -179,6 +182,7 @@ class QueryTests(BasicSharedKeyspaceUnitTestCase):
self.assertIsNotNone(response_future.get_query_trace(max_wait=2.0, query_cl=ConsistencyLevel.ANY).trace_id)
self.assertIsNotNone(response_future.get_query_trace(max_wait=2.0, query_cl=ConsistencyLevel.QUORUM).trace_id)
@notwindows
def test_incomplete_query_trace(self):
"""
Tests to ensure that partial tracing works.
@@ -430,9 +434,9 @@ class PreparedStatementMetdataTest(unittest.TestCase):
future = session.execute_async(select_statement)
results = future.result()
if base_line is None:
base_line = results[0].__dict__.keys()
base_line = results[0]._asdict().keys()
else:
self.assertEqual(base_line, results[0].__dict__.keys())
self.assertEqual(base_line, results[0]._asdict().keys())
cluster.shutdown()

9
tests/unit/test_concurrent.py
View File

@@ -24,6 +24,7 @@ import time
import threading
from six.moves.queue import PriorityQueue
import sys
import platform
from cassandra.cluster import Cluster, Session
from cassandra.concurrent import execute_concurrent, execute_concurrent_with_args
@@ -115,7 +116,6 @@ class TimedCallableInvoker(threading.Thread):
self._stopper.wait(.001)
return
class ConcurrencyTest((unittest.TestCase)):
def test_results_ordering_forward(self):
@@ -231,7 +231,12 @@ class ConcurrencyTest((unittest.TestCase)):
for success, result in results:
self.assertTrue(success)
current_time_added = list(result)[0]
self.assertLess(last_time_added, current_time_added)
#Windows clock granularity makes this equal most of the times
if "Windows" in platform.system():
self.assertLessEqual(last_time_added, current_time_added)
else:
self.assertLess(last_time_added, current_time_added)
last_time_added = current_time_added
def test_recursion_limited(self):