<s> """<STR_LIT>""" <EOL> import StringIO <EOL> import hashlib <EOL> import httplib <EOL> import sys <EOL> import unittest <EOL> import urlparse <EOL> import stubout <EOL> import boto . s3 . connection <EOL> from glance . common import exception <EOL> from glance . common import utils <EOL> from glance . store import BackendException , UnsupportedBackend <EOL> from glance . store . location import get_location_from_uri <EOL> from glance . store . s3 import Store <EOL> from glance . tests import utils as test_utils <EOL> FAKE_UUID = utils . generate_uuid ( ) <EOL> FIVE_KB = ( <NUM_LIT:5> * <NUM_LIT> ) <EOL> S3_CONF = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT:user>' , <EOL> '<STR_LIT>' : '<STR_LIT:key>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def stub_out_s3 ( stubs ) : <EOL> class FakeKey : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , bucket , name ) : <EOL> self . bucket = bucket <EOL> self . name = name <EOL> self . data = None <EOL> self . size = <NUM_LIT:0> <EOL> self . BufferSize = <NUM_LIT> <EOL> def close ( self ) : <EOL> pass <EOL> def exists ( self ) : <EOL> return self . bucket . exists ( self . name ) <EOL> def delete ( self ) : <EOL> self . bucket . delete ( self . name ) <EOL> def compute_md5 ( self , data ) : <EOL> chunk = data . read ( self . BufferSize ) <EOL> checksum = hashlib . md5 ( ) <EOL> while chunk : <EOL> checksum . update ( chunk ) <EOL> chunk = data . read ( self . BufferSize ) <EOL> checksum_hex = checksum . hexdigest ( ) <EOL> return checksum_hex , None <EOL> def set_contents_from_file ( self , fp , replace = False , ** kwargs ) : <EOL> self . data = StringIO . StringIO ( ) <EOL> for bytes in fp : <EOL> self . data . write ( bytes ) <EOL> self . size = self . data . len <EOL> self . data . seek ( <NUM_LIT:0> ) <EOL> self . read = self . data . read <EOL> def get_file ( self ) : <EOL> return self . data <EOL> class FakeBucket : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , keys = None ) : <EOL> self . name = name <EOL> self . keys = keys or { } <EOL> def __str__ ( self ) : <EOL> return self . name <EOL> def exists ( self , key ) : <EOL> return key in self . keys <EOL> def delete ( self , key ) : <EOL> del self . keys [ key ] <EOL> def get_key ( self , key_name , ** kwargs ) : <EOL> key = self . keys . get ( key_name ) <EOL> if not key : <EOL> return FakeKey ( self , key_name ) <EOL> return key <EOL> def new_key ( self , key_name ) : <EOL> new_key = FakeKey ( self , key_name ) <EOL> self . keys [ key_name ] = new_key <EOL> return new_key <EOL> fixture_buckets = { '<STR_LIT>' : FakeBucket ( '<STR_LIT>' ) } <EOL> b = fixture_buckets [ '<STR_LIT>' ] <EOL> k = b . new_key ( FAKE_UUID ) <EOL> k . set_contents_from_file ( StringIO . StringIO ( "<STR_LIT:*>" * FIVE_KB ) ) <EOL> def fake_connection_constructor ( self , * args , ** kwargs ) : <EOL> host = kwargs . get ( '<STR_LIT:host>' ) <EOL> if host . startswith ( '<STR_LIT>' ) or host . startswith ( '<STR_LIT>' ) : <EOL> raise UnsupportedBackend ( host ) <EOL> def fake_get_bucket ( conn , bucket_id ) : <EOL> bucket = fixture_buckets . get ( bucket_id ) <EOL> if not bucket : <EOL> bucket = FakeBucket ( bucket_id ) <EOL> return bucket <EOL> stubs . Set ( boto . s3 . connection . S3Connection , <EOL> '<STR_LIT>' , fake_connection_constructor ) <EOL> stubs . Set ( boto . s3 . connection . S3Connection , <EOL> '<STR_LIT>' , fake_get_bucket ) <EOL> def format_s3_location ( user , key , authurl , bucket , obj ) : <EOL> """<STR_LIT>""" <EOL> scheme = '<STR_LIT>' <EOL> if authurl . startswith ( '<STR_LIT>' ) : <EOL> scheme = '<STR_LIT>' <EOL> authurl = authurl [ <NUM_LIT:8> : ] <EOL> elif authurl . startswith ( '<STR_LIT>' ) : <EOL> authurl = authurl [ <NUM_LIT:7> : ] <EOL> authurl = authurl . strip ( '<STR_LIT:/>' ) <EOL> return "<STR_LIT>" % ( scheme , user , key , authurl , <EOL> bucket , obj ) <EOL> class TestStore ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stubs = stubout . StubOutForTesting ( ) <EOL> stub_out_s3 ( self . stubs ) <EOL> self . store = Store ( test_utils . TestConfigOpts ( S3_CONF ) ) <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stubs . UnsetAll ( ) <EOL> def test_get ( self ) : <EOL> """<STR_LIT>""" <EOL> loc = get_location_from_uri ( <EOL> "<STR_LIT>" % FAKE_UUID ) <EOL> ( image_s3 , image_size ) = self . store . get ( loc ) <EOL> self . assertEqual ( image_size , FIVE_KB ) <EOL> expected_data = "<STR_LIT:*>" * FIVE_KB <EOL> data = "<STR_LIT>" <EOL> for chunk in image_s3 : <EOL> data += chunk <EOL> self . assertEqual ( expected_data , data ) <EOL> def test_get_non_existing ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = "<STR_LIT>" % FAKE_UUID <EOL> loc = get_location_from_uri ( uri ) <EOL> self . assertRaises ( exception . NotFound , self . store . get , loc ) <EOL> uri = "<STR_LIT>" <EOL> loc = get_location_from_uri ( uri ) <EOL> self . assertRaises ( exception . NotFound , self . store . get , loc ) <EOL> def test_add ( self ) : <EOL> """<STR_LIT>""" <EOL> expected_image_id = utils . generate_uuid ( ) <EOL> expected_s3_size = FIVE_KB <EOL> expected_s3_contents = "<STR_LIT:*>" * expected_s3_size <EOL> expected_checksum = hashlib . md5 ( expected_s3_contents ) . hexdigest ( ) <EOL> expected_location = format_s3_location ( <EOL> S3_CONF [ '<STR_LIT>' ] , <EOL> S3_CONF [ '<STR_LIT>' ] , <EOL> S3_CONF [ '<STR_LIT>' ] , <EOL> S3_CONF [ '<STR_LIT>' ] , <EOL> expected_image_id ) <EOL> image_s3 = StringIO . StringIO ( expected_s3_contents ) <EOL> location , size , checksum = self . store . add ( expected_image_id , <EOL> image_s3 , <EOL> expected_s3_size ) <EOL> self . assertEquals ( expected_location , location ) <EOL> self . assertEquals ( expected_s3_size , size ) <EOL> self . assertEquals ( expected_checksum , checksum ) <EOL> loc = get_location_from_uri ( expected_location ) <EOL> ( new_image_s3 , new_image_size ) = self . store . get ( loc ) <EOL> new_image_contents = StringIO . StringIO ( ) <EOL> for chunk in new_image_s3 : <EOL> new_image_contents . write ( chunk ) <EOL> new_image_s3_size = new_image_contents . len <EOL> self . assertEquals ( expected_s3_contents , new_image_contents . getvalue ( ) ) <EOL> self . assertEquals ( expected_s3_size , new_image_s3_size ) <EOL> def test_add_host_variations ( self ) : <EOL> """<STR_LIT>""" <EOL> variations = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:localhost>' , <EOL> '<STR_LIT>' ] <EOL> for variation in variations : <EOL> expected_image_id = utils . generate_uuid ( ) <EOL> expected_s3_size = FIVE_KB <EOL> expected_s3_contents = "<STR_LIT:*>" * expected_s3_size <EOL> expected_checksum = hashlib . md5 ( expected_s3_contents ) . hexdigest ( ) <EOL> new_conf = S3_CONF . copy ( ) <EOL> new_conf [ '<STR_LIT>' ] = variation <EOL> expected_location = format_s3_location ( <EOL> new_conf [ '<STR_LIT>' ] , <EOL> new_conf [ '<STR_LIT>' ] , <EOL> new_conf [ '<STR_LIT>' ] , <EOL> new_conf [ '<STR_LIT>' ] , <EOL> expected_image_id ) <EOL> image_s3 = StringIO . StringIO ( expected_s3_contents ) <EOL> self . store = Store ( test_utils . TestConfigOpts ( new_conf ) ) <EOL> location , size , checksum = self . store . add ( expected_image_id , <EOL> image_s3 , <EOL> expected_s3_size ) <EOL> self . assertEquals ( expected_location , location ) <EOL> self . assertEquals ( expected_s3_size , size ) <EOL> self . assertEquals ( expected_checksum , checksum ) <EOL> loc = get_location_from_uri ( expected_location ) <EOL> ( new_image_s3 , new_image_size ) = self . store . get ( loc ) <EOL> new_image_contents = new_image_s3 . getvalue ( ) <EOL> new_image_s3_size = new_image_s3 . len <EOL> self . assertEquals ( expected_s3_contents , new_image_contents ) <EOL> self . assertEquals ( expected_s3_size , new_image_s3_size ) <EOL> def test_add_already_existing ( self ) : <EOL> """<STR_LIT>""" <EOL> image_s3 = StringIO . StringIO ( "<STR_LIT>" ) <EOL> self . assertRaises ( exception . Duplicate , <EOL> self . store . add , <EOL> FAKE_UUID , image_s3 , <NUM_LIT:0> ) <EOL> def _option_required ( self , key ) : <EOL> conf = S3_CONF . copy ( ) <EOL> del conf [ key ] <EOL> try : <EOL> self . store = Store ( test_utils . TestConfigOpts ( conf ) ) <EOL> return self . store . add == self . store . add_disabled <EOL> except : <EOL> return False <EOL> return False <EOL> def test_no_access_key ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( self . _option_required ( '<STR_LIT>' ) ) <EOL> def test_no_secret_key ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( self . _option_required ( '<STR_LIT>' ) ) <EOL> def test_no_host ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( self . _option_required ( '<STR_LIT>' ) ) <EOL> def test_delete ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = "<STR_LIT>" % FAKE_UUID <EOL> loc = get_location_from_uri ( uri ) <EOL> self . store . delete ( loc ) <EOL> self . assertRaises ( exception . NotFound , self . store . get , loc ) <EOL> def test_delete_non_existing ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = "<STR_LIT>" <EOL> loc = get_location_from_uri ( uri ) <EOL> self . assertRaises ( exception . NotFound , self . store . delete , loc ) </s>
