KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use create_port_forwarding method in tempest

Best Python code snippet using tempest_python

test_port_forwarding_negative.py

Source:test_port_forwarding_negative.py

...35    def test_mapping_same_fip_and_external_port_to_different_dest(self):36        port1 = self.create_port(self.network)37        port2 = self.create_port(self.network)38        fip_for_pf = self.create_floatingip()39        self.create_port_forwarding(40            fip_for_pf['id'],41            internal_port_id=port1['id'],42            internal_ip_address=port1['fixed_ips'][0]['ip_address'],43            internal_port=1111, external_port=2222,44            protocol="tcp")45        self.assertRaises(46            exceptions.BadRequest,47            self.create_port_forwarding,48            fip_for_pf['id'],49            internal_port_id=port2['id'],50            internal_ip_address=port2['fixed_ips'][0]['ip_address'],51            internal_port=3333, external_port=2222,52            protocol="tcp")53    @decorators.attr(type='negative')54    @decorators.idempotent_id('0c229a4c-9f28-11ea-bb37-0242ac130002')55    def test_mapping_different_external_ports_to_the_same_destination(self):56        port = self.create_port(self.network)57        fip_for_pf = self.create_floatingip()58        self.create_port_forwarding(59            fip_for_pf['id'],60            internal_port_id=port['id'],61            internal_ip_address=port['fixed_ips'][0]['ip_address'],62            internal_port=1111, external_port=3333,63            protocol="tcp")64        self.assertRaises(65            exceptions.BadRequest,66            self.create_port_forwarding,67            fip_for_pf['id'],68            internal_port_id=port['id'],69            internal_ip_address=port['fixed_ips'][0]['ip_address'],70            internal_port=1111, external_port=5555,71            protocol="tcp")72    @decorators.attr(type='negative')73    @decorators.idempotent_id('e9d3ffb6-e5bf-421d-acaa-ee6010dfbf14')74    def test_out_of_range_ports(self):75        port = self.create_port(self.network)76        fip_for_pf = self.create_floatingip()77        pf_params = {78            'internal_port_id': port['id'],79            'internal_ip_address': port['fixed_ips'][0]['ip_address'],80            'internal_port': 1111,81            'external_port': 3333,82            'protocol': "tcp"}83        pf = self.create_port_forwarding(fip_for_pf['id'], **pf_params)84        # Check: Invalid input for external_port update85        self.assertRaises(86            exceptions.BadRequest,87            self.update_port_forwarding,88            fip_for_pf['id'], pf['id'], external_port=108343)89        # Check: Invalid input for internal_port update90        self.assertRaises(91            exceptions.BadRequest,92            self.update_port_forwarding,93            fip_for_pf['id'], pf['id'], internal_port=108343)94        # Check: Invalid input for external_port create95        pf_params['internal_port'] = 444496        pf_params['external_port'] = 33333397        self.assertRaises(...

rds-session

Source:rds-session

...15    DEFAULT_PROFILE = os.getenv('AWS_PROFILE')16DEFAULT_LOCAL_PORT = '2422'17DEFAULT_DB_PORT = '3308'18pwd = os.getcwd()19def create_port_forwarding(local_port, DB_port, rds_endpoint):20    cmd = "ssh localhost -p " + local_port + " -L " + DB_port + ":" + rds_endpoint + ":3306 -N"21    os.system(cmd)22def clean_known_host(local_port):23    known_host = pwd + "/.ssh/known_hosts"24    # f = open(known_host, "r")25    # print(f.read())26    with open(known_host, "r") as f:27        lines = f.readlines()28    with open(known_host, "w") as f:29        for line in lines:30            if "[localhost]:" + local_port in line:31                continue32            else:33                f.write(line)34def main():35    parser = argparse.ArgumentParser()36    parser.add_argument('-n','--name',dest='name',required=False, help='rdp.py -n *-idv-* --profile dev|stg|prod --region eu-west-1' )37    parser.add_argument('-p','--profile', dest='profile', default=None, required=False, help='rdp.py -n *-idv-* --profile dev|stg|prod --region eu-west-1')38    parser.add_argument('-r','--region',dest='region',required=False, help='rdp.py -n *-idv-* --profile dev|stg|prod --region eu-west-1' )39    parser.add_argument('-l', '--list',dest='list',required=False,action='store_true',help='Only list resources no start ssm session is triggered')40    parser.add_argument(41        '-t', '--terminal',42        dest='terminal',43        required=False,44        action='store_true',45        help='Instead of a Port forwarding starts a terminal session'46    )47    parser.set_defaults(profile=DEFAULT_PROFILE)48    args, unknown = parser.parse_known_args()49    if args.name is None:50        rds_name = '*'51    else:52        rds_name = args.name53    # filters = [{âââââââ54    #     'Name': 'DBName',55    #     'Values': [rds_name]56    #     }âââââââ,57    #     {âââââââ58    #         'Name': 'DBInstanceStatus',59    #         'Values': ['available']60    #     }âââââââ]61    if args.region is None:62        aws_region = 'eu-west-1'63    else:64        aws_region = args.region65    if args.region is None:66        session = boto3.Session(profile_name=args.profile)67    else:68        session = boto3.Session(region_name=args.region,profile_name=args.profile)69    rds_client = session.client('rds')70    # response = rds_client.describe_db_instances(Filters=filters)71    response = rds_client.describe_db_instances()72    #pprint(response)73    instances = []74    index = 175    # for r in response['Reservations']:76    for i in response['DBInstances']:77        if i['DBInstanceStatus'] == 'available':78            if rds_name != '*':79                if rds_name in i['DBInstanceIdentifier']:80                    instances.append([index, i['DBInstanceIdentifier'],i['Endpoint']['Address'],i['Endpoint']['Port']])81                    index += 182            else:83                instances.append([index, i['DBInstanceIdentifier'],i['Endpoint']['Address'],i['Endpoint']['Port']])84                index += 185    print(tabulate(instances))86    if not args.list:87        try:88            print('\n(Cancel CTRL+C)')89            choice = int(input("Enter a selection number: "))90            choice -= 191            if not args.terminal:92                LOCAL_PORT = input("Local Port [Default 2422]: ")93                DB_PORT = input("Database Port [Default 3308]: ")94                if LOCAL_PORT == '':95                    LOCAL_PORT = DEFAULT_LOCAL_PORT96                else:97                    validate = int(LOCAL_PORT)98                if DB_PORT == '':99                    DB_PORT = DEFAULT_DB_PORT100                else:101                    validate = int(DB_PORT)102                print('\nCreating tunnel to:')103                connection = {104                    'Name': instances[choice][1],105                    'Address': instances[choice][2]106                }107                print(json.dumps(connection, indent=2))108                clean_known_host(LOCAL_PORT)109                create_port_forwarding(LOCAL_PORT,DB_PORT,instances[choice][2])110        except (KeyboardInterrupt, SystemExit):111            print('CTRL+C:\nExit')112        except (IndexError, ValueError):113            print('\nERROR: Invalid Value')114if __name__ == '__main__':...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.