# Copyright (C) 2014 Nippon Telegraph and Telephone Corporation. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or # implied. # See the License for the specific language governing permissions and # limitations under the License. import unittest import requests import json import toml import os import time from ciscoconfparse import CiscoConfParse from peer_info import Peer from peer_info import Destination from peer_info import Path from constant import * class GoBGPTestBase(unittest.TestCase): gobgp_ip = GOBGP_IP gobgp_port = "8080" base_dir = CONFIG_DIRR gobgp_config_file = CONFIG_DIRR + "gobgpd.conf" gobgp_config = None initial_wait_time = 10 wait_per_retry = 5 retry_limit = (60 - initial_wait_time) / wait_per_retry def __init__(self, *args, **kwargs): super(GoBGPTestBase, self).__init__(*args, **kwargs) def setUp(self): self.quagga_configs = [] def retry_routine_for_state(self, addresses, allow_state): in_prepare_quagga = True retry_count = 0 while in_prepare_quagga: if retry_count != 0: print "please wait more (" + str(self.wait_per_retry) + " second)" time.sleep(self.wait_per_retry) if retry_count >= self.retry_limit: print "retry limit" break retry_count += 1 success_count = 0 for address in addresses: # get neighbor state and remote ip from gobgp connections try: neighbor = self.ask_gobgp(NEIGHBOR, address) except Exception: continue if neighbor is None: continue state = neighbor['info']['bgp_state'] remote_ip = neighbor['conf']['remote_ip'] if address == remote_ip and state == allow_state: success_count += 1 if success_count == len(addresses): in_prepare_quagga = False def retry_routine_for_bestpath(self, check_address, target_network, ans_nexthop): # get rib if check_address == "": rib = self.ask_gobgp(GLOBAL_RIB) else: rib = self.ask_gobgp(LOCAL_RIB, check_address) target_exist = False g_dests = rib['Destinations'] for g_dest in g_dests: best_path_idx = g_dest['BestPathIdx'] if target_network == g_dest['Prefix']: target_exist = True g_paths = g_dest['Paths'] idx = 0 if len(g_paths) < 2: print "target path has not been bestpath selected yet." print "please wait more (" + str(self.wait_per_retry) + " second)" time.sleep(self.wait_per_retry) self.retry_routine_for_bestpath(check_address, target_network, ans_nexthop) return for g_path in g_paths: print "best_path_Idx: " + str(best_path_idx) + "idx: " + str(idx) print "pre: ", g_dest['Prefix'], "net: ", g_path['Network'], "next: ", g_path['Nexthop'] if str(best_path_idx) == str(idx): rep_nexthop = g_path['Nexthop'] idx += 1 if target_exist is False: print "target path has not been receive yet." print "please wait more (" + str(self.wait_per_retry) + " second)" time.sleep(self.wait_per_retry) self.retry_routine_for_bestpath(check_address, target_network, ans_nexthop) return self.assertEqual(ans_nexthop, rep_nexthop) def load_gobgp_config(self): try: self.gobgp_config = toml.loads(open(self.gobgp_config_file).read()) except IOError, (errno, strerror): print "I/O error(%s): %s" % (errno, strerror) # load configration from quagga(bgpd.conf) def load_quagga_config(self): dirs = [] try: content = os.listdir(self.base_dir) for item in content: if "q" != item[0]: continue if os.path.isdir(os.path.join(self.base_dir, item)): dirs.append(item) except OSError, (errno, strerror): print "I/O error(%s): %s" % (errno, strerror) for dir in dirs: config_path = self.base_dir + dir + "/bgpd.conf" config = CiscoConfParse(config_path) peer_ip = config.find_objects(r"^!\smy\saddress")[0].text.split(" ")[3] peer_ip_version = config.find_objects(r"^!\smy\sip_version")[0].text.split(" ")[3] peer_id = config.find_objects(r"^bgp\srouter-id")[0].text.split(" ")[2] peer_as = config.find_objects(r"^router\sbgp")[0].text.split(" ")[2] quagga_config = Peer(peer_ip, peer_id, peer_as, peer_ip_version) networks = config.find_objects(r"^network") if len(networks) == 0: continue for network in networks: elems = network.text.split(" ") prefix = elems[1].split("/")[0] network = elems[1] nexthop = peer_ip path = Path(network, nexthop) dest = Destination(prefix) dest.paths.append(path) quagga_config.destinations[prefix] = dest # print "prefix: " + prefix # print "network: " + network # print "nexthop: " + nexthop neighbors = config.find_objects(r"^neighbor\s.*\sremote-as") if len(neighbors) == 0: continue for neighbor in neighbors: elems = neighbor.text.split(" ") neighbor = Peer(elems[1], None, elems[3], None) quagga_config.neighbors.append(neighbor) self.quagga_configs.append(quagga_config) # get address of each neighbor from gobpg configration def get_neighbor_address(self, config): address = [] neighbors_config = config['NeighborList'] for neighbor_config in neighbors_config: neighbor_ip = neighbor_config['NeighborAddress'] address.append(neighbor_ip) return address def check_load_config(self): self.load_gobgp_config() self.load_quagga_config() if self.gobgp_config is None: print "Failed to read the gobgp configuration file" return False if len(self.quagga_configs) == 0: print "Failed to read the quagga configuration file" return False return True def ask_gobgp(self, what, who="", af="ipv4"): url = "http://" + self.gobgp_ip + ":" + self.gobgp_port + "/v1/bgp/" if what == GLOBAL_RIB: url += "/".join([what, af]) elif what == NEIGHBOR: url += "/".join([NEIGHBOR, who]) else: url += "/".join([NEIGHBOR, who, what, af]) r = requests.get(url) result = json.loads(r.text) return result