<s> from time import sleep <EOL> from nose . plugins . attrib import attr <EOL> from couchbase import FMT_JSON , FMT_PICKLE , FMT_BYTES , FMT_UTF8 <EOL> from couchbase . exceptions import ( KeyExistsError , ValueFormatError , <EOL> ArgumentError , NotFoundError , <EOL> NotStoredError ) <EOL> from couchbase . tests . base import ConnectionTestCase <EOL> class UpsertTest ( ConnectionTestCase ) : <EOL> def test_trivial_set ( self ) : <EOL> rv = self . cb . upsert ( self . gen_key ( ) , '<STR_LIT>' ) <EOL> self . assertTrue ( rv ) <EOL> self . assertTrue ( rv . cas > <NUM_LIT:0> ) <EOL> rv = self . cb . upsert ( self . gen_key ( ) , '<STR_LIT>' ) <EOL> self . assertTrue ( rv . cas > <NUM_LIT:0> ) <EOL> def test_set_with_cas ( self ) : <EOL> key = self . gen_key ( '<STR_LIT>' ) <EOL> rv1 = self . cb . upsert ( key , '<STR_LIT>' ) <EOL> self . assertTrue ( rv1 . cas > <NUM_LIT:0> ) <EOL> self . assertRaises ( KeyExistsError , self . cb . upsert , <EOL> key , '<STR_LIT>' , cas = rv1 . cas + <NUM_LIT:1> ) <EOL> rv2 = self . cb . upsert ( key , '<STR_LIT>' , cas = rv1 . cas ) <EOL> self . assertTrue ( rv2 . cas > <NUM_LIT:0> ) <EOL> self . assertNotEqual ( rv1 . cas , rv2 . cas ) <EOL> rv3 = self . cb . upsert ( key , '<STR_LIT>' ) <EOL> self . assertTrue ( rv3 . cas > <NUM_LIT:0> ) <EOL> self . assertNotEqual ( rv3 . cas , rv2 . cas ) <EOL> self . assertNotEqual ( rv3 . cas , rv1 . cas ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_set_with_ttl ( self ) : <EOL> key = self . gen_key ( '<STR_LIT>' ) <EOL> self . cb . upsert ( key , '<STR_LIT>' , ttl = <NUM_LIT:2> ) <EOL> rv = self . cb . get ( key ) <EOL> self . assertEqual ( rv . value , '<STR_LIT>' ) <EOL> sleep ( <NUM_LIT:3> ) <EOL> self . assertRaises ( NotFoundError , self . cb . get , key ) <EOL> def test_set_objects ( self ) : <EOL> key = self . gen_key ( '<STR_LIT>' ) <EOL> for v in ( None , False , True ) : <EOL> for fmt in ( FMT_JSON , FMT_PICKLE ) : <EOL> rv = self . cb . upsert ( key , v , format = fmt ) <EOL> self . assertTrue ( rv . success ) <EOL> rv = self . cb . get ( key ) <EOL> self . assertTrue ( rv . success ) <EOL> self . assertEqual ( rv . value , v ) <EOL> def test_multi_set ( self ) : <EOL> kv = self . gen_kv_dict ( prefix = '<STR_LIT>' ) <EOL> rvs = self . cb . upsert_multi ( kv ) <EOL> self . assertTrue ( rvs . all_ok ) <EOL> for k , v in rvs . items ( ) : <EOL> self . assertTrue ( v . success ) <EOL> self . assertTrue ( v . cas > <NUM_LIT:0> ) <EOL> for k , v in rvs . items ( ) : <EOL> self . assertTrue ( k in rvs ) <EOL> self . assertTrue ( rvs [ k ] . success ) <EOL> self . assertRaises ( ( ArgumentError , TypeError ) , self . cb . upsert_multi , kv , <EOL> cas = <NUM_LIT> ) <EOL> def test_add ( self ) : <EOL> key = self . gen_key ( '<STR_LIT>' ) <EOL> self . cb . remove ( key , quiet = True ) <EOL> rv = self . cb . insert ( key , "<STR_LIT:value>" ) <EOL> self . assertTrue ( rv . cas ) <EOL> self . assertRaises ( KeyExistsError , <EOL> self . cb . insert , key , "<STR_LIT:value>" ) <EOL> def test_replace ( self ) : <EOL> key = self . gen_key ( '<STR_LIT:replace>' ) <EOL> rv = self . cb . upsert ( key , "<STR_LIT:value>" ) <EOL> self . assertTrue ( rv . success ) <EOL> rv = self . cb . replace ( key , "<STR_LIT:value>" ) <EOL> self . assertTrue ( rv . cas ) <EOL> rv = self . cb . replace ( key , "<STR_LIT:value>" , cas = rv . cas ) <EOL> self . assertTrue ( rv . cas ) <EOL> self . assertRaises ( KeyExistsError , <EOL> self . cb . replace , key , "<STR_LIT:value>" , cas = <NUM_LIT> ) <EOL> self . cb . remove ( key , quiet = True ) <EOL> self . assertRaises ( NotFoundError , <EOL> self . cb . replace , key , "<STR_LIT:value>" ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
