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

Added possible fix to test_connect_on_keyspace

Browse Source
This commit is contained in:
bjmb
2017-05-09 16:20:28 -04:00
parent b8f1b77383
commit dc8d840cb1
3 changed files with 12 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/__init__.py
View File

@@ -63,4 +63,4 @@ MONKEY_PATCH_LOOP = bool(os.getenv('MONKEY_PATCH_LOOP', False))
notwindows = unittest.skipUnless(not "Windows" in platform.system(), "This test is not adequate for windows")
notpypy = unittest.skipUnless(not platform.python_implementation() == 'PyPy', "This tests is not suitable for pypy")
notmonkeypatch = unittest.skipUnless(MONKEY_PATCH_LOOP, "Skpping this test because monkey patching is required")
notmonkeypatch = unittest.skipUnless(MONKEY_PATCH_LOOP, "Skipping this test because monkey patching is required")

10
tests/integration/__init__.py
View File

@@ -529,11 +529,17 @@ def setup_keyspace(ipformat=None, wait=True):
WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}'''
execute_with_long_wait_retry(session, ddl)
ddl = '''
ddl_3f = '''
CREATE TABLE test3rf.test (
k int PRIMARY KEY,
v int )'''
execute_with_long_wait_retry(session, ddl)
execute_with_long_wait_retry(session, ddl_3f)
ddl_1f = '''
CREATE TABLE test1rf.test (
k int PRIMARY KEY,
v int )'''
execute_with_long_wait_retry(session, ddl_1f)
except Exception:
traceback.print_exc()

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

@@ -282,15 +282,15 @@ class ClusterTests(unittest.TestCase):
session = cluster.connect()
result = session.execute(
"""
INSERT INTO test3rf.test (k, v) VALUES (8889, 8889)
INSERT INTO test1rf.test (k, v) VALUES (8889, 8889)
""")
self.assertFalse(result)
result = session.execute("SELECT * FROM test3rf.test")
result = session.execute("SELECT * FROM test1rf.test")
self.assertEqual([(8889, 8889)], result)
# test_connect_on_keyspace
session2 = cluster.connect('test3rf')
session2 = cluster.connect('test1rf')
result2 = session2.execute("SELECT * FROM test")
self.assertEqual(result, result2)
cluster.shutdown()