<s> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> import mock <EOL> import six <EOL> from uuid import uuid4 <EOL> from cassandra . cqlengine import columns <EOL> from cassandra . cqlengine . management import sync_table , drop_table <EOL> from cassandra . cqlengine . models import Model <EOL> from cassandra . cqlengine . query import BatchQuery , LWTException <EOL> from cassandra . cqlengine . statements import ConditionalClause <EOL> from tests . integration . cqlengine . base import BaseCassEngTestCase <EOL> from tests . integration import CASSANDRA_VERSION , greaterthancass20 <EOL> class TestConditionalModel ( Model ) : <EOL> id = columns . UUID ( primary_key = True , default = uuid4 ) <EOL> count = columns . Integer ( ) <EOL> text = columns . Text ( required = False ) <EOL> @ unittest . skipUnless ( CASSANDRA_VERSION >= '<STR_LIT>' , "<STR_LIT>" ) <EOL> class TestConditional ( BaseCassEngTestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( TestConditional , cls ) . setUpClass ( ) <EOL> sync_table ( TestConditionalModel ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> super ( TestConditional , cls ) . tearDownClass ( ) <EOL> drop_table ( TestConditionalModel ) <EOL> def test_update_using_conditional ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' ) <EOL> t . text = '<STR_LIT>' <EOL> with mock . patch . object ( self . session , '<STR_LIT>' ) as m : <EOL> t . iff ( text = '<STR_LIT>' ) . save ( ) <EOL> args = m . call_args <EOL> self . assertIn ( '<STR_LIT>' , args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . query_string ) <EOL> def test_update_conditional_success ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> id = t . id <EOL> t . text = '<STR_LIT>' <EOL> t . iff ( text = '<STR_LIT>' ) . save ( ) <EOL> updated = TestConditionalModel . objects ( id = id ) . first ( ) <EOL> self . assertEqual ( updated . count , <NUM_LIT:5> ) <EOL> self . assertEqual ( updated . text , '<STR_LIT>' ) <EOL> def test_update_failure ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' ) <EOL> t . text = '<STR_LIT>' <EOL> t = t . iff ( text = '<STR_LIT>' ) <EOL> with self . assertRaises ( LWTException ) as assertion : <EOL> t . save ( ) <EOL> self . assertEqual ( assertion . exception . existing , { <EOL> '<STR_LIT:text>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> } ) <EOL> def test_blind_update ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' ) <EOL> t . text = '<STR_LIT>' <EOL> uid = t . id <EOL> with mock . patch . object ( self . session , '<STR_LIT>' ) as m : <EOL> TestConditionalModel . objects ( id = uid ) . iff ( text = '<STR_LIT>' ) . update ( text = '<STR_LIT>' ) <EOL> args = m . call_args <EOL> self . assertIn ( '<STR_LIT>' , args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . query_string ) <EOL> def test_blind_update_fail ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' ) <EOL> t . text = '<STR_LIT>' <EOL> uid = t . id <EOL> qs = TestConditionalModel . objects ( id = uid ) . iff ( text = '<STR_LIT>' ) <EOL> with self . assertRaises ( LWTException ) as assertion : <EOL> qs . update ( text = '<STR_LIT>' ) <EOL> self . assertEqual ( assertion . exception . existing , { <EOL> '<STR_LIT:text>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : False , <EOL> } ) <EOL> def test_conditional_clause ( self ) : <EOL> tc = ConditionalClause ( '<STR_LIT>' , <NUM_LIT> ) <EOL> tc . set_context_id ( <NUM_LIT:3> ) <EOL> self . assertEqual ( '<STR_LIT>' , six . text_type ( tc ) ) <EOL> self . assertEqual ( '<STR_LIT>' , str ( tc ) ) <EOL> def test_batch_update_conditional ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> id = t . id <EOL> with BatchQuery ( ) as b : <EOL> t . batch ( b ) . iff ( count = <NUM_LIT:5> ) . update ( text = '<STR_LIT>' ) <EOL> updated = TestConditionalModel . objects ( id = id ) . first ( ) <EOL> self . assertEqual ( updated . text , '<STR_LIT>' ) <EOL> b = BatchQuery ( ) <EOL> updated . batch ( b ) . iff ( count = <NUM_LIT:6> ) . update ( text = '<STR_LIT>' ) <EOL> with self . assertRaises ( LWTException ) as assertion : <EOL> b . execute ( ) <EOL> self . assertEqual ( assertion . exception . existing , { <EOL> '<STR_LIT:id>' : id , <EOL> '<STR_LIT:count>' : <NUM_LIT:5> , <EOL> '<STR_LIT>' : False , <EOL> } ) <EOL> updated = TestConditionalModel . objects ( id = id ) . first ( ) <EOL> self . assertEqual ( updated . text , '<STR_LIT>' ) <EOL> def test_delete_conditional ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> t . iff ( count = <NUM_LIT> ) . delete ( ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> t . iff ( count = <NUM_LIT:5> ) . delete ( ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:0> ) <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count = <NUM_LIT> ) . delete ( ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count = <NUM_LIT:5> ) . delete ( ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:0> ) <EOL> @ greaterthancass20 <EOL> def test_delete_lwt_ne ( self ) : <EOL> """<STR_LIT>""" <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> t . iff ( count__ne = <NUM_LIT:5> ) . delete ( ) <EOL> t . iff ( count__ne = <NUM_LIT:2> ) . delete ( ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:0> ) <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count__ne = <NUM_LIT:5> ) . delete ( ) <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count__ne = <NUM_LIT:2> ) . delete ( ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:0> ) <EOL> @ greaterthancass20 <EOL> def test_update_lwt_ne ( self ) : <EOL> """<STR_LIT>""" <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> t . iff ( count__ne = <NUM_LIT:5> ) . update ( text = '<STR_LIT>' ) <EOL> t . iff ( count__ne = <NUM_LIT:2> ) . update ( text = '<STR_LIT>' ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . first ( ) . text , '<STR_LIT>' ) <EOL> t . delete ( ) <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count__ne = <NUM_LIT:5> ) . update ( text = '<STR_LIT>' ) <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count__ne = <NUM_LIT:2> ) . update ( text = '<STR_LIT>' ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . first ( ) . text , '<STR_LIT>' ) <EOL> t . delete ( ) <EOL> def test_update_to_none ( self ) : <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> t . iff ( count = <NUM_LIT> ) . update ( text = None ) <EOL> self . assertIsNotNone ( TestConditionalModel . objects ( id = t . id ) . first ( ) . text ) <EOL> t . iff ( count = <NUM_LIT:5> ) . update ( text = None ) <EOL> self . assertIsNone ( TestConditionalModel . objects ( id = t . id ) . first ( ) . text ) <EOL> t = TestConditionalModel . create ( text = '<STR_LIT>' , count = <NUM_LIT:5> ) <EOL> self . assertEqual ( TestConditionalModel . objects ( id = t . id ) . count ( ) , <NUM_LIT:1> ) <EOL> with self . assertRaises ( LWTException ) : <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count = <NUM_LIT> ) . update ( text = None ) <EOL> self . assertIsNotNone ( TestConditionalModel . objects ( id = t . id ) . first ( ) . text ) <EOL> TestConditionalModel . objects ( id = t . id ) . iff ( count = <NUM_LIT:5> ) . update ( text = None ) <EOL> self . assertIsNone ( TestConditionalModel . objects ( id = t . id ) . first ( ) . text ) </s>
