<s> """<STR_LIT>""" <EOL> from IPy import IP <EOL> import re <EOL> import time <EOL> from cafe . drivers . unittest . decorators import tags <EOL> from cloudcafe . common . tools . datagen import rand_name <EOL> from cloudroast . networking . networks . fixtures import NetworkingComputeFixture <EOL> from cloudroast . networking . networks . scenario . common import ScenarioMixin <EOL> PING_PACKET_LOSS_REGEX = '<STR_LIT>' <EOL> class L2HostroutesGatewayTest ( NetworkingComputeFixture , <EOL> ScenarioMixin ) : <EOL> """<STR_LIT>""" <EOL> PRIVATE_KEY_PATH = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( L2HostroutesGatewayTest , cls ) . setUpClass ( ) <EOL> cls . servers_client = cls . compute . servers . client <EOL> cls . keypairs_client = cls . compute . keypairs . client <EOL> cls . server_behaviors = cls . compute . servers . behaviors <EOL> cls . subnets_client = cls . net . subnets . client <EOL> cls . networks_behaviors = cls . net . behaviors . networks_behaviors <EOL> cls . subnets_behaviors = cls . net . behaviors . subnets_behaviors <EOL> cls . ports_behaviors = cls . net . behaviors . ports_behaviors <EOL> cls . networks_client = cls . networks_behaviors . client <EOL> def _create_networks ( self ) : <EOL> network , subnet = self . _create_network_with_subnet ( '<STR_LIT>' , <EOL> self . base_cidr ) <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT:default>" ) <EOL> self . assertFalse ( subnet . gateway_ip , msg ) <EOL> self . destination_network = network <EOL> self . destination_subnet = subnet <EOL> next_cidr = self . _next_sequential_cidr ( IP ( self . base_cidr ) ) <EOL> allocation_pools = [ { "<STR_LIT:start>" : str ( IP ( next_cidr [ <NUM_LIT:1> ] . ip ) ) , <EOL> "<STR_LIT:end>" : str ( IP ( next_cidr [ - <NUM_LIT:2> ] . ip ) ) } ] <EOL> network , subnet = self . _create_network_with_subnet ( <EOL> '<STR_LIT>' , str ( next_cidr ) , allocation_pools ) <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> expected_allocation_pools = set ( <EOL> tuple ( x . items ( ) ) for x in allocation_pools ) <EOL> for pool in subnet . allocation_pools : <EOL> self . assertTrue ( tuple ( pool . items ( ) ) in <EOL> expected_allocation_pools , msg ) <EOL> self . network_with_route = network <EOL> self . subnet_with_route = subnet <EOL> def _create_router ( self ) : <EOL> self . router = self . _create_server ( '<STR_LIT>' , [ self . network_with_route , <EOL> self . destination_network ] ) <EOL> resp = self . networks_client . delete_network ( self . network_with_route . id ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> msg = msg . format ( str ( resp . status_code ) ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> , msg ) <EOL> return <EOL> port = self . ports_behaviors . create_port ( <EOL> self . destination_network . id , <EOL> name = '<STR_LIT>' . format ( rand_name ( self . NAMES_PREFIX ) , '<STR_LIT>' ) , <EOL> device_id = self . router . entity . id , <EOL> use_exact_name = True ) . response . entity <EOL> self . delete_ports . append ( port . id ) <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( self . router . entity . id , port . device_id , msg ) <EOL> def _enable_ip_forwarding ( self ) : <EOL> self . router . remote_client = self . _get_remote_client ( self . router ) <EOL> for cmd in self . ENABLE_IP_FORWARDING_CMDS : <EOL> self . _execute_ssh_command ( self . router . remote_client . ssh_client , <EOL> cmd ) <EOL> def _create_communicating_servers ( self ) : <EOL> self . origin = self . _create_server ( '<STR_LIT>' , [ self . network_with_route ] ) <EOL> self . destination = self . _create_server ( '<STR_LIT>' , <EOL> [ self . destination_network ] ) <EOL> self . origin . remote_client = self . _get_remote_client ( self . origin ) <EOL> destination = self . destination_subnet . cidr [ <EOL> : self . destination_subnet . cidr . rindex ( '<STR_LIT:/>' ) ] <EOL> if self . __class__ . __name__ == '<STR_LIT>' : <EOL> nexthop = getattr ( self . router , self . network_with_route . name ) <EOL> self . _execute_ssh_command ( <EOL> self . origin . remote_client . ssh_client , <EOL> '<STR_LIT>' . format ( destination , nexthop ) ) <EOL> route_cmd = '<STR_LIT>' . format ( self . ROUTE_COMMAND , destination ) <EOL> route = self . _execute_ssh_command ( <EOL> self . origin . remote_client . ssh_client , <EOL> route_cmd ) . split ( '<STR_LIT:U+0020>' ) <EOL> msg = "<STR_LIT>" <EOL> self . assertIn ( destination , route [ <NUM_LIT:0> ] , msg ) <EOL> self . assertIn ( <EOL> getattr ( self . router , self . network_with_route . name ) , <EOL> route , msg ) <EOL> self . destination . remote_client = self . _get_remote_client ( <EOL> self . destination ) <EOL> origin = self . subnet_with_route . cidr [ <EOL> : self . subnet_with_route . cidr . rindex ( '<STR_LIT:/>' ) ] <EOL> route_cmd = '<STR_LIT>' . format ( self . ROUTE_COMMAND , origin ) <EOL> route = self . _execute_ssh_command ( <EOL> self . destination . remote_client . ssh_client , <EOL> route_cmd ) <EOL> msg = "<STR_LIT>" <EOL> self . assertFalse ( route , msg ) <EOL> self . _transfer_private_key_to_vm ( self . origin . remote_client . ssh_client , <EOL> self . keypair . private_key , <EOL> self . PRIVATE_KEY_PATH ) <EOL> self . _transfer_private_key_to_vm ( <EOL> self . destination . remote_client . ssh_client , <EOL> self . keypair . private_key , <EOL> self . PRIVATE_KEY_PATH ) <EOL> def _set_host_routes ( self ) : <EOL> self . subnets_client . update_subnet ( <EOL> self . subnet_with_route . id , <EOL> host_routes = [ { "<STR_LIT>" : self . destination_subnet . cidr , <EOL> "<STR_LIT>" : getattr ( self . router , <EOL> self . network_with_route . name ) } ] ) <EOL> def _ping ( self , ssh_client , ip_address ) : <EOL> ping_cmd = self . PING_COMMAND . format ( ip_address ) <EOL> output = ssh_client . execute_command ( ping_cmd ) <EOL> try : <EOL> packet_loss_percent = re . search ( PING_PACKET_LOSS_REGEX , <EOL> output . stdout ) . group ( <NUM_LIT:1> ) <EOL> except Exception : <EOL> return False <EOL> return packet_loss_percent != '<STR_LIT:100>' <EOL> def _ssh ( self , ssh_client , ip_address , cmd ) : <EOL> ssh_cmd = self . SSH_COMMAND . format ( self . PRIVATE_KEY_PATH , ip_address , <EOL> cmd ) <EOL> return ssh_client . execute_command ( ssh_cmd ) <EOL> def _verify_expected_connectivity ( self ) : <EOL> """<STR_LIT>""" <EOL> origin_address = getattr ( self . origin , self . network_with_route . name ) <EOL> destination_address = getattr ( self . destination , <EOL> self . destination_network . name ) <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertTrue ( self . _ping ( <EOL> self . origin . remote_client . ssh_client , destination_address ) , msg ) <EOL> response = self . _ssh ( self . origin . remote_client . ssh_client , <EOL> destination_address , '<STR_LIT>' ) <EOL> self . assertEqual ( response . stdout . strip ( ) , '<STR_LIT>' , msg ) <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertFalse ( self . _ping ( <EOL> self . destination . remote_client . ssh_client , origin_address ) , msg ) <EOL> response = self . _ssh ( self . destination . remote_client . ssh_client , <EOL> origin_address , '<STR_LIT>' ) <EOL> if ( ( '<STR_LIT>' not in response . stderr ) and <EOL> ( '<STR_LIT>' not in response . stderr ) ) : <EOL> self . fail ( msg ) <EOL> def _test_execute ( self ) : <EOL> self . _create_networks ( ) <EOL> self . _create_keypair ( ) <EOL> self . _create_router ( ) <EOL> self . _set_host_routes ( ) <EOL> self . _create_communicating_servers ( ) <EOL> self . _enable_ip_forwarding ( ) <EOL> self . _verify_expected_connectivity ( ) <EOL> class L2HostroutesGatewayTestIPv4 ( L2HostroutesGatewayTest ) : <EOL> """<STR_LIT>""" <EOL> PING_COMMAND = '<STR_LIT>' <EOL> ROUTE_COMMAND = '<STR_LIT>' <EOL> ENABLE_IP_FORWARDING_CMDS = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> SSH_COMMAND = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> NAMES_PREFIX = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( L2HostroutesGatewayTestIPv4 , cls ) . setUpClass ( ) <EOL> cls . base_cidr = '<STR_LIT>' . join ( <EOL> [ cls . subnets_behaviors . config . ipv4_prefix , '<STR_LIT:/>' , <EOL> str ( cls . subnets_behaviors . config . ipv4_suffix ) ] ) <EOL> cls . ip_version = IP ( cls . base_cidr ) . version ( ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_execute_IPv4 ( self ) : <EOL> self . _test_execute ( ) <EOL> class GatewayPoliciesTest ( NetworkingComputeFixture , <EOL> ScenarioMixin ) : <EOL> """<STR_LIT>""" <EOL> NAMES_PREFIX = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( GatewayPoliciesTest , cls ) . setUpClass ( ) <EOL> cls . servers_client = cls . compute . servers . client <EOL> cls . keypairs_client = cls . compute . keypairs . client <EOL> cls . server_behaviors = cls . compute . servers . behaviors <EOL> cls . subnets_client = cls . net . subnets . client <EOL> cls . networks_behaviors = cls . net . behaviors . networks_behaviors <EOL> cls . subnets_behaviors = cls . net . behaviors . subnets_behaviors <EOL> cls . ports_behaviors = cls . net . behaviors . ports_behaviors <EOL> cls . base_cidr = '<STR_LIT>' . join ( <EOL> [ cls . subnets_behaviors . config . ipv4_prefix , '<STR_LIT:/>' , <EOL> str ( cls . subnets_behaviors . config . ipv4_suffix ) ] ) <EOL> cls . ip_version = IP ( cls . base_cidr ) . version ( ) <EOL> cls . _create_keypair ( ) <EOL> cls . access_network , cls . access_subnet = ( <EOL> cls . _create_network_with_subnet ( '<STR_LIT>' , cls . base_cidr ) ) <EOL> cls . gateway = cls . _create_server ( '<STR_LIT>' , [ cls . access_network ] ) <EOL> remote_file_path = '<STR_LIT>' <EOL> cls . gateway . remote_client = cls . _get_remote_client ( cls . gateway ) <EOL> cls . _transfer_private_key_to_vm ( cls . gateway . remote_client . ssh_client , <EOL> cls . keypair . private_key , <EOL> remote_file_path ) <EOL> cls . ssh_command_stub = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> cls . ssh_command_stub = cls . ssh_command_stub . format ( remote_file_path ) <EOL> def test_policy_vm_default_route_is_null ( self ) : <EOL> """<STR_LIT>""" <EOL> vm = self . _create_server ( '<STR_LIT>' , <EOL> [ self . access_network ] , <EOL> public_and_service = False ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . gateway . remote_client = self . _get_remote_client ( self . gateway ) <EOL> ssh_cmd = '<STR_LIT>' . format ( self . ssh_command_stub , <EOL> getattr ( vm , self . access_network . name ) , <EOL> '<STR_LIT>' ) <EOL> output = self . _execute_ssh_command ( <EOL> self . gateway . remote_client . ssh_client , ssh_cmd ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertFalse ( output , msg ) <EOL> def test_policy_set_gateway_for_vm ( self ) : <EOL> """<STR_LIT>""" <EOL> nets = [ None , None ] <EOL> cidr = IP ( self . base_cidr ) <EOL> for i in xrange ( <NUM_LIT:2> ) : <EOL> cidr = self . _next_sequential_cidr ( cidr ) <EOL> allocation_pools = [ { "<STR_LIT:start>" : str ( IP ( cidr [ <NUM_LIT:2> ] . ip ) ) , <EOL> "<STR_LIT:end>" : str ( IP ( cidr [ - <NUM_LIT:2> ] . ip ) ) } ] <EOL> nets [ i ] , _ = self . _create_network_with_subnet ( <EOL> '<STR_LIT>' , str ( cidr ) , allocation_pools = allocation_pools , <EOL> gateway_ip = str ( IP ( cidr [ <NUM_LIT:1> ] . ip ) ) ) <EOL> if i == <NUM_LIT:0> : <EOL> expected_gateway_ip = str ( IP ( cidr [ <NUM_LIT:1> ] . ip ) ) <EOL> nets . append ( self . access_network ) <EOL> vm = self . _create_server ( '<STR_LIT>' , nets , <EOL> public_and_service = False ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . gateway . remote_client = self . _get_remote_client ( self . gateway ) <EOL> ssh_cmd = '<STR_LIT>' . format ( self . ssh_command_stub , <EOL> getattr ( vm , self . access_network . name ) , <EOL> '<STR_LIT>' ) <EOL> output = self . _execute_ssh_command ( <EOL> self . gateway . remote_client . ssh_client , ssh_cmd ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertTrue ( output , msg ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertIn ( expected_gateway_ip , output , msg ) <EOL> def test_policy_set_gateway_for_vm_with_host_routes ( self ) : <EOL> """<STR_LIT>""" <EOL> cidr = self . _next_sequential_cidr ( IP ( self . base_cidr ) ) <EOL> allocation_pools = [ { "<STR_LIT:start>" : str ( IP ( cidr [ <NUM_LIT:2> ] . ip ) ) , <EOL> "<STR_LIT:end>" : str ( IP ( cidr [ - <NUM_LIT:2> ] . ip ) ) } ] <EOL> net , subnet = self . _create_network_with_subnet ( <EOL> '<STR_LIT>' , str ( cidr ) , <EOL> allocation_pools = allocation_pools ) <EOL> expected_gateway_ip = str ( IP ( cidr [ <NUM_LIT:1> ] . ip ) ) <EOL> self . subnets_client . update_subnet ( <EOL> subnet . id , <EOL> host_routes = [ { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : expected_gateway_ip } ] ) <EOL> vm = self . _create_server ( '<STR_LIT>' , <EOL> [ net , self . access_network ] , <EOL> public_and_service = False ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . gateway . remote_client = self . _get_remote_client ( self . gateway ) <EOL> ssh_cmd = '<STR_LIT>' . format ( self . ssh_command_stub , <EOL> getattr ( vm , self . access_network . name ) , <EOL> '<STR_LIT>' ) <EOL> output = self . _execute_ssh_command ( <EOL> self . gateway . remote_client . ssh_client , ssh_cmd ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertTrue ( output , msg ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertIn ( expected_gateway_ip , output , msg ) <EOL> def test_policy_add_route_to_vm_with_host_routes ( self ) : <EOL> """<STR_LIT>""" <EOL> cidr = self . _next_sequential_cidr ( IP ( self . base_cidr ) ) <EOL> allocation_pools = [ { "<STR_LIT:start>" : str ( IP ( cidr [ <NUM_LIT:2> ] . ip ) ) , <EOL> "<STR_LIT:end>" : str ( IP ( cidr [ - <NUM_LIT:2> ] . ip ) ) } ] <EOL> net , subnet = self . _create_network_with_subnet ( <EOL> '<STR_LIT>' , str ( cidr ) , <EOL> allocation_pools = allocation_pools ) <EOL> expected_nexthop = str ( IP ( cidr [ <NUM_LIT:1> ] . ip ) ) <EOL> self . subnets_client . update_subnet ( <EOL> subnet . id , <EOL> host_routes = [ { "<STR_LIT>" : str ( cidr ) , <EOL> "<STR_LIT>" : expected_nexthop } ] ) <EOL> vm = self . _create_server ( '<STR_LIT>' , <EOL> [ net , self . access_network ] , <EOL> public_and_service = False ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . gateway . remote_client = self . _get_remote_client ( self . gateway ) <EOL> range_origin = str ( IP ( cidr [ <NUM_LIT:0> ] . ip ) ) <EOL> route_cmd = '<STR_LIT>' . format ( range_origin ) <EOL> ssh_cmd = '<STR_LIT>' . format ( self . ssh_command_stub , <EOL> getattr ( vm , self . access_network . name ) , <EOL> route_cmd ) <EOL> output = self . _execute_ssh_command ( <EOL> self . gateway . remote_client . ssh_client , ssh_cmd ) . splitlines ( ) <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> for line in output : <EOL> if ( line [ : len ( range_origin ) ] == range_origin and <EOL> expected_nexthop in line ) : <EOL> break <EOL> else : <EOL> self . fail ( msg ) <EOL> class L2HostroutesGatewayTestIPv6 ( L2HostroutesGatewayTest ) : <EOL> """<STR_LIT>""" <EOL> PING_COMMAND = '<STR_LIT>' <EOL> ROUTE_COMMAND = '<STR_LIT>' <EOL> ENABLE_IP_FORWARDING_CMDS = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> SSH_COMMAND = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> NAMES_PREFIX = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( L2HostroutesGatewayTestIPv6 , cls ) . setUpClass ( ) <EOL> cls . base_cidr = '<STR_LIT>' . join ( <EOL> [ cls . subnets_behaviors . config . ipv6_prefix , '<STR_LIT:/>' , <EOL> str ( cls . subnets_behaviors . config . ipv6_suffix ) ] ) <EOL> cls . ip_version = IP ( cls . base_cidr ) . version ( ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_execute_IPv6 ( self ) : <EOL> self . _test_execute ( ) </s>
