<s> import unittest <EOL> from riak import RiakError , RiakObject <EOL> from riak . bucket import RiakBucket , BucketType <EOL> from riak . tests import RUN_BTYPES <EOL> from riak . tests . base import IntegrationTestBase <EOL> from riak . tests . comparison import Comparison <EOL> @ unittest . skipUnless ( RUN_BTYPES , "<STR_LIT>" ) <EOL> class BucketTypeTests ( IntegrationTestBase , unittest . TestCase , Comparison ) : <EOL> def test_btype_init ( self ) : <EOL> btype = self . client . bucket_type ( '<STR_LIT:foo>' ) <EOL> self . assertIsInstance ( btype , BucketType ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , btype . name ) <EOL> self . assertIs ( btype , self . client . bucket_type ( '<STR_LIT:foo>' ) ) <EOL> def test_btype_get_bucket ( self ) : <EOL> btype = self . client . bucket_type ( '<STR_LIT:foo>' ) <EOL> bucket = btype . bucket ( self . bucket_name ) <EOL> self . assertIsInstance ( bucket , RiakBucket ) <EOL> self . assertIs ( btype , bucket . bucket_type ) <EOL> self . assertIs ( bucket , <EOL> self . client . bucket_type ( '<STR_LIT:foo>' ) . bucket ( self . bucket_name ) ) <EOL> self . assertIsNot ( bucket , self . client . bucket ( self . bucket_name ) ) <EOL> def test_btype_default ( self ) : <EOL> defbtype = self . client . bucket_type ( '<STR_LIT:default>' ) <EOL> othertype = self . client . bucket_type ( '<STR_LIT:foo>' ) <EOL> self . assertTrue ( defbtype . is_default ( ) ) <EOL> self . assertFalse ( othertype . is_default ( ) ) <EOL> def test_btype_repr ( self ) : <EOL> defbtype = self . client . bucket_type ( "<STR_LIT:default>" ) <EOL> othertype = self . client . bucket_type ( "<STR_LIT:foo>" ) <EOL> self . assertEqual ( "<STR_LIT>" , str ( defbtype ) ) <EOL> self . assertEqual ( "<STR_LIT>" , str ( othertype ) ) <EOL> self . assertEqual ( "<STR_LIT>" , repr ( defbtype ) ) <EOL> self . assertEqual ( "<STR_LIT>" , repr ( othertype ) ) <EOL> def test_btype_get_props ( self ) : <EOL> defbtype = self . client . bucket_type ( "<STR_LIT:default>" ) <EOL> btype = self . client . bucket_type ( '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError ) : <EOL> defbtype . get_properties ( ) <EOL> props = btype . get_properties ( ) <EOL> self . assertIsInstance ( props , dict ) <EOL> self . assertIn ( '<STR_LIT>' , props ) <EOL> self . assertEqual ( <NUM_LIT:3> , props [ '<STR_LIT>' ] ) <EOL> def test_btype_set_props ( self ) : <EOL> defbtype = self . client . bucket_type ( "<STR_LIT:default>" ) <EOL> btype = self . client . bucket_type ( '<STR_LIT>' ) <EOL> with self . assertRaises ( ValueError ) : <EOL> defbtype . set_properties ( { '<STR_LIT>' : True } ) <EOL> oldprops = btype . get_properties ( ) <EOL> try : <EOL> btype . set_properties ( { '<STR_LIT>' : True } ) <EOL> newprops = btype . get_properties ( ) <EOL> self . assertIsInstance ( newprops , dict ) <EOL> self . assertIn ( '<STR_LIT>' , newprops ) <EOL> self . assertTrue ( newprops [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in oldprops : <EOL> del oldprops [ '<STR_LIT>' ] <EOL> finally : <EOL> btype . set_properties ( oldprops ) <EOL> def test_btype_set_props_immutable ( self ) : <EOL> btype = self . client . bucket_type ( "<STR_LIT>" ) <EOL> with self . assertRaises ( RiakError ) : <EOL> btype . set_property ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_btype_list_buckets ( self ) : <EOL> btype = self . client . bucket_type ( '<STR_LIT>' ) <EOL> bucket = btype . bucket ( self . bucket_name ) <EOL> obj = bucket . new ( self . key_name ) <EOL> obj . data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> obj . store ( ) <EOL> self . assertIn ( bucket , btype . get_buckets ( ) ) <EOL> buckets = [ ] <EOL> for nested_buckets in btype . stream_buckets ( ) : <EOL> buckets . extend ( nested_buckets ) <EOL> self . assertIn ( bucket , buckets ) <EOL> def test_btype_list_keys ( self ) : <EOL> btype = self . client . bucket_type ( '<STR_LIT>' ) <EOL> bucket = btype . bucket ( self . bucket_name ) <EOL> obj = bucket . new ( self . key_name ) <EOL> obj . data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> obj . store ( ) <EOL> self . assertIn ( self . key_name , bucket . get_keys ( ) ) <EOL> keys = [ ] <EOL> for keylist in bucket . stream_keys ( ) : <EOL> keys . extend ( keylist ) <EOL> self . assertIn ( self . key_name , keys ) <EOL> def test_default_btype_list_buckets ( self ) : <EOL> default_btype = self . client . bucket_type ( "<STR_LIT:default>" ) <EOL> bucket = default_btype . bucket ( self . bucket_name ) <EOL> obj = bucket . new ( self . key_name ) <EOL> obj . data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> obj . store ( ) <EOL> self . assertIn ( bucket , default_btype . get_buckets ( ) ) <EOL> buckets = [ ] <EOL> for nested_buckets in default_btype . stream_buckets ( ) : <EOL> buckets . extend ( nested_buckets ) <EOL> self . assertIn ( bucket , buckets ) <EOL> self . assertItemsEqual ( buckets , self . client . get_buckets ( ) ) <EOL> def test_default_btype_list_keys ( self ) : <EOL> btype = self . client . bucket_type ( "<STR_LIT:default>" ) <EOL> bucket = btype . bucket ( self . bucket_name ) <EOL> obj = bucket . new ( self . key_name ) <EOL> obj . data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> obj . store ( ) <EOL> self . assertIn ( self . key_name , bucket . get_keys ( ) ) <EOL> keys = [ ] <EOL> for keylist in bucket . stream_keys ( ) : <EOL> keys . extend ( keylist ) <EOL> self . assertIn ( self . key_name , keys ) <EOL> oldapikeys = self . client . get_keys ( self . client . bucket ( self . bucket_name ) ) <EOL> self . assertItemsEqual ( keys , oldapikeys ) <EOL> def test_multiget_bucket_types ( self ) : <EOL> btype = self . client . bucket_type ( '<STR_LIT>' ) <EOL> bucket = btype . bucket ( self . bucket_name ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> obj = bucket . new ( self . key_name + str ( i ) ) <EOL> obj . data = { '<STR_LIT:id>' : i } <EOL> obj . store ( ) <EOL> mget = bucket . multiget ( [ self . key_name + str ( i ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> for mobj in mget : <EOL> self . assertIsInstance ( mobj , RiakObject ) <EOL> self . assertEqual ( bucket , mobj . bucket ) <EOL> self . assertEqual ( btype , mobj . bucket . bucket_type ) <EOL> def test_write_once_bucket_type ( self ) : <EOL> btype = self . client . bucket_type ( '<STR_LIT>' ) <EOL> bucket = btype . bucket ( self . bucket_name ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> obj = bucket . new ( self . key_name + str ( i ) ) <EOL> obj . data = { '<STR_LIT:id>' : i } <EOL> obj . store ( ) <EOL> mget = bucket . multiget ( [ self . key_name + str ( i ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> for mobj in mget : <EOL> self . assertIsInstance ( mobj , RiakObject ) <EOL> self . assertEqual ( bucket , mobj . bucket ) <EOL> self . assertEqual ( btype , mobj . bucket . bucket_type ) <EOL> props = btype . get_properties ( ) <EOL> self . assertIn ( '<STR_LIT>' , props ) <EOL> self . assertEqual ( True , props [ '<STR_LIT>' ] ) </s>
