test_types_cu.py

try:
	import unittest2 as unittest
except ImportError:
	import unittest # noqa

import logging
_log = logging.getLogger(__name__)

import uuid

from cassandra import cluster as cassandra_cluster



def _keyspace_prepare():
	c = cassandra_cluster.Cluster()
	s = c.connect()
	s.execute("DROP KEYSPACE IF EXISTS \"TestSpace\"")
	s.execute("CREATE KEYSPACE \"TestSpace\" WITH replication = {'class' : 'SimpleStrategy', 'replication_factor': 1}")
	s.set_keyspace("TestSpace")
	return (c, s,)

def _keyspace_shutdown(c, s):
	try:
		s.execute("DROP KEYSPACE IF EXISTS \"TestSpace\"")
		s.shutdown()
	except:
		pass
	try:
		c.shutdown()
	except:
		pass
	return (None, None,)

class TestBlob(unittest.TestCase):
	def __init__(self, *args, **kwds):
		super(TestBlob, self).__init__(*args, **kwds)
		self.c = None
		self.s = None

	def setUp(self):
		self.s, self.c, = _keyspace_shutdown(self.s, self.c)
		self.c, self.s, = _keyspace_prepare()

	def tearDown(self):
		# self.s, self.c, = _keyspace_shutdown()
		pass

	def test_insert_plain(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val blob, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = bytearray("this is a apple", "utf-8", "ignore")
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])

	def test_insert_map(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val map<uuid, blob>, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = {
				uuid.uuid4(): bytearray("this is value-1", "utf-8", "ignore"),
				uuid.uuid4(): bytearray("this is value-2", "utf-8", "ignore"), }
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])

		return (t_id, t_val,)

	def test_update_map(self):
		t_id, t_val, = self.test_insert_map()
		t_extra = {uuid.uuid4(): bytearray("this is value-3", "utf-8", "ignore"), }
		t_val.update(t_extra)
		s = self.s

		s.execute("UPDATE test_blob_table SET t_val = t_val + %s WHERE (t_id = %s)", (t_extra, t_id,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])


	def test_insert_list(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val list<blob>, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = [
				bytearray("this is value-1", "utf-8", "ignore"),
				bytearray("this is value-2", "utf-8", "ignore"),
				bytearray("this is value-3", "utf-8", "ignore"), ]
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, map(str, row[1]))



class TestBoolean(unittest.TestCase):
	def __init__(self, *args, **kwds):
		super(TestBoolean, self).__init__(*args, **kwds)
		self.c = None
		self.s = None

	def setUp(self):
		self.s, self.c, = _keyspace_shutdown(self.s, self.c)
		self.c, self.s, = _keyspace_prepare()

	def tearDown(self):
		# self.s, self.c, = _keyspace_shutdown()
		pass

	def test_insert_plain_T(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val boolean, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = True
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])

		return t_id

	def test_insert_plain_F(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val boolean, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = False
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])

		return t_id

	def test_update_plan_T2T(self):
		t_id = self.test_insert_plain_T()
		t_val = True
		s = self.s

		s.execute("UPDATE test_blob_table SET t_val=%s WHERE t_id = %s", (t_val, t_id,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])



class TestUUID(unittest.TestCase):
	def __init__(self, *args, **kwds):
		super(TestUUID, self).__init__(*args, **kwds)
		self.c = None
		self.s = None

	def setUp(self):
		self.s, self.c, = _keyspace_shutdown(self.s, self.c)
		self.c, self.s, = _keyspace_prepare()

	def tearDown(self):
		# self.s, self.c, = _keyspace_shutdown()
		pass

	def test_insert_plain(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val uuid, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = uuid.uuid4()
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, row[1])

	def test_insert_set(self):
		s = self.s
		s.execute("CREATE TABLE test_blob_table(t_id timeuuid, t_val set<uuid>, PRIMARY KEY(t_id))")

		t_id = uuid.uuid1()
		t_val = set()
		t_val.add(uuid.uuid4())
		t_val.add(uuid.uuid4())
		s.execute("INSERT INTO test_blob_table(t_id, t_val) VALUES(%s, %s)", (t_id, t_val,))

		result = s.execute("SELECT t_id, t_val FROM test_blob_table")
		for row in result:
			self.assertEqual(t_id, row[0])
			self.assertEqual(t_val, set(row[1]))

		return (t_id, t_val,)



if __name__ == '__main__':
	unittest.main()

# vim: ts=4 sw=4 foldmethod=marker ai nowrap