| import unittest |
| import datetime |
| import cassandra |
| from cassandra.cqltypes import CassandraType, BooleanType, lookup_casstype_simple, lookup_casstype, \ |
| AsciiType, LongType, DecimalType, DoubleType, FloatType, Int32Type, UTF8Type, IntegerType, SetType, cql_typename |
| |
| from cassandra.cluster import Cluster |
| |
| |
| class TypeTests(unittest.TestCase): |
| |
| def test_lookup_casstype_simple(self): |
| """ |
| Ensure lookup_casstype_simple returns the correct classes |
| """ |
| |
| self.assertEqual(lookup_casstype_simple('AsciiType'), cassandra.cqltypes.AsciiType) |
| self.assertEqual(lookup_casstype_simple('LongType'), cassandra.cqltypes.LongType) |
| self.assertEqual(lookup_casstype_simple('BytesType'), cassandra.cqltypes.BytesType) |
| self.assertEqual(lookup_casstype_simple('BooleanType'), cassandra.cqltypes.BooleanType) |
| self.assertEqual(lookup_casstype_simple('CounterColumnType'), cassandra.cqltypes.CounterColumnType) |
| self.assertEqual(lookup_casstype_simple('DecimalType'), cassandra.cqltypes.DecimalType) |
| self.assertEqual(lookup_casstype_simple('DoubleType'), cassandra.cqltypes.DoubleType) |
| self.assertEqual(lookup_casstype_simple('FloatType'), cassandra.cqltypes.FloatType) |
| self.assertEqual(lookup_casstype_simple('InetAddressType'), cassandra.cqltypes.InetAddressType) |
| self.assertEqual(lookup_casstype_simple('Int32Type'), cassandra.cqltypes.Int32Type) |
| self.assertEqual(lookup_casstype_simple('UTF8Type'), cassandra.cqltypes.UTF8Type) |
| self.assertEqual(lookup_casstype_simple('DateType'), cassandra.cqltypes.DateType) |
| self.assertEqual(lookup_casstype_simple('TimeUUIDType'), cassandra.cqltypes.TimeUUIDType) |
| self.assertEqual(lookup_casstype_simple('UUIDType'), cassandra.cqltypes.UUIDType) |
| self.assertEqual(lookup_casstype_simple('IntegerType'), cassandra.cqltypes.IntegerType) |
| self.assertEqual(lookup_casstype_simple('MapType'), cassandra.cqltypes.MapType) |
| self.assertEqual(lookup_casstype_simple('ListType'), cassandra.cqltypes.ListType) |
| self.assertEqual(lookup_casstype_simple('SetType'), cassandra.cqltypes.SetType) |
| self.assertEqual(lookup_casstype_simple('CompositeType'), cassandra.cqltypes.CompositeType) |
| self.assertEqual(lookup_casstype_simple('ColumnToCollectionType'), cassandra.cqltypes.ColumnToCollectionType) |
| self.assertEqual(lookup_casstype_simple('ReversedType'), cassandra.cqltypes.ReversedType) |
| |
| self.assertEqual(str(lookup_casstype_simple('unknown')), str(cassandra.cqltypes.mkUnrecognizedType('unknown'))) |
| |
| def test_lookup_casstype(self): |
| """ |
| Ensure lookup_casstype returns the correct classes |
| """ |
| |
| self.assertEqual(lookup_casstype('AsciiType'), cassandra.cqltypes.AsciiType) |
| self.assertEqual(lookup_casstype('LongType'), cassandra.cqltypes.LongType) |
| self.assertEqual(lookup_casstype('BytesType'), cassandra.cqltypes.BytesType) |
| self.assertEqual(lookup_casstype('BooleanType'), cassandra.cqltypes.BooleanType) |
| self.assertEqual(lookup_casstype('CounterColumnType'), cassandra.cqltypes.CounterColumnType) |
| self.assertEqual(lookup_casstype('DecimalType'), cassandra.cqltypes.DecimalType) |
| self.assertEqual(lookup_casstype('DoubleType'), cassandra.cqltypes.DoubleType) |
| self.assertEqual(lookup_casstype('FloatType'), cassandra.cqltypes.FloatType) |
| self.assertEqual(lookup_casstype('InetAddressType'), cassandra.cqltypes.InetAddressType) |
| self.assertEqual(lookup_casstype('Int32Type'), cassandra.cqltypes.Int32Type) |
| self.assertEqual(lookup_casstype('UTF8Type'), cassandra.cqltypes.UTF8Type) |
| self.assertEqual(lookup_casstype('DateType'), cassandra.cqltypes.DateType) |
| self.assertEqual(lookup_casstype('TimeUUIDType'), cassandra.cqltypes.TimeUUIDType) |
| self.assertEqual(lookup_casstype('UUIDType'), cassandra.cqltypes.UUIDType) |
| self.assertEqual(lookup_casstype('IntegerType'), cassandra.cqltypes.IntegerType) |
| self.assertEqual(lookup_casstype('MapType'), cassandra.cqltypes.MapType) |
| self.assertEqual(lookup_casstype('ListType'), cassandra.cqltypes.ListType) |
| self.assertEqual(lookup_casstype('SetType'), cassandra.cqltypes.SetType) |
| self.assertEqual(lookup_casstype('CompositeType'), cassandra.cqltypes.CompositeType) |
| self.assertEqual(lookup_casstype('ColumnToCollectionType'), cassandra.cqltypes.ColumnToCollectionType) |
| self.assertEqual(lookup_casstype('ReversedType'), cassandra.cqltypes.ReversedType) |
| |
| self.assertEqual(str(lookup_casstype('unknown')), str(cassandra.cqltypes.mkUnrecognizedType('unknown'))) |
| |
| self.assertRaises(ValueError, lookup_casstype, 'AsciiType~') |
| |
| # TODO: Do a few more tests |
| # "I would say some parameterized and nested types would be good to test, |
| # like "MapType(AsciiType, IntegerType)" and "ReversedType(AsciiType)" |
| self.assertEqual(str(lookup_casstype(BooleanType(True))), str(BooleanType(True))) |
| |
| def test_cassandratype(self): |
| """ |
| Smoke test cass_parameterized_type_with |
| """ |
| |
| self.assertEqual(LongType.cass_parameterized_type_with(()), 'LongType') |
| self.assertEqual(LongType.cass_parameterized_type_with((), full=True), 'org.apache.cassandra.db.marshal.LongType') |
| self.assertEqual(SetType.cass_parameterized_type_with([DecimalType], full=True), 'org.apache.cassandra.db.marshal.SetType(org.apache.cassandra.db.marshal.DecimalType)') |
| |
| self.assertEqual(LongType.cql_parameterized_type(), 'bigint') |
| self.assertEqual(cassandra.cqltypes.MapType.apply_parameters( |
| cassandra.cqltypes.UTF8Type, cassandra.cqltypes.UTF8Type).cql_parameterized_type(), |
| 'map<text, text>') |
| |
| def test_datetype(self): |
| """ |
| Test cassandra.cqltypes.DateType() construction |
| """ |
| |
| # Ensure all formats can be parsed, without exception |
| for format in cassandra.cqltypes.cql_time_formats: |
| date_string = str(datetime.datetime.now().strftime(format)) |
| cassandra.cqltypes.DateType(date_string) |
| |
| def test_cql_typename(self): |
| """ |
| Smoke test cql_typename |
| """ |
| |
| self.assertEqual(cql_typename('DateType'), 'timestamp') |
| self.assertEqual(cql_typename('org.apache.cassandra.db.marshal.ListType(IntegerType)'), 'list<varint>') |
