<s> import os . path <EOL> import time <EOL> import webob <EOL> import webob . dec <EOL> import webob . exc <EOL> from neutron . agent . linux import dhcp <EOL> from neutron . agent . linux import utils <EOL> from neutron . tests . common import machine_fixtures <EOL> from neutron . tests . common import net_helpers <EOL> from neutron . tests . functional . agent . l3 import framework <EOL> from neutron . tests . functional . agent . linux import helpers <EOL> METADATA_REQUEST_TIMEOUT = <NUM_LIT> <EOL> METADATA_REQUEST_SLEEP = <NUM_LIT:5> <EOL> class MetadataFakeProxyHandler ( object ) : <EOL> def __init__ ( self , status ) : <EOL> self . status = status <EOL> @ webob . dec . wsgify ( ) <EOL> def __call__ ( self , req ) : <EOL> return webob . Response ( status = self . status ) <EOL> class MetadataL3AgentTestCase ( framework . L3AgentTestFramework ) : <EOL> SOCKET_MODE = <NUM_LIT> <EOL> def _create_metadata_fake_server ( self , status ) : <EOL> server = utils . UnixDomainWSGIServer ( '<STR_LIT>' ) <EOL> self . addCleanup ( server . stop ) <EOL> self . useFixture ( <EOL> helpers . RecursivePermDirFixture ( <EOL> os . path . dirname ( self . agent . conf . metadata_proxy_socket ) , <NUM_LIT> ) ) <EOL> server . start ( MetadataFakeProxyHandler ( status ) , <EOL> self . agent . conf . metadata_proxy_socket , <EOL> workers = <NUM_LIT:0> , backlog = <NUM_LIT> , mode = self . SOCKET_MODE ) <EOL> def _query_metadata_proxy ( self , machine ) : <EOL> url = '<STR_LIT>' % { '<STR_LIT:host>' : dhcp . METADATA_DEFAULT_IP , <EOL> '<STR_LIT:port>' : dhcp . METADATA_PORT } <EOL> cmd = '<STR_LIT>' , '<STR_LIT>' , METADATA_REQUEST_TIMEOUT , '<STR_LIT>' , url <EOL> i = <NUM_LIT:0> <EOL> CONNECTION_REFUSED_TIMEOUT = METADATA_REQUEST_TIMEOUT // <NUM_LIT:2> <EOL> while i <= CONNECTION_REFUSED_TIMEOUT : <EOL> try : <EOL> raw_headers = machine . execute ( cmd ) <EOL> break <EOL> except RuntimeError as e : <EOL> if '<STR_LIT>' in str ( e ) : <EOL> time . sleep ( METADATA_REQUEST_SLEEP ) <EOL> i += METADATA_REQUEST_SLEEP <EOL> else : <EOL> self . fail ( '<STR_LIT>' <EOL> '<STR_LIT>' % url ) <EOL> if i > CONNECTION_REFUSED_TIMEOUT : <EOL> self . fail ( '<STR_LIT>' ) <EOL> return raw_headers . splitlines ( ) [ <NUM_LIT:0> ] <EOL> def test_access_to_metadata_proxy ( self ) : <EOL> """<STR_LIT>""" <EOL> router_info = self . generate_router_info ( enable_ha = False ) <EOL> router = self . manage_router ( self . agent , router_info ) <EOL> self . _create_metadata_fake_server ( webob . exc . HTTPOk . code ) <EOL> router_ip_cidr = self . _port_first_ip_cidr ( router . internal_ports [ <NUM_LIT:0> ] ) <EOL> br_int = framework . get_ovs_bridge ( <EOL> self . agent . conf . ovs_integration_bridge ) <EOL> machine = self . useFixture ( <EOL> machine_fixtures . FakeMachine ( <EOL> br_int , <EOL> net_helpers . increment_ip_cidr ( router_ip_cidr ) , <EOL> router_ip_cidr . partition ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] ) ) <EOL> firstline = self . _query_metadata_proxy ( machine ) <EOL> self . assertIn ( str ( webob . exc . HTTPOk . code ) , firstline . split ( ) ) <EOL> class UnprivilegedUserMetadataL3AgentTestCase ( MetadataL3AgentTestCase ) : <EOL> """<STR_LIT>""" <EOL> SOCKET_MODE = <NUM_LIT> <EOL> def setUp ( self ) : <EOL> super ( UnprivilegedUserMetadataL3AgentTestCase , self ) . setUp ( ) <EOL> self . agent . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . agent . conf . set_override ( '<STR_LIT>' , False ) <EOL> class UnprivilegedUserGroupMetadataL3AgentTestCase ( MetadataL3AgentTestCase ) : <EOL> """<STR_LIT>""" <EOL> SOCKET_MODE = <NUM_LIT> <EOL> def setUp ( self ) : <EOL> super ( UnprivilegedUserGroupMetadataL3AgentTestCase , self ) . setUp ( ) <EOL> self . agent . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . agent . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . agent . conf . set_override ( '<STR_LIT>' , False ) </s>
