<s> import json <EOL> import httplib <EOL> import urlparse <EOL> from nova import log as logging <EOL> LOG = logging . getLogger ( '<STR_LIT>' ) <EOL> class OpenStackApiException ( Exception ) : <EOL> def __init__ ( self , message = None , response = None ) : <EOL> self . response = response <EOL> if not message : <EOL> message = '<STR_LIT>' <EOL> if response : <EOL> _status = response . status <EOL> _body = response . read ( ) <EOL> message = _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % locals ( ) <EOL> super ( OpenStackApiException , self ) . __init__ ( message ) <EOL> class OpenStackApiAuthenticationException ( OpenStackApiException ) : <EOL> def __init__ ( self , response = None , message = None ) : <EOL> if not message : <EOL> message = _ ( "<STR_LIT>" ) <EOL> super ( OpenStackApiAuthenticationException , self ) . __init__ ( message , <EOL> response ) <EOL> class OpenStackApiAuthorizationException ( OpenStackApiException ) : <EOL> def __init__ ( self , response = None , message = None ) : <EOL> if not message : <EOL> message = _ ( "<STR_LIT>" ) <EOL> super ( OpenStackApiAuthorizationException , self ) . __init__ ( message , <EOL> response ) <EOL> class OpenStackApiNotFoundException ( OpenStackApiException ) : <EOL> def __init__ ( self , response = None , message = None ) : <EOL> if not message : <EOL> message = _ ( "<STR_LIT>" ) <EOL> super ( OpenStackApiNotFoundException , self ) . __init__ ( message , response ) <EOL> class TestOpenStackClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , auth_user , auth_key , auth_uri ) : <EOL> super ( TestOpenStackClient , self ) . __init__ ( ) <EOL> self . auth_result = None <EOL> self . auth_user = auth_user <EOL> self . auth_key = auth_key <EOL> self . auth_uri = auth_uri <EOL> self . project_id = '<STR_LIT>' <EOL> def request ( self , url , method = '<STR_LIT:GET>' , body = None , headers = None ) : <EOL> _headers = { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } <EOL> _headers . update ( headers or { } ) <EOL> parsed_url = urlparse . urlparse ( url ) <EOL> port = parsed_url . port <EOL> hostname = parsed_url . hostname <EOL> scheme = parsed_url . scheme <EOL> if scheme == '<STR_LIT:http>' : <EOL> conn = httplib . HTTPConnection ( hostname , <EOL> port = port ) <EOL> elif scheme == '<STR_LIT>' : <EOL> conn = httplib . HTTPSConnection ( hostname , <EOL> port = port ) <EOL> else : <EOL> raise OpenStackApiException ( "<STR_LIT>" % url ) <EOL> relative_url = parsed_url . path <EOL> if parsed_url . query : <EOL> relative_url = relative_url + parsed_url . query <EOL> LOG . info ( _ ( "<STR_LIT>" ) % locals ( ) ) <EOL> if body : <EOL> LOG . info ( _ ( "<STR_LIT>" ) % body ) <EOL> conn . request ( method , relative_url , body , _headers ) <EOL> response = conn . getresponse ( ) <EOL> return response <EOL> def _authenticate ( self ) : <EOL> if self . auth_result : <EOL> return self . auth_result <EOL> auth_uri = self . auth_uri <EOL> headers = { '<STR_LIT>' : self . auth_user , <EOL> '<STR_LIT>' : self . auth_key , <EOL> '<STR_LIT>' : self . project_id } <EOL> response = self . request ( auth_uri , <EOL> headers = headers ) <EOL> http_status = response . status <EOL> LOG . debug ( _ ( "<STR_LIT>" ) % locals ( ) ) <EOL> if http_status == <NUM_LIT> : <EOL> raise OpenStackApiAuthenticationException ( response = response ) <EOL> auth_headers = { } <EOL> for k , v in response . getheaders ( ) : <EOL> auth_headers [ k ] = v <EOL> self . auth_result = auth_headers <EOL> return self . auth_result <EOL> def api_request ( self , relative_uri , check_response_status = None , ** kwargs ) : <EOL> auth_result = self . _authenticate ( ) <EOL> base_uri = auth_result [ '<STR_LIT>' ] <EOL> full_uri = '<STR_LIT>' % ( base_uri , relative_uri ) <EOL> headers = kwargs . setdefault ( '<STR_LIT>' , { } ) <EOL> headers [ '<STR_LIT>' ] = auth_result [ '<STR_LIT>' ] <EOL> response = self . request ( full_uri , ** kwargs ) <EOL> http_status = response . status <EOL> LOG . debug ( _ ( "<STR_LIT>" ) % locals ( ) ) <EOL> if check_response_status : <EOL> if not http_status in check_response_status : <EOL> if http_status == <NUM_LIT> : <EOL> raise OpenStackApiNotFoundException ( response = response ) <EOL> elif http_status == <NUM_LIT> : <EOL> raise OpenStackApiAuthorizationException ( response = response ) <EOL> else : <EOL> raise OpenStackApiException ( <EOL> message = _ ( "<STR_LIT>" ) , <EOL> response = response ) <EOL> return response <EOL> def _decode_json ( self , response ) : <EOL> body = response . read ( ) <EOL> LOG . debug ( _ ( "<STR_LIT>" ) % ( body ) ) <EOL> if body : <EOL> return json . loads ( body ) <EOL> else : <EOL> return "<STR_LIT>" <EOL> def api_get ( self , relative_uri , ** kwargs ) : <EOL> kwargs . setdefault ( '<STR_LIT>' , [ <NUM_LIT:200> ] ) <EOL> response = self . api_request ( relative_uri , ** kwargs ) <EOL> return self . _decode_json ( response ) <EOL> def api_post ( self , relative_uri , body , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:POST>' <EOL> if body : <EOL> headers = kwargs . setdefault ( '<STR_LIT>' , { } ) <EOL> headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT:application/json>' <EOL> kwargs [ '<STR_LIT:body>' ] = json . dumps ( body ) <EOL> kwargs . setdefault ( '<STR_LIT>' , [ <NUM_LIT:200> , <NUM_LIT> ] ) <EOL> response = self . api_request ( relative_uri , ** kwargs ) <EOL> return self . _decode_json ( response ) <EOL> def api_put ( self , relative_uri , body , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if body : <EOL> headers = kwargs . setdefault ( '<STR_LIT>' , { } ) <EOL> headers [ '<STR_LIT:Content-Type>' ] = '<STR_LIT:application/json>' <EOL> kwargs [ '<STR_LIT:body>' ] = json . dumps ( body ) <EOL> kwargs . setdefault ( '<STR_LIT>' , [ <NUM_LIT:200> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> response = self . api_request ( relative_uri , ** kwargs ) <EOL> return self . _decode_json ( response ) <EOL> def api_delete ( self , relative_uri , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> kwargs . setdefault ( '<STR_LIT>' , [ <NUM_LIT:200> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> return self . api_request ( relative_uri , ** kwargs ) <EOL> def get_server ( self , server_id ) : <EOL> return self . api_get ( '<STR_LIT>' % server_id ) [ '<STR_LIT>' ] <EOL> def get_servers ( self , detail = True ) : <EOL> rel_url = '<STR_LIT>' if detail else '<STR_LIT>' <EOL> return self . api_get ( rel_url ) [ '<STR_LIT>' ] <EOL> def post_server ( self , server ) : <EOL> return self . api_post ( '<STR_LIT>' , server ) [ '<STR_LIT>' ] <EOL> def put_server ( self , server_id , server ) : <EOL> return self . api_put ( '<STR_LIT>' % server_id , server ) <EOL> def post_server_action ( self , server_id , data ) : <EOL> return self . api_post ( '<STR_LIT>' % server_id , data ) <EOL> def delete_server ( self , server_id ) : <EOL> return self . api_delete ( '<STR_LIT>' % server_id ) <EOL> def get_image ( self , image_id ) : <EOL> return self . api_get ( '<STR_LIT>' % image_id ) [ '<STR_LIT:image>' ] <EOL> def get_images ( self , detail = True ) : <EOL> rel_url = '<STR_LIT>' if detail else '<STR_LIT>' <EOL> return self . api_get ( rel_url ) [ '<STR_LIT>' ] <EOL> def post_image ( self , image ) : <EOL> return self . api_post ( '<STR_LIT>' , image ) [ '<STR_LIT:image>' ] <EOL> def delete_image ( self , image_id ) : <EOL> return self . api_delete ( '<STR_LIT>' % image_id ) <EOL> def get_flavor ( self , flavor_id ) : <EOL> return self . api_get ( '<STR_LIT>' % flavor_id ) [ '<STR_LIT>' ] <EOL> def get_flavors ( self , detail = True ) : <EOL> rel_url = '<STR_LIT>' if detail else '<STR_LIT>' <EOL> return self . api_get ( rel_url ) [ '<STR_LIT>' ] <EOL> def post_flavor ( self , flavor ) : <EOL> return self . api_post ( '<STR_LIT>' , flavor ) [ '<STR_LIT>' ] <EOL> def delete_flavor ( self , flavor_id ) : <EOL> return self . api_delete ( '<STR_LIT>' % flavor_id ) <EOL> def get_volume ( self , volume_id ) : <EOL> return self . api_get ( '<STR_LIT>' % volume_id ) [ '<STR_LIT>' ] <EOL> def get_volumes ( self , detail = True ) : <EOL> rel_url = '<STR_LIT>' if detail else '<STR_LIT>' <EOL> return self . api_get ( rel_url ) [ '<STR_LIT>' ] <EOL> def post_volume ( self , volume ) : <EOL> return self . api_post ( '<STR_LIT>' , volume ) [ '<STR_LIT>' ] <EOL> def delete_volume ( self , volume_id ) : <EOL> return self . api_delete ( '<STR_LIT>' % volume_id ) <EOL> def get_server_volume ( self , server_id , attachment_id ) : <EOL> return self . api_get ( '<STR_LIT>' % <EOL> ( server_id , attachment_id ) ) [ '<STR_LIT>' ] <EOL> def get_server_volumes ( self , server_id ) : <EOL> return self . api_get ( '<STR_LIT>' % <EOL> ( server_id ) ) [ '<STR_LIT>' ] <EOL> def post_server_volume ( self , server_id , volume_attachment ) : <EOL> return self . api_post ( '<STR_LIT>' % <EOL> ( server_id ) , volume_attachment <EOL> ) [ '<STR_LIT>' ] <EOL> def delete_server_volume ( self , server_id , attachment_id ) : <EOL> return self . api_delete ( '<STR_LIT>' % <EOL> ( server_id , attachment_id ) ) </s>
