<s> import pickle <EOL> from time import sleep <EOL> from nose . plugins . attrib import attr <EOL> from couchbase import FMT_JSON , FMT_PICKLE , FMT_UTF8 , FMT_BYTES <EOL> from couchbase . exceptions import ( <EOL> CouchbaseError , ValueFormatError , NotFoundError ) <EOL> from couchbase . result import MultiResult , Result <EOL> from couchbase . tests . base import ConnectionTestCase , SkipTest <EOL> class GetTest ( ConnectionTestCase ) : <EOL> def test_trivial_get ( self ) : <EOL> key = self . gen_key ( '<STR_LIT>' ) <EOL> self . cb . upsert ( key , '<STR_LIT>' ) <EOL> rv = self . cb . get ( key ) <EOL> self . assertEqual ( rv . value , '<STR_LIT>' ) <EOL> rvs = self . cb . get_multi ( [ key ] ) <EOL> self . assertIsInstance ( rvs , self . cls_MultiResult ) <EOL> self . assertEqual ( len ( rvs ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( rvs [ key ] . value , '<STR_LIT>' ) <EOL> def test_get_missing_key ( self ) : <EOL> rv = self . cb . get ( '<STR_LIT>' , quiet = True ) <EOL> self . assertIsNone ( rv . value ) <EOL> self . assertFalse ( rv . success ) <EOL> self . assertRaises ( NotFoundError , self . cb . get , '<STR_LIT>' , <EOL> quiet = False ) <EOL> def test_multi_get ( self ) : <EOL> kv = self . gen_kv_dict ( amount = <NUM_LIT:3> , prefix = '<STR_LIT>' ) <EOL> rvs = self . cb . upsert_multi ( kv ) <EOL> self . assertTrue ( rvs . all_ok ) <EOL> k_subset = list ( kv . keys ( ) ) [ : <NUM_LIT:2> ] <EOL> rvs1 = self . cb . get_multi ( k_subset ) <EOL> self . assertEqual ( len ( rvs1 ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( rvs1 [ k_subset [ <NUM_LIT:0> ] ] . value , kv [ k_subset [ <NUM_LIT:0> ] ] ) <EOL> self . assertEqual ( rvs1 [ k_subset [ <NUM_LIT:1> ] ] . value , kv [ k_subset [ <NUM_LIT:1> ] ] ) <EOL> rv2 = self . cb . get_multi ( kv . keys ( ) ) <EOL> self . assertEqual ( rv2 . keys ( ) , kv . keys ( ) ) <EOL> def test_multi_mixed ( self ) : <EOL> kv_missing = self . gen_kv_dict ( amount = <NUM_LIT:3> , prefix = '<STR_LIT>' ) <EOL> kv_existing = self . gen_kv_dict ( amount = <NUM_LIT:3> , prefix = '<STR_LIT>' ) <EOL> self . cb . remove_multi ( list ( kv_missing . keys ( ) ) + list ( kv_existing . keys ( ) ) , <EOL> quiet = True ) <EOL> self . cb . upsert_multi ( kv_existing ) <EOL> rvs = self . cb . get_multi ( <EOL> list ( kv_existing . keys ( ) ) + list ( kv_missing . keys ( ) ) , <EOL> quiet = True ) <EOL> self . assertFalse ( rvs . all_ok ) <EOL> for k , v in kv_missing . items ( ) : <EOL> self . assertTrue ( k in rvs ) <EOL> self . assertFalse ( rvs [ k ] . success ) <EOL> self . assertTrue ( rvs [ k ] . value is None ) <EOL> self . assertTrue ( NotFoundError . _can_derive ( rvs [ k ] . rc ) ) <EOL> for k , v in kv_existing . items ( ) : <EOL> self . assertTrue ( k in rvs ) <EOL> self . assertTrue ( rvs [ k ] . success ) <EOL> self . assertEqual ( rvs [ k ] . value , kv_existing [ k ] ) <EOL> self . assertEqual ( rvs [ k ] . rc , <NUM_LIT:0> ) <EOL> cb_exc = None <EOL> try : <EOL> self . cb . get_multi ( list ( kv_existing . keys ( ) ) + list ( kv_missing . keys ( ) ) ) <EOL> except NotFoundError as e : <EOL> cb_exc = e <EOL> self . assertTrue ( cb_exc ) <EOL> all_res = cb_exc . all_results <EOL> self . assertTrue ( all_res ) <EOL> self . assertFalse ( all_res . all_ok ) <EOL> for k , v in kv_existing . items ( ) : <EOL> self . assertTrue ( k in all_res ) <EOL> self . assertTrue ( all_res [ k ] . success ) <EOL> self . assertEqual ( all_res [ k ] . value , v ) <EOL> self . assertEqual ( all_res [ k ] . rc , <NUM_LIT:0> ) <EOL> for k , v in kv_missing . items ( ) : <EOL> self . assertTrue ( k in all_res ) <EOL> self . assertFalse ( all_res [ k ] . success ) <EOL> self . assertTrue ( all_res [ k ] . value is None ) <EOL> del cb_exc <EOL> def test_extended_get ( self ) : <EOL> key = self . gen_key ( prefix = '<STR_LIT>' ) <EOL> orig_cas1 = self . cb . upsert ( key , '<STR_LIT>' ) . cas <EOL> rv = self . cb . get ( key ) <EOL> val1 , flags1 , cas1 = rv . value , rv . flags , rv . cas <EOL> self . assertEqual ( val1 , '<STR_LIT>' ) <EOL> self . assertEqual ( flags1 , FMT_JSON ) <EOL> self . assertEqual ( cas1 , orig_cas1 ) <EOL> result1 = self . cb . get ( key ) <EOL> self . assertEqual ( result1 . value , '<STR_LIT>' ) <EOL> self . assertEqual ( result1 . flags , FMT_JSON ) <EOL> self . assertEqual ( result1 . cas , orig_cas1 ) <EOL> result2 = self . cb . get_multi ( [ key ] ) <EOL> self . assertIsInstance ( result2 , self . cls_MultiResult ) <EOL> self . assertTrue ( key in result2 ) <EOL> self . assertEqual ( result2 [ key ] . value , '<STR_LIT>' ) <EOL> self . assertEqual ( result2 [ key ] . flags , FMT_JSON ) <EOL> self . assertEqual ( result2 [ key ] . cas , orig_cas1 ) <EOL> key2 = self . gen_key ( '<STR_LIT>' ) <EOL> cas2 = self . cb . upsert ( key2 , '<STR_LIT>' ) . cas <EOL> key3 = self . gen_key ( '<STR_LIT>' ) <EOL> cas3 = self . cb . upsert ( key3 , '<STR_LIT>' ) . cas <EOL> results = self . cb . get_multi ( [ key2 , key3 ] ) <EOL> self . assertEqual ( results [ key3 ] . value , '<STR_LIT>' ) <EOL> self . assertEqual ( results [ key3 ] . flags , FMT_JSON ) <EOL> self . assertEqual ( results [ key3 ] . cas , cas3 ) <EOL> rv = self . cb . get ( '<STR_LIT>' , quiet = True ) <EOL> val4 , flags4 , cas4 = rv . value , rv . flags , rv . cas <EOL> self . assertEqual ( val4 , None ) <EOL> self . assertEqual ( flags4 , <NUM_LIT> ) <EOL> self . assertEqual ( cas4 , <NUM_LIT:0> ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_get_ttl ( self ) : <EOL> key = self . gen_key ( '<STR_LIT>' ) <EOL> self . cb . remove ( key , quiet = True ) <EOL> self . cb . upsert ( key , "<STR_LIT>" ) <EOL> rv = self . cb . get ( key , ttl = <NUM_LIT:1> ) <EOL> self . assertEqual ( rv . value , "<STR_LIT>" ) <EOL> sleep ( <NUM_LIT:2> ) <EOL> rv = self . cb . get ( key , quiet = True ) <EOL> self . assertFalse ( rv . success ) <EOL> self . assertTrue ( NotFoundError . _can_derive ( rv . rc ) ) <EOL> @ attr ( '<STR_LIT>' ) <EOL> def test_get_multi_ttl ( self ) : <EOL> kvs = self . gen_kv_dict ( amount = <NUM_LIT:2> , prefix = '<STR_LIT>' ) <EOL> self . cb . upsert_multi ( kvs ) <EOL> rvs = self . cb . get_multi ( list ( kvs . keys ( ) ) , ttl = <NUM_LIT:1> ) <EOL> for k , v in rvs . items ( ) : <EOL> self . assertEqual ( v . value , kvs [ k ] ) <EOL> sleep ( <NUM_LIT:2> ) <EOL> rvs = self . cb . get_multi ( list ( kvs . keys ( ) ) , quiet = True ) <EOL> for k , v in rvs . items ( ) : <EOL> self . assertFalse ( v . success ) <EOL> self . assertTrue ( k in kvs ) <EOL> self . assertTrue ( NotFoundError . _can_derive ( v . rc ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
