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

updates can use consistency levels

Browse Source
This commit is contained in:
Jon Haddad
2013-10-24 17:54:57 -07:00
parent e590a8fb6d
commit da58f8b2fb
2 changed files with 18 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
cqlengine/models.py
View File

@@ -100,7 +100,7 @@ class ConsistencyDescriptor(object):
def __get__(self, instance, model):
if instance:
def consistency_setter(consistency):
instance._consistency = consistency
instance.__consistency__ = consistency
return instance
return consistency_setter
@@ -416,7 +416,10 @@ class BaseModel(object):
is_new = self.pk is None
self.validate()
self.__dmlquery__(self.__class__, self, batch=self._batch, ttl=self._ttl).save()
self.__dmlquery__(self.__class__, self,
batch=self._batch,
ttl=self._ttl,
consistency=self.__consistency__).save()
#reset the value managers
for v in self._values.values():
@@ -450,7 +453,7 @@ class BaseModel(object):
self.__dmlquery__(self.__class__, self,
batch=self._batch,
ttl=self._ttl,
consistency=self.consistency).update()
consistency=self.__consistency__).update()
#reset the value managers
for v in self._values.values():

13
cqlengine/tests/test_consistency.py
View File

@@ -28,8 +28,9 @@ class BaseConsistencyTest(BaseCassEngTestCase):
class TestConsistency(BaseConsistencyTest):
def test_create_uses_consistency(self):
qs = TestConsistencyModel.consistency(ALL)
with mock.patch.object(ConnectionPool, 'execute') as m:
TestConsistencyModel.consistency(ALL).create(text="i am not fault tolerant this way")
qs.create(text="i am not fault tolerant this way")
args = m.call_args
self.assertEqual(ALL, args[0][2])
@@ -38,4 +39,14 @@ class TestConsistency(BaseConsistencyTest):
qs = TestConsistencyModel.consistency(ALL)
self.assertTrue(isinstance(qs, TestConsistencyModel.__queryset__), type(qs))
def test_update_uses_consistency(self):
t = TestConsistencyModel.create(text="bacon and eggs")
t.text = "ham sandwich"
with mock.patch.object(ConnectionPool, 'execute') as m:
t.consistency(ALL).save()
args = m.call_args
self.assertEqual(ALL, args[0][2])