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

fixed a whole bunch of unicode references, fixed incorrect encoding of clauses

Browse Source
This commit is contained in:
Jon Haddad
2014-08-13 17:55:42 -07:00
parent e64a7de36d
commit 9486baa300
8 changed files with 44 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
cqlengine/functions.py
View File

@@ -1,9 +1,17 @@
from datetime import datetime
from uuid import uuid1
import sys
import six
from cqlengine.exceptions import ValidationError
# move to central spot
class QueryValue(object):
class UnicodeMixin(object):
if sys.version_info > (3, 0):
__str__ = lambda x: x.__unicode__()
else:
__str__ = lambda x: six.text_type(x).encode('utf-8')
class QueryValue(UnicodeMixin):
"""
Base class for query filter values. Subclasses of these classes can
be passed into .filter() keyword args

1
cqlengine/query.py
View File

@@ -20,6 +20,7 @@ class QueryException(CQLEngineException): pass
class DoesNotExist(QueryException): pass
class MultipleObjectsReturned(QueryException): pass
import six
class AbstractQueryableColumn(object):
"""

13
cqlengine/tests/operators/test_where_operators.py
View File

@@ -1,6 +1,7 @@
from unittest import TestCase
from cqlengine.operators import *
import six
class TestWhereOperators(TestCase):
@@ -20,11 +21,11 @@ class TestWhereOperators(TestCase):
def test_operator_rendering(self):
""" tests symbols are rendered properly """
self.assertEqual("=", unicode(EqualsOperator()))
self.assertEqual("IN", unicode(InOperator()))
self.assertEqual(">", unicode(GreaterThanOperator()))
self.assertEqual(">=", unicode(GreaterThanOrEqualOperator()))
self.assertEqual("<", unicode(LessThanOperator()))
self.assertEqual("<=", unicode(LessThanOrEqualOperator()))
self.assertEqual("=", six.text_type(EqualsOperator()))
self.assertEqual("IN", six.text_type(InOperator()))
self.assertEqual(">", six.text_type(GreaterThanOperator()))
self.assertEqual(">=", six.text_type(GreaterThanOrEqualOperator()))
self.assertEqual("<", six.text_type(LessThanOperator()))
self.assertEqual("<=", six.text_type(LessThanOrEqualOperator()))

12
cqlengine/tests/statements/test_delete_statement.py
View File

@@ -1,7 +1,7 @@
from unittest import TestCase
from cqlengine.statements import DeleteStatement, WhereClause, MapDeleteClause
from cqlengine.operators import *
import six
class DeleteStatementTests(TestCase):
@@ -14,24 +14,24 @@ class DeleteStatementTests(TestCase):
def test_field_rendering(self):
""" tests that fields are properly added to the select statement """
ds = DeleteStatement('table', ['f1', 'f2'])
self.assertTrue(unicode(ds).startswith('DELETE "f1", "f2"'), unicode(ds))
self.assertTrue(six.text_type(ds).startswith('DELETE "f1", "f2"'), six.text_type(ds))
self.assertTrue(str(ds).startswith('DELETE "f1", "f2"'), str(ds))
def test_none_fields_rendering(self):
""" tests that a '*' is added if no fields are passed in """
ds = DeleteStatement('table', None)
self.assertTrue(unicode(ds).startswith('DELETE FROM'), unicode(ds))
self.assertTrue(six.text_type(ds).startswith('DELETE FROM'), six.text_type(ds))
self.assertTrue(str(ds).startswith('DELETE FROM'), str(ds))
def test_table_rendering(self):
ds = DeleteStatement('table', None)
self.assertTrue(unicode(ds).startswith('DELETE FROM table'), unicode(ds))
self.assertTrue(six.text_type(ds).startswith('DELETE FROM table'), six.text_type(ds))
self.assertTrue(str(ds).startswith('DELETE FROM table'), str(ds))
def test_where_clause_rendering(self):
ds = DeleteStatement('table', None)
ds.add_where_clause(WhereClause('a', EqualsOperator(), 'b'))
self.assertEqual(unicode(ds), 'DELETE FROM table WHERE "a" = %(0)s', unicode(ds))
self.assertEqual(six.text_type(ds), 'DELETE FROM table WHERE "a" = %(0)s', six.text_type(ds))
def test_context_update(self):
ds = DeleteStatement('table', None)
@@ -39,7 +39,7 @@ class DeleteStatementTests(TestCase):
ds.add_where_clause(WhereClause('a', EqualsOperator(), 'b'))
ds.update_context_id(7)
self.assertEqual(unicode(ds), 'DELETE "d"[%(8)s] FROM table WHERE "a" = %(7)s')
self.assertEqual(six.text_type(ds), 'DELETE "d"[%(8)s] FROM table WHERE "a" = %(7)s')
self.assertEqual(ds.get_context(), {'7': 'b', '8': 3})
def test_context(self):

