104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
from unittest import skip
|
|
from cqlengine.tests.base import BaseCassEngTestCase
|
|
|
|
from cqlengine.management import create_table
|
|
from cqlengine.management import delete_table
|
|
from cqlengine.models import Model
|
|
from cqlengine import columns
|
|
|
|
class TestModel(Model):
|
|
count = columns.Integer()
|
|
text = columns.Text(required=False)
|
|
|
|
class TestModelIO(BaseCassEngTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestModelIO, cls).setUpClass()
|
|
create_table(TestModel)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(TestModelIO, cls).tearDownClass()
|
|
delete_table(TestModel)
|
|
|
|
def test_model_save_and_load(self):
|
|
"""
|
|
Tests that models can be saved and retrieved
|
|
"""
|
|
tm = TestModel.create(count=8, text='123456789')
|
|
tm2 = TestModel.objects(id=tm.pk).first()
|
|
|
|
for cname in tm._columns.keys():
|
|
self.assertEquals(getattr(tm, cname), getattr(tm2, cname))
|
|
|
|
def test_model_updating_works_properly(self):
|
|
"""
|
|
Tests that subsequent saves after initial model creation work
|
|
"""
|
|
tm = TestModel.objects.create(count=8, text='123456789')
|
|
|
|
tm.count = 100
|
|
tm.save()
|
|
|
|
tm2 = TestModel.objects(id=tm.pk).first()
|
|
self.assertEquals(tm.count, tm2.count)
|
|
|
|
def test_model_deleting_works_properly(self):
|
|
"""
|
|
Tests that an instance's delete method deletes the instance
|
|
"""
|
|
tm = TestModel.create(count=8, text='123456789')
|
|
tm.delete()
|
|
tm2 = TestModel.objects(id=tm.pk).first()
|
|
self.assertIsNone(tm2)
|
|
|
|
def test_column_deleting_works_properly(self):
|
|
"""
|
|
"""
|
|
tm = TestModel.create(count=8, text='123456789')
|
|
tm.text = None
|
|
tm.save()
|
|
|
|
tm2 = TestModel.objects(id=tm.pk).first()
|
|
assert tm2.text is None
|
|
assert tm2._values['text'].previous_value is None
|
|
|
|
|
|
def test_a_sensical_error_is_raised_if_you_try_to_create_a_table_twice(self):
|
|
"""
|
|
"""
|
|
create_table(TestModel)
|
|
create_table(TestModel)
|
|
|
|
class IndexDefinitionModel(Model):
|
|
key = columns.UUID(primary_key=True)
|
|
val = columns.Text(index=True)
|
|
|
|
class TestIndexedColumnDefinition(BaseCassEngTestCase):
|
|
|
|
def test_exception_isnt_raised_if_an_index_is_defined_more_than_once(self):
|
|
create_table(IndexDefinitionModel)
|
|
create_table(IndexDefinitionModel)
|
|
|
|
class ReservedWordModel(Model):
|
|
token = columns.Text(primary_key=True)
|
|
insert = columns.Integer(index=True)
|
|
|
|
class TestQueryQuoting(BaseCassEngTestCase):
|
|
|
|
def test_reserved_cql_words_can_be_used_as_column_names(self):
|
|
"""
|
|
"""
|
|
create_table(ReservedWordModel)
|
|
|
|
model1 = ReservedWordModel.create(token='1', insert=5)
|
|
|
|
model2 = ReservedWordModel.filter(token='1')
|
|
|
|
assert len(model2) == 1
|
|
assert model1.token == model2[0].token
|
|
assert model1.insert == model2[0].insert
|
|
|
|
|