| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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. |
| |
| import unittest |
| |
| from cassandra.encoder import Encoder |
| from cassandra.protocol import ColumnMetadata |
| from cassandra.query import (bind_params, ValueSequence, PreparedStatement, |
| BoundStatement, UNSET_VALUE) |
| from cassandra.cqltypes import Int32Type |
| from cassandra.util import OrderedDict |
| |
| |
| class ParamBindingTest(unittest.TestCase): |
| |
| def test_bind_sequence(self): |
| result = bind_params("%s %s %s", (1, "a", 2.0), Encoder()) |
| self.assertEqual(result, "1 'a' 2.0") |
| |
| def test_bind_map(self): |
| result = bind_params("%(a)s %(b)s %(c)s", dict(a=1, b="a", c=2.0), Encoder()) |
| self.assertEqual(result, "1 'a' 2.0") |
| |
| def test_sequence_param(self): |
| result = bind_params("%s", (ValueSequence((1, "a", 2.0)),), Encoder()) |
| self.assertEqual(result, "(1, 'a', 2.0)") |
| |
| def test_generator_param(self): |
| result = bind_params("%s", ((i for i in range(3)),), Encoder()) |
| self.assertEqual(result, "[0, 1, 2]") |
| |
| def test_none_param(self): |
| result = bind_params("%s", (None,), Encoder()) |
| self.assertEqual(result, "NULL") |
| |
| def test_list_collection(self): |
| result = bind_params("%s", (['a', 'b', 'c'],), Encoder()) |
| self.assertEqual(result, "['a', 'b', 'c']") |
| |
| def test_set_collection(self): |
| result = bind_params("%s", (set(['a', 'b']),), Encoder()) |
| self.assertIn(result, ("{'a', 'b'}", "{'b', 'a'}")) |
| |
| def test_map_collection(self): |
| vals = OrderedDict() |
| vals['a'] = 'a' |
| vals['b'] = 'b' |
| vals['c'] = 'c' |
| result = bind_params("%s", (vals,), Encoder()) |
| self.assertEqual(result, "{'a': 'a', 'b': 'b', 'c': 'c'}") |
| |
| def test_quote_escaping(self): |
| result = bind_params("%s", ("""'ef''ef"ef""ef'""",), Encoder()) |
| self.assertEqual(result, """'''ef''''ef"ef""ef'''""") |
| |
| def test_float_precision(self): |
| f = 3.4028234663852886e+38 |
| self.assertEqual(float(bind_params("%s", (f,), Encoder())), f) |
| |
| |
| class BoundStatementTestV1(unittest.TestCase): |
| |
| protocol_version = 1 |
| |
| @classmethod |
| def setUpClass(cls): |
| column_metadata = [ColumnMetadata('keyspace', 'cf', 'rk0', Int32Type), |
| ColumnMetadata('keyspace', 'cf', 'rk1', Int32Type), |
| ColumnMetadata('keyspace', 'cf', 'ck0', Int32Type), |
| ColumnMetadata('keyspace', 'cf', 'v0', Int32Type)] |
| cls.prepared = PreparedStatement(column_metadata=column_metadata, |
| query_id=None, |
| routing_key_indexes=[1, 0], |
| query=None, |
| keyspace='keyspace', |
| protocol_version=cls.protocol_version, result_metadata=None, |
| result_metadata_id=None) |
| cls.bound = BoundStatement(prepared_statement=cls.prepared) |
| |
| def test_invalid_argument_type(self): |
| values = (0, 0, 0, 'string not int') |
| try: |
| self.bound.bind(values) |
| except TypeError as e: |
| self.assertIn('v0', str(e)) |
| self.assertIn('Int32Type', str(e)) |
| self.assertIn('str', str(e)) |
| else: |
| self.fail('Passed invalid type but exception was not thrown') |
| |
| values = (['1', '2'], 0, 0, 0) |
| |
| try: |
| self.bound.bind(values) |
| except TypeError as e: |
| self.assertIn('rk0', str(e)) |
| self.assertIn('Int32Type', str(e)) |
| self.assertIn('list', str(e)) |
| else: |
| self.fail('Passed invalid type but exception was not thrown') |
| |
| def test_inherit_fetch_size(self): |
| keyspace = 'keyspace1' |
| column_family = 'cf1' |
| |
| column_metadata = [ |
| ColumnMetadata(keyspace, column_family, 'foo1', Int32Type), |
| ColumnMetadata(keyspace, column_family, 'foo2', Int32Type) |
| ] |
| |
| prepared_statement = PreparedStatement(column_metadata=column_metadata, |
| query_id=None, |
| routing_key_indexes=[], |
| query=None, |
| keyspace=keyspace, |
| protocol_version=self.protocol_version, |
| result_metadata=None, |
| result_metadata_id=None) |
| prepared_statement.fetch_size = 1234 |
| bound_statement = BoundStatement(prepared_statement=prepared_statement) |
| self.assertEqual(1234, bound_statement.fetch_size) |
| |
| def test_too_few_parameters_for_routing_key(self): |
| self.assertRaises(ValueError, self.prepared.bind, (1,)) |
| |
| bound = self.prepared.bind((1, 2)) |
| self.assertEqual(bound.keyspace, 'keyspace') |
| |
| def test_dict_missing_routing_key(self): |
| self.assertRaises(KeyError, self.bound.bind, {'rk0': 0, 'ck0': 0, 'v0': 0}) |
| self.assertRaises(KeyError, self.bound.bind, {'rk1': 0, 'ck0': 0, 'v0': 0}) |
| |
| def test_missing_value(self): |
| self.assertRaises(KeyError, self.bound.bind, {'rk0': 0, 'rk1': 0, 'ck0': 0}) |
| |
| def test_extra_value(self): |
| self.bound.bind({'rk0': 0, 'rk1': 0, 'ck0': 0, 'v0': 0, 'should_not_be_here': 123}) # okay to have extra keys in dict |
| self.assertEqual(self.bound.values, [b'\x00' * 4] * 4) # four encoded zeros |
| self.assertRaises(ValueError, self.bound.bind, (0, 0, 0, 0, 123)) |
| |
| def test_values_none(self): |
| # should have values |
| self.assertRaises(ValueError, self.bound.bind, None) |
| |
| # prepared statement with no values |
| prepared_statement = PreparedStatement(column_metadata=[], |
| query_id=None, |
| routing_key_indexes=[], |
| query=None, |
| keyspace='whatever', |
| protocol_version=self.protocol_version, |
| result_metadata=None, |
| result_metadata_id=None) |
| bound = prepared_statement.bind(None) |
| self.assertListEqual(bound.values, []) |
| |
| def test_bind_none(self): |
| self.bound.bind({'rk0': 0, 'rk1': 0, 'ck0': 0, 'v0': None}) |
| self.assertEqual(self.bound.values[-1], None) |
| |
| old_values = self.bound.values |
| self.bound.bind((0, 0, 0, None)) |
| self.assertIsNot(self.bound.values, old_values) |
| self.assertEqual(self.bound.values[-1], None) |
| |
| def test_unset_value(self): |
| self.assertRaises(ValueError, self.bound.bind, {'rk0': 0, 'rk1': 0, 'ck0': 0, 'v0': UNSET_VALUE}) |
| self.assertRaises(ValueError, self.bound.bind, (0, 0, 0, UNSET_VALUE)) |
| |
| |
| class BoundStatementTestV2(BoundStatementTestV1): |
| protocol_version = 2 |
| |
| |
| class BoundStatementTestV3(BoundStatementTestV1): |
| protocol_version = 3 |
| |
| |
| class BoundStatementTestV4(BoundStatementTestV1): |
| protocol_version = 4 |
| |
| def test_dict_missing_routing_key(self): |
| # in v4 it implicitly binds UNSET_VALUE for missing items, |
| # UNSET_VALUE is ValueError for routing keys |
| self.assertRaises(ValueError, self.bound.bind, {'rk0': 0, 'ck0': 0, 'v0': 0}) |
| self.assertRaises(ValueError, self.bound.bind, {'rk1': 0, 'ck0': 0, 'v0': 0}) |
| |
| def test_missing_value(self): |
| # in v4 missing values are UNSET_VALUE |
| self.bound.bind({'rk0': 0, 'rk1': 0, 'ck0': 0}) |
| self.assertEqual(self.bound.values[-1], UNSET_VALUE) |
| |
| old_values = self.bound.values |
| self.bound.bind((0, 0, 0)) |
| self.assertIsNot(self.bound.values, old_values) |
| self.assertEqual(self.bound.values[-1], UNSET_VALUE) |
| |
| def test_unset_value(self): |
| self.bound.bind({'rk0': 0, 'rk1': 0, 'ck0': 0, 'v0': UNSET_VALUE}) |
| self.assertEqual(self.bound.values[-1], UNSET_VALUE) |
| |
| self.bound.bind((0, 0, 0, UNSET_VALUE)) |
| self.assertEqual(self.bound.values[-1], UNSET_VALUE) |
| |
| |
| class BoundStatementTestV5(BoundStatementTestV4): |
| protocol_version = 5 |
