1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
|
# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from tests.integration import get_server_versions, use_singledc, \
BasicSharedKeyspaceUnitTestCaseWFunctionTable, BasicSharedKeyspaceUnitTestCase, execute_until_pass, TestCluster
import unittest
from cassandra.cluster import ResultSet, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.query import tuple_factory, named_tuple_factory, dict_factory, ordered_dict_factory
from cassandra.util import OrderedDict
def setup_module():
use_singledc()
class NameTupleFactory(BasicSharedKeyspaceUnitTestCase):
def setUp(self):
self.common_setup(1, execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(row_factory=named_tuple_factory)})
ddl = '''
CREATE TABLE {0}.{1} (
k int PRIMARY KEY,
v1 text,
v2 text,
v3 text)'''.format(self.ks_name, self.function_table_name)
self.session.execute(ddl)
execute_until_pass(self.session, ddl)
def test_sanitizing(self):
"""
Test to ensure that same named results are surfaced in the NamedTupleFactory
Creates a table with a few different text fields. Inserts a few values in that table.
It then fetches the values and confirms that despite all be being selected as the same name
they are propagated in the result set differently.
@since 3.3
@jira_ticket PYTHON-467
@expected_result duplicate named results have unique row names.
@test_category queries
"""
for x in range(5):
insert1 = '''
INSERT INTO {0}.{1}
( k , v1, v2, v3 )
VALUES
( 1 , 'v1{2}', 'v2{2}','v3{2}' )
'''.format(self.keyspace_name, self.function_table_name, str(x))
self.session.execute(insert1)
query = "SELECT v1 AS duplicate, v2 AS duplicate, v3 AS duplicate from {0}.{1}".format(self.ks_name, self.function_table_name)
rs = self.session.execute(query)
row = rs[0]
self.assertTrue(hasattr(row, 'duplicate'))
self.assertTrue(hasattr(row, 'duplicate_'))
self.assertTrue(hasattr(row, 'duplicate__'))
class RowFactoryTests(BasicSharedKeyspaceUnitTestCaseWFunctionTable):
"""
Test different row_factories and access code
"""
@classmethod
def setUpClass(cls):
cls.common_setup(rf=1, create_class_table=True)
q = "INSERT INTO {0}.{1} (k, v) VALUES (%s, %s)".format(cls.ks_name, cls.ks_name)
cls.session.execute(q, (1, 1))
cls.session.execute(q, (2, 2))
cls.select = "SELECT * FROM {0}.{1}".format(cls.ks_name, cls.ks_name)
def _results_from_row_factory(self, row_factory):
cluster = TestCluster(
execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(row_factory=row_factory)}
)
with cluster:
return cluster.connect().execute(self.select)
def test_tuple_factory(self):
result = self._results_from_row_factory(tuple_factory)
self.assertIsInstance(result, ResultSet)
self.assertIsInstance(result[0], tuple)
for row in result:
self.assertEqual(row[0], row[1])
self.assertEqual(result[0][0], result[0][1])
self.assertEqual(result[0][0], 1)
self.assertEqual(result[1][0], result[1][1])
self.assertEqual(result[1][0], 2)
def test_named_tuple_factory(self):
result = self._results_from_row_factory(named_tuple_factory)
self.assertIsInstance(result, ResultSet)
result = list(result)
for row in result:
self.assertEqual(row.k, row.v)
self.assertEqual(result[0].k, result[0].v)
self.assertEqual(result[0].k, 1)
self.assertEqual(result[1].k, result[1].v)
self.assertEqual(result[1].k, 2)
def _test_dict_factory(self, row_factory, row_type):
result = self._results_from_row_factory(row_factory)
self.assertIsInstance(result, ResultSet)
self.assertIsInstance(result[0], row_type)
for row in result:
self.assertEqual(row['k'], row['v'])
self.assertEqual(result[0]['k'], result[0]['v'])
self.assertEqual(result[0]['k'], 1)
self.assertEqual(result[1]['k'], result[1]['v'])
self.assertEqual(result[1]['k'], 2)
def test_dict_factory(self):
self._test_dict_factory(dict_factory, dict)
def test_ordered_dict_factory(self):
self._test_dict_factory(ordered_dict_factory, OrderedDict)
def test_generator_row_factory(self):
"""
Test that ResultSet.one() works with a row_factory that contains a generator.
@since 3.16
@jira_ticket PYTHON-1026
@expected_result one() returns the first row
@test_category queries
"""
def generator_row_factory(column_names, rows):
return _gen_row_factory(rows)
def _gen_row_factory(rows):
for r in rows:
yield r
session = self.session
session.row_factory = generator_row_factory
session.execute('''
INSERT INTO {0}.{1}
( k , v )
VALUES
( 1 , 1 )
'''.format(self.keyspace_name, self.function_table_name))
result = session.execute(self.select)
self.assertIsInstance(result, ResultSet)
first_row = result.one()
self.assertEqual(first_row[0], first_row[1])
class NamedTupleFactoryAndNumericColNamesTests(unittest.TestCase):
"""
Test for PYTHON-122: Improve Error Handling/Reporting for named_tuple_factory and Numeric Column Names
"""
@classmethod
def setup_class(cls):
cls.cluster = TestCluster()
cls.session = cls.cluster.connect()
cls._cass_version, cls._cql_version = get_server_versions()
ddl = '''
CREATE TABLE test1rf.table_num_col ( key blob PRIMARY KEY, "626972746864617465" blob )'''
cls.session.execute(ddl)
@classmethod
def teardown_class(cls):
cls.session.execute("DROP TABLE test1rf.table_num_col")
cls.cluster.shutdown()
def test_no_exception_on_select(self):
"""
no exception on SELECT for numeric column name
"""
try:
self.session.execute('SELECT * FROM test1rf.table_num_col')
except ValueError as e:
self.fail("Unexpected ValueError exception: %s" % e.message)
def test_can_select_using_alias(self):
"""
can SELECT "<numeric col name>" AS aliases
"""
if self._cass_version < (2, 0, 0):
raise unittest.SkipTest("Alias in SELECT not supported before 2.0")
try:
self.session.execute('SELECT key, "626972746864617465" AS my_col from test1rf.table_num_col')
except ValueError as e:
self.fail("Unexpected ValueError exception: %s" % e.message)
def test_can_select_with_dict_factory(self):
"""
can SELECT numeric column using dict_factory
"""
with TestCluster(
execution_profiles={EXEC_PROFILE_DEFAULT: ExecutionProfile(row_factory=dict_factory)}
) as cluster:
try:
cluster.connect().execute('SELECT * FROM test1rf.table_num_col')
except ValueError as e:
self.fail("Unexpected ValueError exception: %s" % e.message)
|
