<s> """<STR_LIT>""" <EOL> import hashlib <EOL> import json <EOL> import os <EOL> import shutil <EOL> import thread <EOL> import time <EOL> import BaseHTTPServer <EOL> import httplib2 <EOL> from glance . tests import functional <EOL> from glance . tests . utils import ( skip_if_disabled , <EOL> execute , <EOL> xattr_writes_supported ) <EOL> FIVE_KB = <NUM_LIT:5> * <NUM_LIT> <EOL> class RemoteImageHandler ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> def do_HEAD ( s ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in s . path : <EOL> s . send_response ( <NUM_LIT:200> ) <EOL> s . send_header ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> s . send_header ( '<STR_LIT>' , FIVE_KB ) <EOL> s . end_headers ( ) <EOL> return <EOL> else : <EOL> self . send_error ( <NUM_LIT> , '<STR_LIT>' % self . path ) <EOL> return <EOL> def do_GET ( s ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in s . path : <EOL> s . send_response ( <NUM_LIT:200> ) <EOL> s . send_header ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) <EOL> s . send_header ( '<STR_LIT>' , FIVE_KB ) <EOL> s . end_headers ( ) <EOL> image_data = '<STR_LIT:*>' * FIVE_KB <EOL> s . wfile . write ( image_data ) <EOL> self . wfile . close ( ) <EOL> return <EOL> else : <EOL> self . send_error ( <NUM_LIT> , '<STR_LIT>' % self . path ) <EOL> return <EOL> class BaseCacheMiddlewareTest ( object ) : <EOL> @ skip_if_disabled <EOL> def test_cache_middleware_transparent ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup ( ) <EOL> self . start_servers ( ** self . __dict__ . copy ( ) ) <EOL> api_port = self . api_port <EOL> registry_port = self . registry_port <EOL> image_data = "<STR_LIT:*>" * FIVE_KB <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:True>' } <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:POST>' , headers = headers , <EOL> body = image_data ) <EOL> self . assertEqual ( response . status , <NUM_LIT> ) <EOL> data = json . loads ( content ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT>' ] , <EOL> hashlib . md5 ( image_data ) . hexdigest ( ) ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT:size>' ] , FIVE_KB ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT:name>' ] , "<STR_LIT>" ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT>' ] , True ) <EOL> image_id = data [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] <EOL> image_cached_path = os . path . join ( self . api_server . image_cache_dir , <EOL> image_id ) <EOL> self . assertFalse ( os . path . exists ( image_cached_path ) ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> image_id ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> image_cached_path = os . path . join ( self . api_server . image_cache_dir , <EOL> image_id ) <EOL> i = <NUM_LIT:0> <EOL> while not os . path . exists ( image_cached_path ) and i < <NUM_LIT:30> : <EOL> time . sleep ( <NUM_LIT> ) <EOL> i = i + <NUM_LIT:1> <EOL> self . assertTrue ( os . path . exists ( image_cached_path ) ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> image_id ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> self . assertFalse ( os . path . exists ( image_cached_path ) ) <EOL> self . stop_servers ( ) <EOL> @ skip_if_disabled <EOL> def test_cache_remote_image ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup ( ) <EOL> self . start_servers ( ** self . __dict__ . copy ( ) ) <EOL> api_port = self . api_port <EOL> registry_port = self . registry_port <EOL> server_class = BaseHTTPServer . HTTPServer <EOL> remote_server = server_class ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , RemoteImageHandler ) <EOL> remote_ip , remote_port = remote_server . server_address <EOL> def serve_requests ( httpd ) : <EOL> httpd . serve_forever ( ) <EOL> thread . start_new_thread ( serve_requests , ( remote_server , ) ) <EOL> remote_uri = '<STR_LIT>' % ( remote_ip , remote_port ) <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:True>' , <EOL> '<STR_LIT>' : remote_uri } <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:POST>' , headers = headers ) <EOL> self . assertEqual ( response . status , <NUM_LIT> ) <EOL> data = json . loads ( content ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT:size>' ] , FIVE_KB ) <EOL> image_id = data [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> image_id ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> self . assertEqual ( int ( response [ '<STR_LIT>' ] ) , FIVE_KB ) <EOL> remote_server . shutdown ( ) <EOL> self . stop_servers ( ) <EOL> class BaseCacheManageMiddlewareTest ( object ) : <EOL> """<STR_LIT>""" <EOL> def verify_no_images ( self ) : <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( data [ '<STR_LIT>' ] ) ) <EOL> def add_image ( self , name ) : <EOL> """<STR_LIT>""" <EOL> image_data = "<STR_LIT:*>" * FIVE_KB <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:%s>' % name , <EOL> '<STR_LIT>' : '<STR_LIT:True>' } <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:POST>' , headers = headers , <EOL> body = image_data ) <EOL> self . assertEqual ( response . status , <NUM_LIT> ) <EOL> data = json . loads ( content ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT>' ] , <EOL> hashlib . md5 ( image_data ) . hexdigest ( ) ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT:size>' ] , FIVE_KB ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT:name>' ] , name ) <EOL> self . assertEqual ( data [ '<STR_LIT:image>' ] [ '<STR_LIT>' ] , True ) <EOL> return data [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] <EOL> def verify_no_cached_images ( self ) : <EOL> """<STR_LIT>""" <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> self . assertEqual ( data [ '<STR_LIT>' ] , [ ] ) <EOL> @ skip_if_disabled <EOL> def test_cache_manage_get_cached_images ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup ( ) <EOL> self . start_servers ( ** self . __dict__ . copy ( ) ) <EOL> api_port = self . api_port <EOL> registry_port = self . registry_port <EOL> self . verify_no_images ( ) <EOL> image_id = self . add_image ( "<STR_LIT>" ) <EOL> self . verify_no_cached_images ( ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> image_id ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> cached_images = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( cached_images ) ) <EOL> self . assertEqual ( image_id , cached_images [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , cached_images [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> image_id ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> cached_images = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( cached_images ) ) <EOL> self . assertEqual ( image_id , cached_images [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , cached_images [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . stop_servers ( ) <EOL> @ skip_if_disabled <EOL> def test_cache_manage_delete_cached_images ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup ( ) <EOL> self . start_servers ( ** self . __dict__ . copy ( ) ) <EOL> api_port = self . api_port <EOL> registry_port = self . registry_port <EOL> self . verify_no_images ( ) <EOL> ids = { } <EOL> for x in xrange ( <NUM_LIT:0> , <NUM_LIT:4> ) : <EOL> ids [ x ] = self . add_image ( "<STR_LIT>" % str ( x ) ) <EOL> self . verify_no_cached_images ( ) <EOL> for x in xrange ( <NUM_LIT:0> , <NUM_LIT:4> ) : <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> ids [ x ] ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> , <EOL> "<STR_LIT>" % ids [ x ] ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> cached_images = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:4> , len ( cached_images ) ) <EOL> for x in xrange ( <NUM_LIT:4> , <NUM_LIT:0> ) : <EOL> self . assertEqual ( ids [ x ] , cached_images [ x ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , cached_images [ x ] [ '<STR_LIT>' ] ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> ids [ <NUM_LIT:2> ] ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> cached_images = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:3> , len ( cached_images ) ) <EOL> self . assertTrue ( ids [ <NUM_LIT:2> ] not in [ x [ '<STR_LIT>' ] for x in cached_images ] ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> cached_images = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:0> , len ( cached_images ) ) <EOL> self . stop_servers ( ) <EOL> @ skip_if_disabled <EOL> def test_queue_and_prefetch ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cleanup ( ) <EOL> self . start_servers ( ** self . __dict__ . copy ( ) ) <EOL> api_port = self . api_port <EOL> registry_port = self . registry_port <EOL> cache_config_filepath = os . path . join ( self . test_dir , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> cache_file_options = { <EOL> '<STR_LIT>' : self . api_server . image_cache_dir , <EOL> '<STR_LIT>' : self . image_cache_driver , <EOL> '<STR_LIT>' : self . api_server . registry_port , <EOL> '<STR_LIT>' : os . path . join ( self . test_dir , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> } <EOL> with open ( cache_config_filepath , '<STR_LIT:w>' ) as cache_file : <EOL> cache_file . write ( """<STR_LIT>""" % cache_file_options ) <EOL> with open ( cache_config_filepath . replace ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> '<STR_LIT:w>' ) as paste_file : <EOL> paste_file . write ( """<STR_LIT>""" ) <EOL> self . verify_no_images ( ) <EOL> ids = { } <EOL> for x in xrange ( <NUM_LIT:0> , <NUM_LIT:4> ) : <EOL> ids [ x ] = self . add_image ( "<STR_LIT>" % str ( x ) ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port , <EOL> ids [ <NUM_LIT:0> ] ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> self . verify_no_cached_images ( ) <EOL> cmd = "<STR_LIT>" % cache_config_filepath <EOL> exitcode , out , err = execute ( cmd ) <EOL> self . assertEqual ( <NUM_LIT:0> , exitcode ) <EOL> self . assertEqual ( '<STR_LIT>' , out . strip ( ) , out ) <EOL> path = "<STR_LIT>" % ( "<STR_LIT>" , self . api_port ) <EOL> http = httplib2 . Http ( ) <EOL> response , content = http . request ( path , '<STR_LIT:GET>' ) <EOL> self . assertEqual ( response . status , <NUM_LIT:200> ) <EOL> data = json . loads ( content ) <EOL> self . assertTrue ( '<STR_LIT>' in data ) <EOL> cached_images = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( cached_images ) ) <EOL> self . assertTrue ( ids [ <NUM_LIT:0> ] in [ r [ '<STR_LIT>' ] <EOL> for r in data [ '<STR_LIT>' ] ] ) <EOL> self . stop_servers ( ) <EOL> class TestImageCacheXattr ( functional . FunctionalTest , <EOL> BaseCacheMiddlewareTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> if getattr ( self , '<STR_LIT>' , False ) : <EOL> return <EOL> if not getattr ( self , '<STR_LIT>' , False ) : <EOL> try : <EOL> import xattr <EOL> except ImportError : <EOL> self . inited = True <EOL> self . disabled = True <EOL> self . disabled_message = ( "<STR_LIT>" ) <EOL> return <EOL> self . inited = True <EOL> self . disabled = False <EOL> self . image_cache_driver = "<STR_LIT>" <EOL> super ( TestImageCacheXattr , self ) . setUp ( ) <EOL> self . api_server . deployment_flavor = "<STR_LIT>" <EOL> if not xattr_writes_supported ( self . test_dir ) : <EOL> self . inited = True <EOL> self . disabled = True <EOL> self . disabled_message = ( "<STR_LIT>" ) <EOL> return <EOL> def tearDown ( self ) : <EOL> if os . path . exists ( self . api_server . image_cache_dir ) : <EOL> shutil . rmtree ( self . api_server . image_cache_dir ) <EOL> class TestImageCacheManageXattr ( functional . FunctionalTest , <EOL> BaseCacheManageMiddlewareTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> if getattr ( self , '<STR_LIT>' , False ) : <EOL> return <EOL> if not getattr ( self , '<STR_LIT>' , False ) : <EOL> try : <EOL> import xattr <EOL> except ImportError : <EOL> self . inited = True <EOL> self . disabled = True <EOL> self . disabled_message = ( "<STR_LIT>" ) <EOL> return <EOL> self . inited = True <EOL> self . disabled = False <EOL> self . image_cache_driver = "<STR_LIT>" <EOL> super ( TestImageCacheManageXattr , self ) . setUp ( ) <EOL> self . api_server . deployment_flavor = "<STR_LIT>" <EOL> if not xattr_writes_supported ( self . test_dir ) : <EOL> self . inited = True <EOL> self . disabled = True <EOL> self . disabled_message = ( "<STR_LIT>" ) <EOL> return <EOL> def tearDown ( self ) : <EOL> if os . path . exists ( self . api_server . image_cache_dir ) : <EOL> shutil . rmtree ( self . api_server . image_cache_dir ) <EOL> class TestImageCacheSqlite ( functional . FunctionalTest , <EOL> BaseCacheMiddlewareTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> if getattr ( self , '<STR_LIT>' , False ) : <EOL> return <EOL> if not getattr ( self , '<STR_LIT>' , False ) : <EOL> try : <EOL> import sqlite3 <EOL> except ImportError : <EOL> self . inited = True <EOL> self . disabled = True <EOL> self . disabled_message = ( "<STR_LIT>" ) <EOL> return <EOL> self . inited = True <EOL> self . disabled = False <EOL> super ( TestImageCacheSqlite , self ) . setUp ( ) <EOL> self . api_server . deployment_flavor = "<STR_LIT>" <EOL> def tearDown ( self ) : <EOL> if os . path . exists ( self . api_server . image_cache_dir ) : <EOL> shutil . rmtree ( self . api_server . image_cache_dir ) <EOL> class TestImageCacheManageSqlite ( functional . FunctionalTest , <EOL> BaseCacheManageMiddlewareTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> if getattr ( self , '<STR_LIT>' , False ) : <EOL> return <EOL> if not getattr ( self , '<STR_LIT>' , False ) : <EOL> try : <EOL> import sqlite3 <EOL> except ImportError : <EOL> self . inited = True <EOL> self . disabled = True <EOL> self . disabled_message = ( "<STR_LIT>" ) <EOL> return <EOL> self . inited = True <EOL> self . disabled = False <EOL> self . image_cache_driver = "<STR_LIT>" <EOL> super ( TestImageCacheManageSqlite , self ) . setUp ( ) <EOL> self . api_server . deployment_flavor = "<STR_LIT>" <EOL> def tearDown ( self ) : <EOL> if os . path . exists ( self . api_server . image_cache_dir ) : <EOL> shutil . rmtree ( self . api_server . image_cache_dir ) </s>
