<s> import json <EOL> import os . path <EOL> import webob <EOL> from lxml import etree <EOL> from nova import context <EOL> from nova import test <EOL> from nova . api import openstack <EOL> from nova . api . openstack import extensions <EOL> from nova . api . openstack import flavors <EOL> from nova . api . openstack import wsgi <EOL> from nova . api . openstack import xmlutil <EOL> from nova . tests . api . openstack import fakes <EOL> NS = "<STR_LIT>" <EOL> ATOMNS = "<STR_LIT>" <EOL> response_body = "<STR_LIT>" <EOL> class StubController ( object ) : <EOL> def __init__ ( self , body ) : <EOL> self . body = body <EOL> def index ( self , req ) : <EOL> return self . body <EOL> class StubExtensionManager ( object ) : <EOL> def __init__ ( self , resource_ext = None , action_ext = None , request_ext = None ) : <EOL> self . resource_ext = resource_ext <EOL> self . action_ext = action_ext <EOL> self . request_ext = request_ext <EOL> def get_name ( self ) : <EOL> return "<STR_LIT>" <EOL> def get_alias ( self ) : <EOL> return "<STR_LIT>" <EOL> def get_description ( self ) : <EOL> return "<STR_LIT>" <EOL> def get_resources ( self ) : <EOL> resource_exts = [ ] <EOL> if self . resource_ext : <EOL> resource_exts . append ( self . resource_ext ) <EOL> return resource_exts <EOL> def get_actions ( self ) : <EOL> action_exts = [ ] <EOL> if self . action_ext : <EOL> action_exts . append ( self . action_ext ) <EOL> return action_exts <EOL> def get_request_extensions ( self ) : <EOL> request_extensions = [ ] <EOL> if self . request_ext : <EOL> request_extensions . append ( self . request_ext ) <EOL> return request_extensions <EOL> class ExtensionControllerTest ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ExtensionControllerTest , self ) . setUp ( ) <EOL> ext_path = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> self . flags ( osapi_extensions_path = ext_path ) <EOL> self . ext_list = [ <EOL> "<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>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> self . ext_list . sort ( ) <EOL> def test_list_extensions_json ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> data = json . loads ( response . body ) <EOL> names = [ x [ '<STR_LIT:name>' ] for x in data [ '<STR_LIT>' ] ] <EOL> names . sort ( ) <EOL> self . assertEqual ( names , self . ext_list ) <EOL> ( fox_ext , ) = [ <EOL> x for x in data [ '<STR_LIT>' ] if x [ '<STR_LIT>' ] == '<STR_LIT>' ] <EOL> self . assertEqual ( fox_ext , { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ ] <EOL> } , <EOL> ) <EOL> def test_get_extension_json ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> data = json . loads ( response . body ) <EOL> self . assertEqual ( data [ '<STR_LIT>' ] , { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ ] } ) <EOL> def test_get_non_existing_extension_json ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_int ) <EOL> def test_list_extensions_xml ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> request . accept = "<STR_LIT>" <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> print response . body <EOL> root = etree . XML ( response . body ) <EOL> self . assertEqual ( root . tag . split ( '<STR_LIT>' ) [ <NUM_LIT:0> ] , NS ) <EOL> exts = root . findall ( '<STR_LIT>' . format ( NS ) ) <EOL> self . assertEqual ( len ( exts ) , len ( self . ext_list ) ) <EOL> ( fox_ext , ) = [ x for x in exts if x . get ( '<STR_LIT>' ) == '<STR_LIT>' ] <EOL> self . assertEqual ( fox_ext . get ( '<STR_LIT:name>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( fox_ext . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( fox_ext . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( fox_ext . findtext ( '<STR_LIT>' . format ( NS ) ) , <EOL> '<STR_LIT>' ) <EOL> xmlutil . validate_schema ( root , '<STR_LIT>' ) <EOL> def test_get_extension_xml ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> request . accept = "<STR_LIT>" <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> xml = response . body <EOL> print xml <EOL> root = etree . XML ( xml ) <EOL> self . assertEqual ( root . tag . split ( '<STR_LIT>' ) [ <NUM_LIT:0> ] , NS ) <EOL> self . assertEqual ( root . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( root . get ( '<STR_LIT:name>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( root . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( root . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> self . assertEqual ( root . findtext ( '<STR_LIT>' . format ( NS ) ) , <EOL> '<STR_LIT>' ) <EOL> xmlutil . validate_schema ( root , '<STR_LIT>' ) <EOL> class ResourceExtensionTest ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ResourceExtensionTest , self ) . setUp ( ) <EOL> ext_path = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> self . flags ( osapi_extensions_path = ext_path ) <EOL> def test_no_extension_present ( self ) : <EOL> manager = StubExtensionManager ( None ) <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app , manager ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_int ) <EOL> def test_get_resources ( self ) : <EOL> res_ext = extensions . ResourceExtension ( '<STR_LIT>' , <EOL> StubController ( response_body ) ) <EOL> manager = StubExtensionManager ( res_ext ) <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app , manager ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> self . assertEqual ( response_body , response . body ) <EOL> def test_get_resources_with_controller ( self ) : <EOL> res_ext = extensions . ResourceExtension ( '<STR_LIT>' , <EOL> StubController ( response_body ) ) <EOL> manager = StubExtensionManager ( res_ext ) <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app , manager ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> self . assertEqual ( response_body , response . body ) <EOL> class InvalidExtension ( object ) : <EOL> def get_alias ( self ) : <EOL> return "<STR_LIT>" <EOL> class ExtensionManagerTest ( test . TestCase ) : <EOL> response_body = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ExtensionManagerTest , self ) . setUp ( ) <EOL> ext_path = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> self . flags ( osapi_extensions_path = ext_path ) <EOL> def test_get_resources ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> self . assertEqual ( response_body , response . body ) <EOL> def test_invalid_extensions ( self ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> ext_mgr = ext_midware . ext_mgr <EOL> ext_mgr . add_extension ( InvalidExtension ( ) ) <EOL> self . assertTrue ( '<STR_LIT>' in ext_mgr . extensions ) <EOL> self . assertTrue ( '<STR_LIT>' not in ext_mgr . extensions ) <EOL> class ActionExtensionTest ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ActionExtensionTest , self ) . setUp ( ) <EOL> ext_path = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> self . flags ( osapi_extensions_path = ext_path ) <EOL> def _send_server_action_request ( self , url , body ) : <EOL> app = openstack . APIRouterV11 ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( url ) <EOL> request . method = '<STR_LIT:POST>' <EOL> request . content_type = '<STR_LIT:application/json>' <EOL> request . body = json . dumps ( body ) <EOL> response = request . get_response ( ext_midware ) <EOL> return response <EOL> def test_extended_action ( self ) : <EOL> body = dict ( add_tweedle = dict ( name = "<STR_LIT:test>" ) ) <EOL> url = "<STR_LIT>" <EOL> response = self . _send_server_action_request ( url , body ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> self . assertEqual ( "<STR_LIT>" , response . body ) <EOL> body = dict ( delete_tweedle = dict ( name = "<STR_LIT:test>" ) ) <EOL> response = self . _send_server_action_request ( url , body ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> self . assertEqual ( "<STR_LIT>" , response . body ) <EOL> def test_invalid_action_body ( self ) : <EOL> body = dict ( blah = dict ( name = "<STR_LIT:test>" ) ) <EOL> url = "<STR_LIT>" <EOL> response = self . _send_server_action_request ( url , body ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_int ) <EOL> def test_invalid_action ( self ) : <EOL> body = dict ( blah = dict ( name = "<STR_LIT:test>" ) ) <EOL> url = "<STR_LIT>" <EOL> response = self . _send_server_action_request ( url , body ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_int ) <EOL> class RequestExtensionTest ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( RequestExtensionTest , self ) . setUp ( ) <EOL> ext_path = os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) <EOL> self . flags ( osapi_extensions_path = ext_path ) <EOL> def test_get_resources_with_stub_mgr ( self ) : <EOL> def _req_handler ( req , res ) : <EOL> data = json . loads ( res . body ) <EOL> data [ '<STR_LIT>' ] [ '<STR_LIT>' ] = req . GET . get ( '<STR_LIT>' ) <EOL> res . body = json . dumps ( data ) <EOL> return res <EOL> req_ext = extensions . RequestExtension ( '<STR_LIT:GET>' , <EOL> '<STR_LIT>' , <EOL> _req_handler ) <EOL> manager = StubExtensionManager ( None , None , req_ext ) <EOL> app = fakes . wsgi_app ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app , manager ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> request . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> response_data = json . loads ( response . body ) <EOL> self . assertEqual ( '<STR_LIT>' , response_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> def test_get_resources_with_mgr ( self ) : <EOL> app = fakes . wsgi_app ( ) <EOL> ext_midware = extensions . ExtensionMiddleware ( app ) <EOL> request = webob . Request . blank ( "<STR_LIT>" ) <EOL> request . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response = request . get_response ( ext_midware ) <EOL> self . assertEqual ( <NUM_LIT:200> , response . status_int ) <EOL> response_data = json . loads ( response . body ) <EOL> self . assertEqual ( '<STR_LIT>' , response_data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , response_data [ '<STR_LIT>' ] ) <EOL> class ExtensionsXMLSerializerTest ( test . TestCase ) : <EOL> def test_serialize_extenstion ( self ) : <EOL> serializer = extensions . ExtensionsXMLSerializer ( ) <EOL> data = { '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } } <EOL> xml = serializer . serialize ( data , '<STR_LIT>' ) <EOL> print xml <EOL> root = etree . XML ( xml ) <EOL> ext_dict = data [ '<STR_LIT>' ] <EOL> self . assertEqual ( root . findtext ( '<STR_LIT>' . format ( NS ) ) , <EOL> ext_dict [ '<STR_LIT:description>' ] ) <EOL> for key in [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . assertEqual ( root . get ( key ) , ext_dict [ key ] ) <EOL> link_nodes = root . findall ( '<STR_LIT>' . format ( ATOMNS ) ) <EOL> self . assertEqual ( len ( link_nodes ) , <NUM_LIT:2> ) <EOL> for i , link in enumerate ( ext_dict [ '<STR_LIT>' ] ) : <EOL> for key , value in link . items ( ) : <EOL> self . assertEqual ( link_nodes [ i ] . get ( key ) , value ) <EOL> xmlutil . validate_schema ( root , '<STR_LIT>' ) <EOL> def test_serialize_extensions ( self ) : <EOL> serializer = extensions . ExtensionsXMLSerializer ( ) <EOL> data = { "<STR_LIT>" : [ { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] } , <EOL> { "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] } ] } <EOL> xml = serializer . serialize ( data , '<STR_LIT:index>' ) <EOL> print xml <EOL> root = etree . XML ( xml ) <EOL> ext_elems = root . findall ( '<STR_LIT>' . format ( NS ) ) <EOL> self . assertEqual ( len ( ext_elems ) , <NUM_LIT:2> ) <EOL> for i , ext_elem in enumerate ( ext_elems ) : <EOL> ext_dict = data [ '<STR_LIT>' ] [ i ] <EOL> self . assertEqual ( ext_elem . findtext ( '<STR_LIT>' . format ( NS ) ) , <EOL> ext_dict [ '<STR_LIT:description>' ] ) <EOL> for key in [ '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . assertEqual ( ext_elem . get ( key ) , ext_dict [ key ] ) <EOL> link_nodes = ext_elem . findall ( '<STR_LIT>' . format ( ATOMNS ) ) <EOL> self . assertEqual ( len ( link_nodes ) , <NUM_LIT:2> ) <EOL> for i , link in enumerate ( ext_dict [ '<STR_LIT>' ] ) : <EOL> for key , value in link . items ( ) : <EOL> self . assertEqual ( link_nodes [ i ] . get ( key ) , value ) <EOL> xmlutil . validate_schema ( root , '<STR_LIT>' ) </s>
