<s> from boto . ec2 import networkinterface <EOL> import netaddr <EOL> import tempest . cloudscaling . thirdparty . scenario . aws_compat . base as aws_base <EOL> from tempest . common . utils import data_utils <EOL> from tempest . common . utils . linux import remote_client <EOL> from tempest import test <EOL> import logging <EOL> logging . getLogger ( '<STR_LIT>' ) . setLevel ( logging . CRITICAL ) <EOL> class VPC_NAT_Scenario ( aws_base . BaseAWSTest ) : <EOL> """<STR_LIT>""" <EOL> class Context ( object ) : <EOL> vpc = None <EOL> internet_gateway = None <EOL> web_subnet = None <EOL> db_subnet = None <EOL> main_route_table = None <EOL> custom_route_table = None <EOL> web_security_group = None <EOL> nat_security_group = None <EOL> db_security_group = None <EOL> web_instance = None <EOL> db_instance = None <EOL> nat_instance = None <EOL> @ classmethod <EOL> @ test . safe_setup <EOL> def setUpClass ( cls ) : <EOL> super ( VPC_NAT_Scenario , cls ) . setUpClass ( ) <EOL> cls . ctx = cls . Context ( ) <EOL> cls . zone = cls . config . boto . aws_zone <EOL> cfg = cls . config . cloudscaling <EOL> cls . ssh_user = cfg . general_ssh_user_name <EOL> cls . vpc_cidr = netaddr . IPNetwork ( cfg . vpc_cidr ) <EOL> cls . web_subnet , cls . db_subnet = cls . vpc_cidr . subnet ( <EOL> cfg . vpc_subnet_prefix , <NUM_LIT:2> ) <EOL> cls . test_client_cidr = netaddr . IPNetwork ( cfg . test_client_cidr ) <EOL> cls . image_id = cls . _prepare_image_id ( cfg . general_image_name ) <EOL> cls . keypair = cls . _prepare_key_pair ( ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> if cls . ctx is not None : <EOL> for group in [ cls . ctx . web_security_group , <EOL> cls . ctx . nat_security_group , <EOL> cls . ctx . db_security_group ] : <EOL> if not group : <EOL> continue <EOL> try : <EOL> cls . _revoke_security_group_linked_rules ( group ) <EOL> except Exception : <EOL> pass <EOL> super ( VPC_NAT_Scenario , cls ) . tearDownClass ( ) <EOL> @ classmethod <EOL> def _revoke_security_group_linked_rules ( cls , group ) : <EOL> groups = cls . vpc_client . get_all_security_groups ( group_ids = [ group . id ] ) <EOL> if len ( groups ) == <NUM_LIT:0> : <EOL> return <EOL> sg = groups [ <NUM_LIT:0> ] <EOL> for rule in sg . rules : <EOL> for grant in rule . grants : <EOL> if not grant . cidr_ip : <EOL> cls . vpc_client . revoke_security_group ( <EOL> group_id = sg . id , <EOL> ip_protocol = rule . ip_protocol , <EOL> from_port = rule . from_port , <EOL> to_port = rule . to_port , <EOL> src_security_group_group_id = grant . groupId ) <EOL> for rule in sg . rules_egress : <EOL> for grant in rule . grants : <EOL> if not grant . cidr_ip : <EOL> cls . vpc_client . revoke_security_group_egress ( <EOL> sg . id , <EOL> rule . ip_protocol , <EOL> from_port = rule . from_port , <EOL> to_port = rule . to_port , <EOL> src_group_id = grant . groupId ) <EOL> def test_000_create_vpc ( self ) : <EOL> """<STR_LIT>""" <EOL> dhcp_opts = self . vpc_client . create_dhcp_options ( <EOL> domain_name_servers = [ '<STR_LIT>' ] ) <EOL> self . assertIsNotNone ( dhcp_opts ) <EOL> self . assertTrue ( dhcp_opts . id ) <EOL> self . addResourceCleanUp ( self . vpc_client . delete_dhcp_options , <EOL> dhcp_opts . id ) <EOL> vpc = self . vpc_client . create_vpc ( str ( self . vpc_cidr ) ) <EOL> self . assertIsNotNone ( vpc ) <EOL> self . assertTrue ( vpc . id ) <EOL> self . addResourceCleanUp ( self . vpc_client . delete_vpc , vpc . id ) <EOL> self . assertTrue ( self . vpc_client . associate_dhcp_options ( dhcp_opts . id , <EOL> vpc . id ) ) <EOL> self . ctx . vpc = vpc <EOL> def test_001_create_internet_gateway ( self ) : <EOL> """<STR_LIT>""" <EOL> ig = self . vpc_client . create_internet_gateway ( ) <EOL> self . assertIsNotNone ( ig ) <EOL> self . assertTrue ( ig . id ) <EOL> self . addResourceCleanUp ( self . _destroy_internet_gateway , ig ) <EOL> status = self . vpc_client . attach_internet_gateway ( ig . id , <EOL> self . ctx . vpc . id ) <EOL> self . assertTrue ( status ) <EOL> self . ctx . internet_gateway = ig <EOL> def test_010_create_subnets ( self ) : <EOL> """<STR_LIT>""" <EOL> sn = self . vpc_client . create_subnet ( self . ctx . vpc . id , <EOL> str ( self . web_subnet ) , <EOL> self . zone ) <EOL> self . assertIsNotNone ( sn ) <EOL> self . assertTrue ( sn . id ) <EOL> self . addResourceCleanUp ( self . vpc_client . delete_subnet , sn . id ) <EOL> self . ctx . web_subnet = sn <EOL> sn = self . vpc_client . create_subnet ( self . ctx . vpc . id , <EOL> str ( self . db_subnet ) , <EOL> self . zone ) <EOL> self . assertIsNotNone ( sn ) <EOL> self . assertTrue ( sn . id ) <EOL> self . addResourceCleanUp ( self . vpc_client . delete_subnet , sn . id ) <EOL> self . ctx . db_subnet = sn <EOL> def test_020_get_main_route_table ( self ) : <EOL> """<STR_LIT>""" <EOL> rtables = self . vpc_client . get_all_route_tables ( <EOL> filters = [ ( "<STR_LIT>" , self . ctx . vpc . id ) ] ) <EOL> self . assertIsNotNone ( rtables ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( rtables ) ) <EOL> self . ctx . main_route_table = rtables [ <NUM_LIT:0> ] <EOL> def test_025_create_custom_route_table ( self ) : <EOL> """<STR_LIT>""" <EOL> rtable = self . vpc_client . create_route_table ( self . ctx . vpc . id ) <EOL> self . assertIsNotNone ( rtable ) <EOL> self . assertTrue ( rtable . id ) <EOL> self . addResourceCleanUp ( self . vpc_client . delete_route_table , rtable . id ) <EOL> ig = self . ctx . internet_gateway <EOL> status = self . vpc_client . create_route ( rtable . id , "<STR_LIT>" , <EOL> gateway_id = ig . id ) <EOL> self . assertTrue ( status ) <EOL> association_id = self . vpc_client . associate_route_table ( <EOL> rtable . id , self . ctx . web_subnet . id ) <EOL> self . assertTrue ( association_id ) <EOL> self . addResourceCleanUp ( self . vpc_client . disassociate_route_table , <EOL> association_id ) <EOL> self . ctx . custom_route_table = rtable <EOL> def test_050_create_security_groups ( self ) : <EOL> """<STR_LIT>""" <EOL> sg = self . vpc_client . create_security_group ( <EOL> data_utils . rand_name ( "<STR_LIT>" ) , <EOL> data_utils . rand_name ( "<STR_LIT>" ) , <EOL> self . ctx . vpc . id ) <EOL> self . assertIsNotNone ( sg ) <EOL> self . assertTrue ( sg . id ) <EOL> self . addResourceCleanUp ( self . _destroy_security_group_wait , sg ) <EOL> self . ctx . web_security_group = sg <EOL> sg = self . vpc_client . create_security_group ( <EOL> data_utils . rand_name ( "<STR_LIT>" ) , <EOL> data_utils . rand_name ( "<STR_LIT>" ) , <EOL> self . ctx . vpc . id ) <EOL> self . assertIsNotNone ( sg ) <EOL> self . assertTrue ( sg . id ) <EOL> self . addResourceCleanUp ( self . _destroy_security_group_wait , sg ) <EOL> self . ctx . nat_security_group = sg <EOL> sg = self . vpc_client . create_security_group ( <EOL> data_utils . rand_name ( "<STR_LIT>" ) , <EOL> data_utils . rand_name ( "<STR_LIT>" ) , <EOL> self . ctx . vpc . id ) <EOL> self . assertIsNotNone ( sg ) <EOL> self . assertTrue ( sg . id ) <EOL> self . addResourceCleanUp ( self . _destroy_security_group_wait , sg ) <EOL> self . ctx . db_security_group = sg <EOL> sg = self . ctx . web_security_group <EOL> status = self . vpc_client . revoke_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , <EOL> src_group_id = self . ctx . db_security_group . id ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , <EOL> src_group_id = self . ctx . db_security_group . id ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , <EOL> src_group_id = self . ctx . db_security_group . id ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> cidr_ip = str ( self . test_client_cidr ) ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , cidr_ip = str ( self . test_client_cidr ) ) <EOL> self . assertTrue ( status ) <EOL> sg = self . ctx . nat_security_group <EOL> status = self . vpc_client . revoke_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> cidr_ip = str ( self . db_subnet ) ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> cidr_ip = str ( self . db_subnet ) ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , from_port = <NUM_LIT> , to_port = <NUM_LIT> , <EOL> cidr_ip = str ( self . test_client_cidr ) ) <EOL> self . assertTrue ( status ) <EOL> sg = self . ctx . db_security_group <EOL> status = self . vpc_client . revoke_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group_egress ( <EOL> sg . id , "<STR_LIT>" , <NUM_LIT> , <NUM_LIT> , cidr_ip = "<STR_LIT>" ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , <EOL> from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , <EOL> src_security_group_group_id = self . ctx . web_security_group . id ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , <EOL> from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , <EOL> src_security_group_group_id = self . ctx . web_security_group . id ) <EOL> self . assertTrue ( status ) <EOL> status = self . vpc_client . authorize_security_group ( <EOL> group_id = sg . id , ip_protocol = "<STR_LIT>" , <EOL> from_port = <NUM_LIT> , <EOL> to_port = <NUM_LIT> , <EOL> src_security_group_group_id = self . ctx . web_security_group . id ) <EOL> self . assertTrue ( status ) <EOL> def test_100_launch_nat_instance ( self ) : <EOL> """<STR_LIT>""" <EOL> interface_web = networkinterface . NetworkInterfaceSpecification ( <EOL> subnet_id = self . ctx . web_subnet . id , <EOL> groups = [ self . ctx . nat_security_group . id ] ) <EOL> interface_db = networkinterface . NetworkInterfaceSpecification ( <EOL> subnet_id = self . ctx . db_subnet . id , <EOL> groups = [ self . ctx . nat_security_group . id ] ) <EOL> reservation = self . vpc_client . run_instances ( <EOL> self . image_id , <EOL> key_name = self . keypair . name , <EOL> instance_type = self . instance_type , <EOL> placement = self . zone , <EOL> network_interfaces = ( <EOL> networkinterface . NetworkInterfaceCollection ( <EOL> interface_web , interface_db ) ) <EOL> ) <EOL> self . assertIsNotNone ( reservation ) <EOL> self . addResourceCleanUp ( self . destroy_reservation , reservation ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( reservation . instances ) ) <EOL> instance = reservation . instances [ <NUM_LIT:0> ] <EOL> if instance . state != "<STR_LIT>" : <EOL> self . assertInstanceStateWait ( instance , "<STR_LIT>" ) <EOL> instance . ip_address = self . _prepare_public_ip ( <EOL> instance , <EOL> instance . interfaces [ <NUM_LIT:0> ] . id ) <EOL> internal_interface_id = instance . interfaces [ <NUM_LIT:1> ] . id <EOL> status = self . vpc_client . modify_network_interface_attribute ( <EOL> internal_interface_id , <EOL> attr = '<STR_LIT>' , <EOL> value = False ) <EOL> self . assertTrue ( status ) <EOL> rtable = self . ctx . main_route_table <EOL> status = self . vpc_client . create_route ( <EOL> rtable . id , "<STR_LIT>" , <EOL> interface_id = internal_interface_id ) <EOL> self . assertTrue ( status ) <EOL> self . ctx . nat_instance = instance <EOL> def test_101_launch_instances ( self ) : <EOL> """<STR_LIT>""" <EOL> reservation = self . vpc_client . run_instances ( <EOL> self . image_id , <EOL> key_name = self . keypair . name , <EOL> security_group_ids = [ self . ctx . web_security_group . id ] , <EOL> instance_type = self . instance_type , <EOL> placement = self . zone , <EOL> subnet_id = self . ctx . web_subnet . id ) <EOL> self . assertIsNotNone ( reservation ) <EOL> self . addResourceCleanUp ( self . destroy_reservation , reservation ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( reservation . instances ) ) <EOL> instance = reservation . instances [ <NUM_LIT:0> ] <EOL> if instance . state != "<STR_LIT>" : <EOL> self . assertInstanceStateWait ( instance , "<STR_LIT>" ) <EOL> instance . ip_address = self . _prepare_public_ip ( instance ) <EOL> self . ctx . web_instance = instance <EOL> reservation = self . vpc_client . run_instances ( <EOL> self . image_id , <EOL> key_name = self . keypair . name , <EOL> security_group_ids = [ self . ctx . db_security_group . id ] , <EOL> instance_type = self . instance_type , <EOL> placement = self . zone , <EOL> subnet_id = self . ctx . db_subnet . id ) <EOL> self . assertIsNotNone ( reservation ) <EOL> self . addResourceCleanUp ( self . destroy_reservation , reservation ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( reservation . instances ) ) <EOL> instance = reservation . instances [ <NUM_LIT:0> ] <EOL> if instance . state != "<STR_LIT>" : <EOL> self . assertInstanceStateWait ( instance , "<STR_LIT>" ) <EOL> self . ctx . db_instance = instance <EOL> def test_102_tune_nat_instance ( self ) : <EOL> """<STR_LIT>""" <EOL> instance = self . ctx . nat_instance <EOL> address = instance . ip_address <EOL> ssh = remote_client . RemoteClient ( address , <EOL> self . ssh_user , <EOL> pkey = self . keypair . material ) <EOL> ssh . exec_command ( "<STR_LIT>" <EOL> "<STR_LIT>" % str ( self . vpc_cidr ) ) <EOL> ssh . exec_command ( "<STR_LIT>" ) <EOL> ssh . exec_command ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> ssh . exec_command ( "<STR_LIT>" ) <EOL> def test_200_check_connectivity ( self ) : <EOL> """<STR_LIT>""" <EOL> web_ip = self . ctx . web_instance . ip_address <EOL> db_ip = self . ctx . db_instance . private_ip_address <EOL> ssh = remote_client . RemoteClient ( web_ip , <EOL> self . ssh_user , <EOL> pkey = self . keypair . material ) <EOL> ssh_conn = ssh . ssh_client . _get_ssh_connection ( ) <EOL> sftp = ssh_conn . open_sftp ( ) <EOL> fr = sftp . file ( "<STR_LIT>" , '<STR_LIT:wb>' ) <EOL> fr . set_pipelined ( True ) <EOL> fr . write ( self . keypair . material ) <EOL> fr . close ( ) <EOL> ssh_conn . close ( ) <EOL> ssh . exec_command ( '<STR_LIT>' ) <EOL> ssh . exec_command ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> { "<STR_LIT:user>" : self . ssh_user , "<STR_LIT>" : db_ip } ) </s>