7
cqlengine/tests/statements/test_insert_statement.py
View File

@@ -1,6 +1,7 @@
from unittest import TestCase
from cqlengine.statements import InsertStatement, StatementException, AssignmentClause
import six
class InsertStatementTests(TestCase):
@@ -16,7 +17,7 @@ class InsertStatementTests(TestCase):
ist.add_assignment_clause(AssignmentClause('c', 'd'))
self.assertEqual(
unicode(ist),
six.text_type(ist),
'INSERT INTO table ("a", "c") VALUES (%(0)s, %(1)s)'
)
@@ -27,7 +28,7 @@ class InsertStatementTests(TestCase):
ist.update_context_id(4)
self.assertEqual(
unicode(ist),
six.text_type(ist),
'INSERT INTO table ("a", "c") VALUES (%(4)s, %(5)s)'
)
ctx = ist.get_context()
@@ -37,4 +38,4 @@ class InsertStatementTests(TestCase):
ist = InsertStatement('table', ttl=60)
ist.add_assignment_clause(AssignmentClause('a', 'b'))
ist.add_assignment_clause(AssignmentClause('c', 'd'))
self.assertIn('USING TTL 60', unicode(ist))
self.assertIn('USING TTL 60', six.text_type(ist))

18
cqlengine/tests/statements/test_select_statement.py
View File

@@ -1,7 +1,7 @@
from unittest import TestCase
from cqlengine.statements import SelectStatement, WhereClause
from cqlengine.operators import *
import six
class SelectStatementTests(TestCase):
@@ -13,31 +13,31 @@ class SelectStatementTests(TestCase):
def test_field_rendering(self):
""" tests that fields are properly added to the select statement """
ss = SelectStatement('table', ['f1', 'f2'])
self.assertTrue(unicode(ss).startswith('SELECT "f1", "f2"'), unicode(ss))
self.assertTrue(six.text_type(ss).startswith('SELECT "f1", "f2"'), six.text_type(ss))
self.assertTrue(str(ss).startswith('SELECT "f1", "f2"'), str(ss))
def test_none_fields_rendering(self):
""" tests that a '*' is added if no fields are passed in """
ss = SelectStatement('table')
self.assertTrue(unicode(ss).startswith('SELECT *'), unicode(ss))
self.assertTrue(six.text_type(ss).startswith('SELECT *'), six.text_type(ss))
self.assertTrue(str(ss).startswith('SELECT *'), str(ss))
def test_table_rendering(self):
ss = SelectStatement('table')
self.assertTrue(unicode(ss).startswith('SELECT * FROM table'), unicode(ss))
self.assertTrue(six.text_type(ss).startswith('SELECT * FROM table'), six.text_type(ss))
self.assertTrue(str(ss).startswith('SELECT * FROM table'), str(ss))
def test_where_clause_rendering(self):
ss = SelectStatement('table')
ss.add_where_clause(WhereClause('a', EqualsOperator(), 'b'))
self.assertEqual(unicode(ss), 'SELECT * FROM table WHERE "a" = %(0)s', unicode(ss))
self.assertEqual(six.text_type(ss), 'SELECT * FROM table WHERE "a" = %(0)s', six.text_type(ss))
def test_count(self):
ss = SelectStatement('table', count=True, limit=10, order_by='d')
ss.add_where_clause(WhereClause('a', EqualsOperator(), 'b'))
self.assertEqual(unicode(ss), 'SELECT COUNT(*) FROM table WHERE "a" = %(0)s LIMIT 10', unicode(ss))
self.assertIn('LIMIT', unicode(ss))
self.assertNotIn('ORDER', unicode(ss))
self.assertEqual(six.text_type(ss), 'SELECT COUNT(*) FROM table WHERE "a" = %(0)s LIMIT 10', six.text_type(ss))
self.assertIn('LIMIT', six.text_type(ss))
self.assertNotIn('ORDER', six.text_type(ss))
def test_context(self):
ss = SelectStatement('table')
@@ -63,7 +63,7 @@ class SelectStatementTests(TestCase):
limit=15,
allow_filtering=True
)
qstr = unicode(ss)
qstr = six.text_type(ss)
self.assertIn('LIMIT 15', qstr)
self.assertIn('ORDER BY x, y', qstr)
self.assertIn('ALLOW FILTERING', qstr)

