Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use replace_route method in localstack

Best Python code snippet using localstack_python

failover_gw.py

Source:failover_gw.py

...14    if not m:15        return None16    else:17        return m.group(1)18def replace_route(route, gw):19    res = subprocess.check_output([ 'ip', 'route', 'replace', route, 'via', gw])20    res = res.decode('utf-8')21    logging.debug("'ip route replace %s via %s' returned '%s'", route, gw, res)22    return True23def main():24    config_file = "failover_gw.yaml"25    config = yaml.safe_load(open(config_file, 'r'))26    if config['debug']:27        log_level = "DEBUG"28    else:29        log_level = "INFO"30    log_format='%(asctime)s (%(levelname)s) %(message)s'31    logging.basicConfig(level=log_level, format=log_format)32    poll_interval = config['poll_interval'] or 1033    gw_interface = config['gw_interface'] or 'eth0'34    check_ip = config['check_ip']35    managed_route = config['managed_route']36    primary_gw = config['primary_gw']37    backup_gw = config['backup_gw']38    ping_timeout = config['ping_timeout'] or 200039    ping_count = config['ping_count'] or  340    logging.info("Started")41    while True:42        # Arp request primary and backup gateway looking for a response43        arp_primary = ArpRequest(primary_gw, gw_interface)44        primary_is_up = arp_primary.request()45        if primary_is_up:46            ping_via_primary = Ping(check_ip,47                                    ping_timeout,48                                    raw=True,49                                    source_interface=gw_interface,50                                    dest_mac=arp_primary.getHardwareAddress())51            res = ping_via_primary.run(ping_count)52            if res.ret_code == 0:53                logging.info("Ping %s via primary gateway %s RTT=%.3f,%.3f,%.3fms",54                             check_ip, primary_gw,55                             res.min_rtt, res.avg_rtt, res.max_rtt)56                primary_gw_up = True57            else:58                logging.info("Ping %s via primary gateway %s failed : %s",59                             check_ip, primary_gw, res.output)60                primary_gw_up = False61        else:62            logging.info("Primary gateway %s is down", primary_gw)63            primary_gw_up = False64        arp_backup = ArpRequest(backup_gw, gw_interface)65        backup_is_up = arp_backup.request()66        if backup_is_up:67            ping_via_backup = Ping(check_ip,68                                   ping_timeout,69                                   raw=True,70                                   source_interface=gw_interface,71                                   dest_mac=arp_backup.getHardwareAddress())72            res = ping_via_backup.run(ping_count)73            if res.ret_code == 0:74                logging.info("Ping %s via backup gateway %s RTT=%.3f,%.3f,%.3fms",75                             check_ip, backup_gw,76                             res.min_rtt, res.avg_rtt, res.max_rtt)77                backup_gw_up = True78            else:79                logging.info("Ping %s via backup gateway %s failed : %s",80                             check_ip, backup_gw, res.output)81                backup_gw_up = False82            backup_gw_up = (res.ret_code == 0)83        else:84            logging.info("Backup gateway %s is down", backup_gw)85            backup_gw_up = False86        logging.debug("primary gateway ARP is %d GW is %d secondary gateway ARP is %d GW is %d",87                      primary_is_up, primary_gw_up, backup_is_up, backup_gw_up)88        current_gw = get_route_gw(managed_route)89        logging.debug("Current gateway %s" % current_gw)90        if primary_gw_up:91            if managed_route and current_gw != primary_gw:92                logging.info("Switch route %s to primary gateway %s",93                             managed_route, primary_gw)94                replace_route(managed_route, primary_gw)95        elif backup_gw_up:96            if managed_route and current_gw != backup_gw:97                logging.info("Switch route %s to backup gateway %s",98                             managed_route, backup_gw)99                replace_route(managed_route, backup_gw)100        else:101            logging.info("No routes for %s as neither gateway responding for %s",102                         managed_route, check_ip)103        time.sleep(config['poll_interval'])104if __name__ == '__main__':...

check_ftdv_status.py.erb

Source:check_ftdv_status.py.erb