10
cqlengine/tests/statements/test_update_statement.py
View File

@@ -1,14 +1,14 @@
from unittest import TestCase
from cqlengine.statements import UpdateStatement, WhereClause, AssignmentClause
from cqlengine.operators import *
import six
class UpdateStatementTests(TestCase):
def test_table_rendering(self):
""" tests that fields are properly added to the select statement """
us = UpdateStatement('table')
self.assertTrue(unicode(us).startswith('UPDATE table SET'), unicode(us))
self.assertTrue(six.text_type(us).startswith('UPDATE table SET'), six.text_type(us))
self.assertTrue(str(us).startswith('UPDATE table SET'), str(us))
def test_rendering(self):
@@ -16,7 +16,7 @@ class UpdateStatementTests(TestCase):
us.add_assignment_clause(AssignmentClause('a', 'b'))
us.add_assignment_clause(AssignmentClause('c', 'd'))
us.add_where_clause(WhereClause('a', EqualsOperator(), 'x'))
self.assertEqual(unicode(us), 'UPDATE table SET "a" = %(0)s, "c" = %(1)s WHERE "a" = %(2)s', unicode(us))
self.assertEqual(six.text_type(us), 'UPDATE table SET "a" = %(0)s, "c" = %(1)s WHERE "a" = %(2)s', six.text_type(us))
def test_context(self):
us = UpdateStatement('table')
@@ -31,12 +31,12 @@ class UpdateStatementTests(TestCase):
us.add_assignment_clause(AssignmentClause('c', 'd'))
us.add_where_clause(WhereClause('a', EqualsOperator(), 'x'))
us.update_context_id(3)
self.assertEqual(unicode(us), 'UPDATE table SET "a" = %(4)s, "c" = %(5)s WHERE "a" = %(3)s')
self.assertEqual(six.text_type(us), 'UPDATE table SET "a" = %(4)s, "c" = %(5)s WHERE "a" = %(3)s')
self.assertEqual(us.get_context(), {'4': 'b', '5': 'd', '3': 'x'})
def test_additional_rendering(self):
us = UpdateStatement('table', ttl=60)
us.add_assignment_clause(AssignmentClause('a', 'b'))
us.add_where_clause(WhereClause('a', EqualsOperator(), 'x'))
self.assertIn('USING TTL 60', unicode(us))
self.assertIn('USING TTL 60', six.text_type(us))

3
cqlengine/tests/statements/test_where_clause.py
View File

@@ -1,4 +1,5 @@
from unittest import TestCase
import six
from cqlengine.operators import EqualsOperator
from cqlengine.statements import StatementException, WhereClause
@@ -15,7 +16,7 @@ class TestWhereClause(TestCase):
wc = WhereClause('a', EqualsOperator(), 'c')
wc.set_context_id(5)
self.assertEqual('"a" = %(5)s', unicode(wc), unicode(wc))
self.assertEqual('"a" = %(5)s', six.text_type(wc), six.text_type(wc))
self.assertEqual('"a" = %(5)s', str(wc), type(wc))
def test_equality_method(self):