...14rtb = vpc.get_all_route_tables(route_table_ids=routeTable)15apiURL = 'https://<%= @apiurl %>'16headers = {'content-type': "application/json", 'cache-control': "no-cache"}17def routePdFTDV1():18    vpc.replace_route(route_table_id=routeTable,19                      destination_cidr_block='0.0.0.0/0',20                      interface_id=eniFTDv1)21    vpc.replace_route(route_table_id=routeTable,22                      destination_cidr_block='<%= @ftdv_vpn_ip1 %>',23                      interface_id=eniFTDv1)24    vpc.replace_route(route_table_id=routeTable,25                      destination_cidr_block='<%= @ftdv_vpn_ip2 %>',26                      interface_id=eniFTDv1)27def routePdFTDV2():28    vpc.replace_route(route_table_id=routeTable,29                      destination_cidr_block='0.0.0.0/0',30                      interface_id=eniFTDv2)31    vpc.replace_route(route_table_id=routeTable,32                      destination_cidr_block='<%= @ftdv_vpn_ip1 %>',33                      interface_id=eniFTDv2)34    vpc.replace_route(route_table_id=routeTable,35                      destination_cidr_block='<%= @ftdv_vpn_ip2 %>',36                      interface_id=eniFTDv2)37for rt in rtb:38    for a in rt.routes:39        eniRTB = a.interface_id40valid = True41while valid:42    insidePing1 = os.system("ping -c 15 -w 5 "+insideFTDv1+"|grep -c '100% packet loss'")43    insidePing2 = os.system("ping -c 15 -w 5 "+insideFTDv2+"|grep -c '100% packet loss'")44    # insidePing1 IP exit with 0 failure45    # Lock file doesn't exsist46    if insidePing1 == 0 and (not os.path.isfile('/tmp/ftdv_lock')):47        # if eniRTB not equal to ENI2 then change route48        if eniRTB == eniFTDv1:...

Test.py

Source:Test.py

2route_input = "6863930,90000260498780,90000260412950,5765410,5765420,877579610,877579620,491431860,6647991,422366081,422366071,6714661,422366201,422366191,90000285124710,90000285201350,132084800,622185390,422361390,422361400,6808220,6809490,6809500,5849390,5791461,6714681,877579451,877579441,6716081,5762391,422362640,422362650,422362630,6738250,132085000,132085010,6787611,422362531,422362521,7008661,132085031,132085021,7008701,7008691,6643861,6799400,6799410,126347950,126347960,90000242392280,90000242392200,90000242392320,90000242315780,300503480,300503490,152550970,152550990,152551010,152551030,152551050,6629510,90000284502220,90000284359420,6823670,6823690,6823660,6823610,300503770,300503790,300503800,6629470,6629410,6629420,6823560,6824140,6824160,701371610,701371620,6808920,6808940,6808970,6659330,622586230,622586240,5823321,6839311,6671701,6671691,5747601747601,5747611,6846120,6846130,5791380,5777010,6678250,6644370,6644380,6856660,6677690,6644420,6644430,6677730,6677740,6677790,500623910,500623920,6677750,501738330,501738340,6858750,6858760"3# comm = "1000,2,200,5,10001"4# comm = "1000,2,3,4,5,10001"5link_input = "6669771,6669791,6669781,5777051,5823321,6839311,6671701,6671691,5747601"6def replace_route(link, route):7    comm_list = link.split(",")8    route_list = route.split(",")9    flag = False10    for i in range(len(comm_list)):11        replace_start_index = -112        replace_end_index = -113        if comm_list[i] in route_list:14            replace_start_index = route_list.index(comm_list[i])15        check_end_index = -116        if i != len(comm_list):17            for i_r in reversed(range(len(comm_list))):18                if i_r > i and comm_list[i_r] in route_list:19                    replace_end_index = route_list.index(comm_list[i_r])20                    check_end_index = i_r21                    break22        if (replace_end_index > replace_start_index) and (replace_start_index > 0) and (replace_end_index > 0):23            replace_str = ','.join(comm_list[i:check_end_index+1])24            be_replace_str = ','.join(route_list[replace_start_index:replace_end_index+1])25            if be_replace_str.strip() != replace_str.strip():26                route = route.replace(be_replace_str, replace_str)27                flag = True28            # print("replace_str: ", replace_str)29            # print("be_replace_str: ", be_replace_str)30            # print("s: ", s)31            break32    return route, flag33r, f = replace_route(link_input, route_input)34print("r=", r)...

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.