From 2448d00200e92121979241dee8e6eaaf95246560 Mon Sep 17 00:00:00 2001 From: nguggarigoud Date: Mon, 31 May 2021 20:22:59 -0700 Subject: [PATCH] tests: Adding ospfv3 basic functionality test cases 1. Adding APIs to configure and verify ospfv3. 2. Adding ospfv3 base functionality testcase. Signed-off-by: nguggarigoud --- tests/topotests/lib/common_config.py | 20 +- tests/topotests/lib/ospf.py | 969 +++++++++++++++++- .../ospfv3_rte_calc.json | 173 ++++ .../ospfv3_single_area.json | 190 ++++ .../test_ospfv3_rte_calc.py | 374 +++++++ .../test_ospfv3_single_area.py | 417 ++++++++ 6 files changed, 2096 insertions(+), 47 deletions(-) create mode 100644 tests/topotests/ospfv3_basic_functionality/ospfv3_rte_calc.json create mode 100644 tests/topotests/ospfv3_basic_functionality/ospfv3_single_area.json create mode 100644 tests/topotests/ospfv3_basic_functionality/test_ospfv3_rte_calc.py create mode 100644 tests/topotests/ospfv3_basic_functionality/test_ospfv3_single_area.py diff --git a/tests/topotests/lib/common_config.py b/tests/topotests/lib/common_config.py index ee7cd6a7af..3f78f020bc 100644 --- a/tests/topotests/lib/common_config.py +++ b/tests/topotests/lib/common_config.py @@ -123,6 +123,17 @@ DEBUG_LOGS = { "debug ospf te", "debug ospf zebra", ], + "ospf6": [ + "debug ospf6 event", + "debug ospf6 ism", + "debug ospf6 lsa", + "debug ospf6 nsm", + "debug ospf6 nssa", + "debug ospf6 packet all", + "debug ospf6 sr", + "debug ospf6 te", + "debug ospf6 zebra", + ], } if config.has_option("topogen", "verbosity"): @@ -422,7 +433,10 @@ def check_router_status(tgen): daemons.append("zebra") if "pimd" in result: daemons.append("pimd") - + if "ospfd" in result: + daemons.append("ospfd") + if "ospf6d" in result: + daemons.append("ospf6d") rnode.startDaemons(daemons) except Exception as e: @@ -890,6 +904,10 @@ def topo_daemons(tgen, topo): for val in topo["routers"][rtr]["links"].values(): if "pim" in val and "pimd" not in daemon_list: daemon_list.append("pimd") + if "ospf" in val and "ospfd" not in daemon_list: + daemon_list.append("ospfd") + if "ospf6" in val and "ospf6d" not in daemon_list: + daemon_list.append("ospf6d") break return daemon_list diff --git a/tests/topotests/lib/ospf.py b/tests/topotests/lib/ospf.py index 218dd7eb3e..3f39b93d8c 100644 --- a/tests/topotests/lib/ospf.py +++ b/tests/topotests/lib/ospf.py @@ -18,13 +18,16 @@ # OF THIS SOFTWARE. # -from copy import deepcopy import traceback +import ipaddr +import ipaddress +import sys + +from copy import deepcopy from time import sleep from lib.topolog import logger -import ipaddr from lib.topotest import frr_unicode - +from ipaddress import IPv6Address # Import common_config to use commomnly used APIs from lib.common_config import ( create_common_configuration, @@ -86,10 +89,21 @@ def create_router_ospf(tgen, topo, input_dict=None, build=False, load_config=Tru logger.debug("Router %s: 'ospf' not present in input_dict", router) continue - result = __create_ospf_global(tgen, input_dict, router, build, load_config) + result = __create_ospf_global( + tgen, input_dict, router, build, load_config) if result is True: ospf_data = input_dict[router]["ospf"] + for router in input_dict.keys(): + if "ospf6" not in input_dict[router]: + logger.debug("Router %s: 'ospf6' not present in input_dict", router) + continue + + result = __create_ospf_global( + tgen, input_dict, router, build, load_config, ospf='ospf6') + if result is True: + ospf_data = input_dict[router]["ospf6"] + logger.debug("Exiting lib API: create_router_ospf()") return result @@ -158,6 +172,7 @@ def __create_ospf_global( config_data.append(cmd) + # router id router_id = ospf_data.setdefault("router_id", None) del_router_id = ospf_data.setdefault("del_router_id", False) @@ -166,6 +181,33 @@ def __create_ospf_global( if router_id: config_data.append("{} router-id {}".format(ospf, router_id)) + # log-adjacency-changes + log_adj_changes = ospf_data.setdefault("log_adj_changes", None) + del_log_adj_changes = ospf_data.setdefault("del_log_adj_changes", False) + if del_log_adj_changes: + config_data.append("no log-adjacency-changes detail") + if log_adj_changes: + config_data.append("log-adjacency-changes {}".format( + log_adj_changes)) + + # aggregation timer + aggr_timer = ospf_data.setdefault("aggr_timer", None) + del_aggr_timer = ospf_data.setdefault("del_aggr_timer", False) + if del_aggr_timer: + config_data.append("no aggregation timer") + if aggr_timer: + config_data.append("aggregation timer {}".format( + aggr_timer)) + + # maximum path information + ecmp_data = ospf_data.setdefault("maximum-paths", {}) + if ecmp_data: + cmd = "maximum-paths {}".format(ecmp_data) + del_action = ospf_data.setdefault("del_max_path", False) + if del_action: + cmd = "no maximum-paths" + config_data.append(cmd) + # redistribute command redistribute_data = ospf_data.setdefault("redistribute", {}) if redistribute_data: @@ -203,6 +245,34 @@ def __create_ospf_global( cmd = "no {}".format(cmd) config_data.append(cmd) + #def route information + def_rte_data = ospf_data.setdefault("default-information", {}) + if def_rte_data: + if "originate" not in def_rte_data: + logger.debug("Router %s: 'originate key' not present in " + "input_dict", router) + else: + cmd = "default-information originate" + + if "always" in def_rte_data: + cmd = cmd + " always" + + if "metric" in def_rte_data: + cmd = cmd + " metric {}".format(def_rte_data["metric"]) + + if "metric-type" in def_rte_data: + cmd = cmd + " metric-type {}".format(def_rte_data[ + "metric-type"]) + + if "route-map" in def_rte_data: + cmd = cmd + " route-map {}".format(def_rte_data[ + "route-map"]) + + del_action = def_rte_data.setdefault("delete", False) + if del_action: + cmd = "no {}".format(cmd) + config_data.append(cmd) + # area interface information for ospf6d only if ospf == "ospf6": area_iface = ospf_data.setdefault("neighbors", {}) @@ -217,6 +287,21 @@ def __create_ospf_global( cmd = "no {}".format(cmd) config_data.append(cmd) + try: + if "area" in input_dict[router]['links'][neighbor][ + 'ospf6']: + iface = input_dict[router]["links"][neighbor]["interface"] + cmd = "interface {} area {}".format( + iface, input_dict[router]['links'][neighbor][ + 'ospf6']['area']) + if input_dict[router]['links'][neighbor].setdefault( + "delete", False): + cmd = "no {}".format(cmd) + config_data.append(cmd) + except KeyError: + pass + + # summary information summary_data = ospf_data.setdefault("summary-address", {}) if summary_data: @@ -427,11 +512,11 @@ def config_ospf_interface(tgen, topo, input_dict=None, build=False, load_config= result = create_common_configuration( tgen, router, config_data, "interface_config", build=build ) - logger.debug("Exiting lib API: create_igmp_config()") + logger.debug("Exiting lib API: config_ospf_interface()") return result -def clear_ospf(tgen, router): +def clear_ospf(tgen, router, ospf=None): """ This API is to clear ospf neighborship by running clear ip ospf interface * command, @@ -451,11 +536,16 @@ def clear_ospf(tgen, router): return False rnode = tgen.routers()[router] - # Clearing OSPF - logger.info("Clearing ospf process for router %s..", router) + if ospf: + version = "ipv6" + else: + version = "ip" - run_frr_cmd(rnode, "clear ip ospf interface ") + cmd = "clear {} ospf interface".format(version) + logger.info( + "Clearing ospf process on router %s.. using command '%s'", router, cmd) + run_frr_cmd(rnode, cmd) logger.debug("Exiting lib API: clear_ospf()") @@ -684,65 +774,188 @@ def verify_ospf_neighbor(tgen, topo, dut=None, input_dict=None, lan=False, expec ################################ # Verification procs ################################ -@retry(attempts=40, wait=2, return_is_str=True) -def verify_ospf6_neighbor(tgen, topo, expected=True): +@retry(attempts=10, wait=2, return_is_str=True) +def verify_ospf6_neighbor(tgen, topo, dut=None, input_dict=None, lan=False): """ This API is to verify ospf neighborship by running - show ip ospf neighbour command, + show ipv6 ospf neighbour command, Parameters ---------- * `tgen` : Topogen object * `topo` : json file data - * `expected` : expected results from API, by-default True + * `dut`: device under test + * `input_dict` : Input dict data, required when configuring from testcase + * `lan` : verify neighbors in lan topology Usage ----- - Check FULL neighbors. - verify_ospf_neighbor(tgen, topo) + 1. To check FULL neighbors. + verify_ospf_neighbor(tgen, topo, dut=dut) - result = verify_ospf_neighbor(tgen, topo) + 2. To check neighbors with their roles. + input_dict = { + "r0": { + "ospf6": { + "neighbors": { + "r1": { + "state": "Full", + "role": "DR" + }, + "r2": { + "state": "Full", + "role": "DROther" + }, + "r3": { + "state": "Full", + "role": "DROther" + } + } + } + } + } + result = verify_ospf6_neighbor(tgen, topo, dut, input_dict, lan=True) Returns ------- True or False (Error Message) """ - - logger.debug("Entering lib API: verify_ospf6_neighbor()") + logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name)) result = False - for router, rnode in tgen.routers().items(): - if "ospf6" not in topo["routers"][router]: - continue - logger.info("Verifying OSPF6 neighborship on router %s:", router) - show_ospf_json = run_frr_cmd( - rnode, "show ipv6 ospf6 neighbor json", isjson=True - ) + if input_dict: + for router, rnode in tgen.routers().items(): + if 'ospf6' not in topo['routers'][router]: + continue - if not show_ospf_json: - return "OSPF6 is not running" - - ospf_nbr_list = topo["routers"][router]["ospf6"]["neighbors"] - no_of_peer = 0 - for ospf_nbr in ospf_nbr_list: - ospf_nbr_rid = topo["routers"][ospf_nbr]["ospf6"]["router_id"] - for neighbor in show_ospf_json["neighbors"]: - if neighbor["neighborId"] == ospf_nbr_rid: - nh_state = neighbor["state"] - break - else: - return "[DUT: {}] OSPF6 peer {} missing".format(router, ospf_nbr_rid) + if dut is not None and dut != router: + continue + + logger.info("Verifying OSPF neighborship on router %s:", router) + show_ospf_json = run_frr_cmd(rnode, + "show ipv6 ospf neighbor json", isjson=True) + # Verifying output dictionary show_ospf_json is empty or not + if not bool(show_ospf_json): + errormsg = "OSPF6 is not running" + return errormsg + + ospf_data_list = input_dict[router]["ospf6"] + ospf_nbr_list = ospf_data_list['neighbors'] + + for ospf_nbr, nbr_data in ospf_nbr_list.items(): + data_ip = data_rid = topo['routers'][ospf_nbr]['ospf6']['router_id'] + if ospf_nbr in data_ip: + nbr_details = nbr_data[ospf_nbr] + elif lan: + for switch in topo['switches']: + if 'ospf6' in topo['switches'][switch]['links'][router]: + neighbor_ip = data_ip + else: + continue + else: + neighbor_ip = data_ip[router]['ipv6'].split("/")[0] + + nh_state = None + neighbor_ip = neighbor_ip.lower() + nbr_rid = data_rid + get_index_val = dict((d['neighborId'], dict( \ + d, index=index)) for (index, d) in enumerate( \ + show_ospf_json['neighbors'])) + try: + nh_state = get_index_val.get(neighbor_ip)['state'] + intf_state = get_index_val.get(neighbor_ip)['ifState'] + except TypeError: + errormsg = "[DUT: {}] OSPF peer {} missing,from "\ + "{} ".format(router, + nbr_rid, ospf_nbr) + return errormsg + + nbr_state = nbr_data.setdefault("state",None) + nbr_role = nbr_data.setdefault("role",None) + + if nbr_state: + if nbr_state == nh_state: + logger.info("[DUT: {}] OSPF6 Nbr is {}:{} State {}".format + (router, ospf_nbr, nbr_rid, nh_state)) + result = True + else: + errormsg = ("[DUT: {}] OSPF6 is not Converged, neighbor" + " state is {} , Expected state is {}".format(router, + nh_state, nbr_state)) + return errormsg + if nbr_role: + if nbr_role == intf_state: + logger.info("[DUT: {}] OSPF6 Nbr is {}: {} Role {}".format( + router, ospf_nbr, nbr_rid, nbr_role)) + else: + errormsg = ("[DUT: {}] OSPF6 is not Converged with rid" + "{}, role is {}, Expected role is {}".format(router, + nbr_rid, intf_state, nbr_role)) + return errormsg + continue + else: + + for router, rnode in tgen.routers().items(): + if 'ospf6' not in topo['routers'][router]: + continue + + if dut is not None and dut != router: + continue + + logger.info("Verifying OSPF6 neighborship on router %s:", router) + show_ospf_json = run_frr_cmd(rnode, + "show ipv6 ospf neighbor json", isjson=True) + # Verifying output dictionary show_ospf_json is empty or not + if not bool(show_ospf_json): + errormsg = "OSPF6 is not running" + return errormsg + + ospf_data_list = topo["routers"][router]["ospf6"] + ospf_neighbors = ospf_data_list['neighbors'] + total_peer = 0 + total_peer = len(ospf_neighbors.keys()) + no_of_ospf_nbr = 0 + ospf_nbr_list = ospf_data_list['neighbors'] + no_of_peer = 0 + for ospf_nbr, nbr_data in ospf_nbr_list.items(): + data_ip = data_rid = topo['routers'][ospf_nbr]['ospf6']['router_id'] + if ospf_nbr in data_ip: + nbr_details = nbr_data[ospf_nbr] + elif lan: + for switch in topo['switches']: + if 'ospf6' in topo['switches'][switch]['links'][router]: + neighbor_ip = data_ip + else: + continue + else: + neighbor_ip = data_ip + + nh_state = None + neighbor_ip = neighbor_ip.lower() + nbr_rid = data_rid + get_index_val = dict((d['neighborId'], dict( \ + d, index=index)) for (index, d) in enumerate( \ + show_ospf_json['neighbors'])) + try: + nh_state = get_index_val.get(neighbor_ip)['state'] + intf_state = get_index_val.get(neighbor_ip)['ifState'] + except TypeError: + errormsg = "[DUT: {}] OSPF peer {} missing,from "\ + "{} ".format(router, + nbr_rid, ospf_nbr) + return errormsg - if nh_state == "Full": - no_of_peer += 1 + if nh_state == 'Full': + no_of_peer += 1 - if no_of_peer == len(ospf_nbr_list): - logger.info("[DUT: {}] OSPF6 is Converged".format(router)) - result = True - else: - return "[DUT: {}] OSPF6 is not Converged".format(router) + if no_of_peer == total_peer: + logger.info("[DUT: {}] OSPF6 is Converged".format(router)) + result = True + else: + errormsg = ("[DUT: {}] OSPF6 is not Converged".format(router)) + return errormsg - logger.debug("Exiting API: verify_ospf6_neighbor()") + logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name)) return result @@ -1355,3 +1568,667 @@ def verify_ospf_summary(tgen, topo, dut, input_dict, expected=True): logger.debug("Exiting API: verify_ospf_summary()") return result + + + +@retry(attempts=10, wait=3, return_is_str=True) +def verify_ospf6_rib(tgen, dut, input_dict, next_hop=None, + tag=None, metric=None, fib=None): + """ + This API is to verify ospf routes by running + show ip ospf route command. + + Parameters + ---------- + * `tgen` : Topogen object + * `dut`: device under test + * `input_dict` : Input dict data, required when configuring from testcase + * `next_hop` : next to be verified + * `tag` : tag to be verified + * `metric` : metric to be verified + * `fib` : True if the route is installed in FIB. + + Usage + ----- + input_dict = { + "r1": { + "static_routes": [ + { + "network": ip_net, + "no_of_ip": 1, + "routeType": "N" + } + ] + } + } + + result = verify_ospf6_rib(tgen, dut, input_dict,next_hop=nh) + + Returns + ------- + True or False (Error Message) + """ + + logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name)) + result = False + router_list = tgen.routers() + additional_nexthops_in_required_nhs = [] + found_hops = [] + for routerInput in input_dict.keys(): + for router, rnode in router_list.iteritems(): + if router != dut: + continue + + logger.info("Checking router %s RIB:", router) + + # Verifying RIB routes + command = "show ipv6 ospf route" + + found_routes = [] + missing_routes = [] + + if "static_routes" in input_dict[routerInput] or \ + "prefix" in input_dict[routerInput]: + if "prefix" in input_dict[routerInput]: + static_routes = input_dict[routerInput]["prefix"] + else: + static_routes = input_dict[routerInput]["static_routes"] + + + for static_route in static_routes: + cmd = "{}".format(command) + + cmd = "{} json".format(cmd) + + ospf_rib_json = run_frr_cmd(rnode, cmd, isjson=True) + + # Fix for PR 2644182 + try: + ospf_rib_json = ospf_rib_json['routes'] + except KeyError: + pass + + # Verifying output dictionary ospf_rib_json is not empty + if bool(ospf_rib_json) is False: + errormsg = "[DUT: {}] No routes found in OSPF6 route " \ + "table".format(router) + return errormsg + + network = static_route["network"] + no_of_ip = static_route.setdefault("no_of_ip", 1) + _tag = static_route.setdefault("tag", None) + _rtype = static_route.setdefault("routeType", None) + + + # Generating IPs for verification + ip_list = generate_ips(network, no_of_ip) + st_found = False + nh_found = False + for st_rt in ip_list: + st_rt = str(ipaddress.ip_network(frr_unicode(st_rt))) + + _addr_type = validate_ip_address(st_rt) + if _addr_type != 'ipv6': + continue + + if st_rt in ospf_rib_json: + + st_found = True + found_routes.append(st_rt) + + if fib and next_hop: + if type(next_hop) is not list: + next_hop = [next_hop] + + for mnh in range(0, len(ospf_rib_json[st_rt])): + if 'fib' in ospf_rib_json[st_rt][ + mnh]["nextHops"][0]: + found_hops.append([rib_r[ + "ip"] for rib_r in ospf_rib_json[ + st_rt][mnh]["nextHops"]]) + + if found_hops[0]: + missing_list_of_nexthops = \ + set(found_hops[0]).difference(next_hop) + additional_nexthops_in_required_nhs = \ + set(next_hop).difference(found_hops[0]) + + if additional_nexthops_in_required_nhs: + logger.info( + "Nexthop " + "%s is not active for route %s in " + "RIB of router %s\n", + additional_nexthops_in_required_nhs, + st_rt, dut) + errormsg = ( + "Nexthop {} is not active" + " for route {} in RIB of router" + " {}\n".format( + additional_nexthops_in_required_nhs, + st_rt, dut)) + return errormsg + else: + nh_found = True + + elif next_hop and fib is None: + if type(next_hop) is not list: + next_hop = [next_hop] + found_hops = [rib_r['nextHop'] for rib_r in + ospf_rib_json[st_rt][ + "nextHops"]] + + if found_hops: + missing_list_of_nexthops = \ + set(found_hops).difference(next_hop) + additional_nexthops_in_required_nhs = \ + set(next_hop).difference(found_hops) + if additional_nexthops_in_required_nhs: + logger.info( + "Missing nexthop %s for route"\ + " %s in RIB of router %s\n", \ + additional_nexthops_in_required_nhs, \ + st_rt, dut) + errormsg=("Nexthop {} is Missing for "\ + "route {} in RIB of router {}\n".format( + additional_nexthops_in_required_nhs, + st_rt, dut)) + return errormsg + else: + nh_found = True + if _rtype: + if "destinationType" not in ospf_rib_json[ + st_rt]: + errormsg = ("[DUT: {}]: destinationType missing" + "for route {} in OSPF RIB \n".\ + format(dut, st_rt)) + return errormsg + elif _rtype != ospf_rib_json[st_rt][ + "destinationType"]: + errormsg = ("[DUT: {}]: destinationType mismatch" + "for route {} in OSPF RIB \n".\ + format(dut, st_rt)) + return errormsg + else: + logger.info("DUT: {}]: Found destinationType {}" + "for route {}".\ + format(dut, _rtype, st_rt)) + if tag: + if "tag" not in ospf_rib_json[ + st_rt]: + errormsg = ("[DUT: {}]: tag is not" + " present for" + " route {} in RIB \n".\ + format(dut, st_rt + )) + return errormsg + + if _tag != ospf_rib_json[ + st_rt]["tag"]: + errormsg = ("[DUT: {}]: tag value {}" + " is not matched for" + " route {} in RIB \n".\ + format(dut, _tag, st_rt, + )) + return errormsg + + if metric is not None: + if "type2cost" not in ospf_rib_json[ + st_rt]: + errormsg = ("[DUT: {}]: metric is" + " not present for" + " route {} in RIB \n".\ + format(dut, st_rt)) + return errormsg + + if metric != ospf_rib_json[ + st_rt]["type2cost"]: + errormsg = ("[DUT: {}]: metric value " + "{} is not matched for " + "route {} in RIB \n".\ + format(dut, metric, st_rt, + )) + return errormsg + + else: + missing_routes.append(st_rt) + + if nh_found: + logger.info("[DUT: {}]: Found next_hop {} for all OSPF" + " routes in RIB".format(router, next_hop)) + + if len(missing_routes) > 0: + errormsg = ("[DUT: {}]: Missing route in RIB, " + "routes: {}".\ + format(dut, missing_routes)) + return errormsg + + if found_routes: + logger.info("[DUT: %s]: Verified routes in RIB, found" + " routes are: %s\n", dut, found_routes) + result = True + + logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name)) + return result + + +@retry(attempts=3, wait=2, return_is_str=True) +def verify_ospf6_interface(tgen, topo, dut=None,lan=False, input_dict=None): + """ + This API is to verify ospf routes by running + show ip ospf interface command. + + Parameters + ---------- + * `tgen` : Topogen object + * `topo` : topology descriptions + * `dut`: device under test + * `lan`: if set to true this interface belongs to LAN. + * `input_dict` : Input dict data, required when configuring from testcase + + Usage + ----- + input_dict= { + 'r0': { + 'links':{ + 's1': { + 'ospf6':{ + 'priority':98, + 'timerDeadSecs': 4, + 'area': '0.0.0.3', + 'mcastMemberOspfDesignatedRouters': True, + 'mcastMemberOspfAllRouters': True, + 'ospfEnabled': True, + + } + } + } + } + } + result = verify_ospf_interface(tgen, topo, dut=dut, input_dict=input_dict) + + Returns + ------- + True or False (Error Message) + """ + + logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name)) + result = False + + for router, rnode in tgen.routers().iteritems(): + if 'ospf6' not in topo['routers'][router]: + continue + + if dut is not None and dut != router: + continue + + logger.info("Verifying OSPF interface on router %s:", router) + show_ospf_json = run_frr_cmd(rnode, "show ipv6 ospf interface json", + isjson=True) + + # Verifying output dictionary show_ospf_json is empty or not + if not bool(show_ospf_json): + errormsg = "OSPF6 is not running" + return errormsg + + # To find neighbor ip type + ospf_intf_data = input_dict[router]["links"] + for ospf_intf, intf_data in ospf_intf_data.items(): + intf = topo['routers'][router]['links'][ospf_intf]['interface'] + if intf in show_ospf_json: + for intf_attribute in intf_data['ospf6']: + if intf_data['ospf6'][intf_attribute] is not list: + if intf_data['ospf6'][intf_attribute] == show_ospf_json[ + intf][intf_attribute]: + logger.info("[DUT: %s] OSPF6 interface %s: %s is %s", + router, intf, intf_attribute, intf_data['ospf6'][ + intf_attribute]) + elif intf_data['ospf6'][intf_attribute] is list: + for addr_list in len(show_ospf_json[intf][intf_attribute]): + if show_ospf_json[intf][intf_attribute][addr_list][ + 'address'].split('/')[0] == intf_data['ospf6'][ + 'internetAddress'][0]['address']: + break + else: + errormsg= "[DUT: {}] OSPF6 interface {}: {} is {}, \ + Expected is {}".format(router, intf, intf_attribute, + intf_data['ospf6'][intf_attribute], intf_data['ospf6'][ + intf_attribute]) + return errormsg + else: + errormsg= "[DUT: {}] OSPF6 interface {}: {} is {}, \ + Expected is {}".format(router, intf, intf_attribute, + intf_data['ospf6'][intf_attribute], intf_data['ospf6'][ + intf_attribute]) + return errormsg + result = True + logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name)) + return result + + +@retry(attempts=11, wait=2, return_is_str=True) +def verify_ospf6_database(tgen, topo, dut, input_dict): + """ + This API is to verify ospf lsa's by running + show ip ospf database command. + + Parameters + ---------- + * `tgen` : Topogen object + * `dut`: device under test + * `input_dict` : Input dict data, required when configuring from testcase + * `topo` : next to be verified + + Usage + ----- + input_dict = { + "areas": { + "0.0.0.0": { + "routerLinkStates": { + "100.1.1.0-100.1.1.0": { + "LSID": "100.1.1.0", + "Advertised router": "100.1.1.0", + "LSA Age": 130, + "Sequence Number": "80000006", + "Checksum": "a703", + "Router links": 3 + } + }, + "networkLinkStates": { + "10.0.0.2-100.1.1.1": { + "LSID": "10.0.0.2", + "Advertised router": "100.1.1.1", + "LSA Age": 137, + "Sequence Number": "80000001", + "Checksum": "9583" + } + }, + }, + } + } + result = verify_ospf_database(tgen, topo, dut, input_dict) + + Returns + ------- + True or False (Error Message) + """ + + result = False + router = dut + logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name)) + + if 'ospf' not in topo['routers'][dut]: + errormsg = "[DUT: {}] OSPF is not configured on the router.".format( + dut) + return errormsg + + rnode = tgen.routers()[dut] + + logger.info("Verifying OSPF interface on router %s:", dut) + show_ospf_json = run_frr_cmd(rnode, "show ip ospf database json", + isjson=True) + # Verifying output dictionary show_ospf_json is empty or not + if not bool(show_ospf_json): + errormsg = "OSPF is not running" + return errormsg + + # for inter and inter lsa's + ospf_db_data = input_dict.setdefault("areas", None) + ospf_external_lsa = input_dict.setdefault( + 'asExternalLinkStates', None) + + if ospf_db_data: + for ospf_area, area_lsa in ospf_db_data.items(): + if ospf_area in show_ospf_json['areas']: + if 'routerLinkStates' in area_lsa: + for lsa in area_lsa['routerLinkStates']: + for rtrlsa in show_ospf_json['areas'][ospf_area][ + 'routerLinkStates']: + if lsa['lsaId'] == rtrlsa['lsaId'] and \ + lsa['advertisedRouter'] == rtrlsa[ + 'advertisedRouter']: + result = True + break + if result: + logger.info( + "[DUT: %s] OSPF LSDB area %s:Router " + "LSA %s", router, ospf_area, lsa) + break + else: + errormsg = \ + "[DUT: {}] OSPF LSDB area {}: expected" \ + " Router LSA is {}".format(router, ospf_area, lsa) + return errormsg + + if 'networkLinkStates' in area_lsa: + for lsa in area_lsa['networkLinkStates']: + for netlsa in show_ospf_json['areas'][ospf_area][ + 'networkLinkStates']: + if lsa in show_ospf_json['areas'][ospf_area][ + 'networkLinkStates']: + if lsa['lsaId'] == netlsa['lsaId'] and \ + lsa['advertisedRouter'] == netlsa[ + 'advertisedRouter']: + result = True + break + if result: + logger.info( + "[DUT: %s] OSPF LSDB area %s:Network " + "LSA %s", router, ospf_area, lsa) + break + else: + errormsg = \ + "[DUT: {}] OSPF LSDB area {}: expected" \ + " Network LSA is {}".format(router, ospf_area, lsa) + return errormsg + + if 'summaryLinkStates' in area_lsa: + for lsa in area_lsa['summaryLinkStates']: + for t3lsa in show_ospf_json['areas'][ospf_area][ + 'summaryLinkStates']: + if lsa['lsaId'] == t3lsa['lsaId'] and \ + lsa['advertisedRouter'] == t3lsa[ + 'advertisedRouter']: + result = True + break + if result: + logger.info( + "[DUT: %s] OSPF LSDB area %s:Summary " + "LSA %s", router, ospf_area, lsa) + break + else: + errormsg = \ + "[DUT: {}] OSPF LSDB area {}: expected" \ + " Summary LSA is {}".format(router, ospf_area, lsa) + return errormsg + + if 'nssaExternalLinkStates' in area_lsa: + for lsa in area_lsa['nssaExternalLinkStates']: + for t7lsa in show_ospf_json['areas'][ospf_area][ + 'nssaExternalLinkStates']: + if lsa['lsaId'] == t7lsa['lsaId'] and \ + lsa['advertisedRouter'] == t7lsa[ + 'advertisedRouter']: + result = True + break + if result: + logger.info( + "[DUT: %s] OSPF LSDB area %s:Type7 " + "LSA %s", router, ospf_area, lsa) + break + else: + errormsg = \ + "[DUT: {}] OSPF LSDB area {}: expected" \ + " Type7 LSA is {}".format(router, ospf_area, lsa) + return errormsg + + if 'asbrSummaryLinkStates' in area_lsa: + for lsa in area_lsa['asbrSummaryLinkStates']: + for t4lsa in show_ospf_json['areas'][ospf_area][ + 'asbrSummaryLinkStates']: + if lsa['lsaId'] == t4lsa['lsaId'] and \ + lsa['advertisedRouter'] == t4lsa[ + 'advertisedRouter']: + result = True + break + if result: + logger.info( + "[DUT: %s] OSPF LSDB area %s:ASBR Summary " + "LSA %s", router, ospf_area, lsa) + result = True + else: + errormsg = \ + "[DUT: {}] OSPF LSDB area {}: expected" \ + " ASBR Summary LSA is {}".format( + router, ospf_area, lsa) + return errormsg + + if 'linkLocalOpaqueLsa' in area_lsa: + for lsa in area_lsa['linkLocalOpaqueLsa']: + try: + for lnklsa in show_ospf_json['areas'][ospf_area][ + 'linkLocalOpaqueLsa']: + if lsa['lsaId'] in lnklsa['lsaId'] and \ + 'linkLocalOpaqueLsa' in show_ospf_json[ + 'areas'][ospf_area]: + logger.info(( + "[DUT: FRR] OSPF LSDB area %s:Opaque-LSA" + "%s", ospf_area, lsa)) + result = True + else: + errormsg = ("[DUT: FRR] OSPF LSDB area: {} " + "expected Opaque-LSA is {}, Found is {}".format( + ospf_area, lsa, show_ospf_json)) + raise ValueError (errormsg) + return errormsg + except KeyError: + errormsg = ("[DUT: FRR] linkLocalOpaqueLsa Not " + "present") + return errormsg + + if ospf_external_lsa: + for lsa in ospf_external_lsa: + try: + for t5lsa in show_ospf_json['asExternalLinkStates']: + if lsa['lsaId'] == t5lsa['lsaId'] and \ + lsa['advertisedRouter'] == t5lsa[ + 'advertisedRouter']: + result = True + break + except KeyError: + result = False + if result: + logger.info( + "[DUT: %s] OSPF LSDB:External LSA %s", + router, lsa) + result = True + else: + errormsg = \ + "[DUT: {}] OSPF LSDB : expected" \ + " External LSA is {}".format(router, lsa) + return errormsg + + logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name)) + return result + + + +def config_ospf6_interface (tgen, topo, input_dict=None, build=False, + load_config=True): + """ + API to configure ospf on router. + + Parameters + ---------- + * `tgen` : Topogen object + * `topo` : json file data + * `input_dict` : Input dict data, required when configuring from testcase + * `build` : Only for initial setup phase this is set as True. + * `load_config` : Loading the config to router this is set as True. + + Usage + ----- + r1_ospf_auth = { + "r1": { + "links": { + "r2": { + "ospf": { + "authentication": 'message-digest', + "authentication-key": "ospf", + "message-digest-key": "10" + } + } + } + } + } + result = config_ospf6_interface(tgen, topo, r1_ospf_auth) + + Returns + ------- + True or False + """ + logger.debug("Entering lib API: {}".format(sys._getframe().f_code.co_name)) + result = False + if not input_dict: + input_dict = deepcopy(topo) + else: + input_dict = deepcopy(input_dict) + for router in input_dict.keys(): + config_data = [] + for lnk in input_dict[router]['links'].keys(): + if "ospf6" not in input_dict[router]['links'][lnk]: + logger.debug("Router %s: ospf6 configs is not present in" + "input_dict, passed input_dict", router, + input_dict) + continue + ospf_data = input_dict[router]['links'][lnk]['ospf6'] + data_ospf_area = ospf_data.setdefault("area", None) + data_ospf_auth = ospf_data.setdefault("authentication", None) + data_ospf_dr_priority = ospf_data.setdefault("priority", None) + data_ospf_cost = ospf_data.setdefault("cost", None) + data_ospf_mtu = ospf_data.setdefault("mtu_ignore", None) + + try: + intf = topo['routers'][router]['links'][lnk]['interface'] + except KeyError: + intf = topo['switches'][router]['links'][lnk]['interface'] + + # interface + cmd = "interface {}".format(intf) + + config_data.append(cmd) + # interface area config + if data_ospf_area: + cmd = "ipv6 ospf area {}".format(data_ospf_area) + config_data.append(cmd) + + # interface ospf dr priority + if data_ospf_dr_priority: + cmd = "ipv6 ospf priority {}".format( + ospf_data["priority"]) + if 'del_action' in ospf_data: + cmd = "no {}".format(cmd) + config_data.append(cmd) + + # interface ospf cost + if data_ospf_cost: + cmd = "ipv6 ospf cost {}".format( + ospf_data["cost"]) + if 'del_action' in ospf_data: + cmd = "no {}".format(cmd) + config_data.append(cmd) + + # interface ospf mtu + if data_ospf_mtu: + cmd = "ipv6 ospf mtu-ignore" + if 'del_action' in ospf_data: + cmd = "no {}".format(cmd) + config_data.append(cmd) + + if build: + return config_data + else: + result = create_common_configuration(tgen, router, config_data, + "interface_config", + build=build) + logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name)) + return result diff --git a/tests/topotests/ospfv3_basic_functionality/ospfv3_rte_calc.json b/tests/topotests/ospfv3_basic_functionality/ospfv3_rte_calc.json new file mode 100644 index 0000000000..3669b3a554 --- /dev/null +++ b/tests/topotests/ospfv3_basic_functionality/ospfv3_rte_calc.json @@ -0,0 +1,173 @@ +{ + "feature": [ + "bgp" + ], + "address_types": [ + "ipv6" + ], + "ipv6base": "fd00::", + "ipv6mask": 64, + "link_ip_start": { + "ipv6": "fd00::", + "v6mask": 64 + }, + "lo_prefix": { + "ipv6": "2001:db8:f::", + "v6mask": 128 + }, + "routers": { + "r0": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r1": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r2": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + } + }, + "ospf6": { + "router_id": "100.1.1.0", + "neighbors": { + "r1": {}, + "r2": {} + } + } + }, + "r1": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r0": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r2": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + } + }, + "ospf6": { + "router_id": "100.1.1.1", + "neighbors": { + "r0": {}, + "r2": {}, + "r3": {} + } + } + }, + "r2": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r0": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r1": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + } + }, + "ospf6": { + "router_id": "100.1.1.2", + "neighbors": { + "r1": {}, + "r0": {}, + "r3": {} + } + } + }, + "r3": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r0": { + "ipv6": "auto" + }, + "r1": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r2": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + } + }, + "ospf6": { + "router_id": "100.1.1.3", + "neighbors": { + "r1": {}, + "r2": {} + } + } + } + } +} \ No newline at end of file diff --git a/tests/topotests/ospfv3_basic_functionality/ospfv3_single_area.json b/tests/topotests/ospfv3_basic_functionality/ospfv3_single_area.json new file mode 100644 index 0000000000..d93eb1f217 --- /dev/null +++ b/tests/topotests/ospfv3_basic_functionality/ospfv3_single_area.json @@ -0,0 +1,190 @@ +{ + "address_types": [ + "ipv6" + ], + "ipv6base": "fd00::", + "ipv6mask": 64, + "link_ip_start": { + "ipv6": "fd00::", + "v6mask": 64 + }, + "lo_prefix": { + "ipv6": "2001:db8:f::", + "v6mask": 128 + }, + "routers": { + "r0": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r1": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r2": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4, + "network": "point-to-point" + } + } + }, + "ospf6": { + "router_id": "100.1.1.0", + "neighbors": { + "r1": {}, + "r2": {}, + "r3": {} + } + } + }, + "r1": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r0": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r2": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3-link0": { + "ipv6": "auto", + "description": "DummyIntftoR3" + } + }, + "ospf6": { + "router_id": "100.1.1.1", + "neighbors": { + "r0": {}, + "r2": {}, + "r3": {} + } + } + }, + "r2": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r0": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r1": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r3": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + } + }, + "ospf6": { + "router_id": "100.1.1.2", + "neighbors": { + "r1": {}, + "r0": {}, + "r3": {} + } + } + }, + "r3": { + "links": { + "lo": { + "ipv6": "auto", + "type": "loopback" + }, + "r0": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4, + "network": "point-to-point" + } + }, + "r1": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r2": { + "ipv6": "auto", + "ospf6": { + "area": "0.0.0.0", + "hello_interval": 1, + "dead_interval": 4 + } + }, + "r1-link0": { + "ipv6": "auto", + "description": "DummyIntftoR1", + "ospf6": { + "area": "0.0.0.0" + } + } + }, + "ospf6": { + "router_id": "1.0.4.17", + "neighbors": { + "r0": {}, + "r1": {}, + "r2": {} + } + } + } + } +} \ No newline at end of file diff --git a/tests/topotests/ospfv3_basic_functionality/test_ospfv3_rte_calc.py b/tests/topotests/ospfv3_basic_functionality/test_ospfv3_rte_calc.py new file mode 100644 index 0000000000..4aa71bfb16 --- /dev/null +++ b/tests/topotests/ospfv3_basic_functionality/test_ospfv3_rte_calc.py @@ -0,0 +1,374 @@ +#!/usr/bin/python +# +# Copyright (c) 2021 by VMware, Inc. ("VMware") +# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. +# ("NetDEF") in this file. +# +# Permission to use, copy, modify, and/or distribute this software +# for any purpose with or without fee is hereby granted, provided +# that the above copyright notice and this permission notice appear +# in all copies. +# +# THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES +# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF +# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR +# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY +# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, +# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS +# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE +# OF THIS SOFTWARE. +# + + +"""OSPF Basic Functionality Automation.""" +import os +import sys +import time +import pytest +import json +from copy import deepcopy +from ipaddress import IPv4Address +from lib.topotest import frr_unicode + +# Save the Current Working Directory to find configuration files. +CWD = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(os.path.join(CWD, "../")) +sys.path.append(os.path.join(CWD, "../lib/")) + +# pylint: disable=C0413 +# Import topogen and topotest helpers +from mininet.topo import Topo +from lib.topogen import Topogen, get_topogen +import ipaddress +from lib.bgp import verify_bgp_convergence, create_router_bgp + +# Import topoJson from lib, to create topology and initial configuration +from lib.common_config import ( + start_topology, + write_test_header, + write_test_footer, + reset_config_on_routers, + verify_rib, + create_static_routes, + step, + create_route_maps, + shutdown_bringup_interface, + create_interfaces_cfg, + topo_daemons, + get_frr_ipv6_linklocal, +) + +from lib.topolog import logger +from lib.topojson import build_topo_from_json, build_config_from_json + +from lib.ospf import ( + verify_ospf6_neighbor, + config_ospf_interface, + clear_ospf, + verify_ospf6_rib, + create_router_ospf, + verify_ospf6_interface, + verify_ospf6_database, + config_ospf6_interface, +) + +from ipaddress import IPv6Address + +# Global variables +topo = None + +# Reading the data from JSON File for topology creation +jsonFile = "{}/ospfv3_rte_calc.json".format(CWD) +try: + with open(jsonFile, "r") as topoJson: + topo = json.load(topoJson) +except IOError: + assert False, "Could not read file {}".format(jsonFile) + +NETWORK = { + "ipv6": [ + "11.0.20.1/32", + "11.0.20.2/32", + "11.0.20.3/32", + "11.0.20.4/32", + "11.0.20.5/32", + ], + "ipv6": ["2::1/128", "2::2/128", "2::3/128", "2::4/128", "2::5/128"], +} +TOPOOLOGY = """ + Please view in a fixed-width font such as Courier. + +---+ A1 +---+ + +R1 +------------+R2 | + +-+-+- +--++ + | -- -- | + | -- A0 -- | + A0| ---- | + | ---- | A2 + | -- -- | + | -- -- | + +-+-+- +-+-+ + +R0 +-------------+R3 | + +---+ A3 +---+ +""" + +TESTCASES = """ +1. OSPF Cost - verifying ospf interface cost functionality +""" + + +class CreateTopo(Topo): + """ + Test topology builder. + + * `Topo`: Topology object + """ + + def build(self, *_args, **_opts): + """Build function.""" + tgen = get_topogen(self) + + # Building topology from json file + build_topo_from_json(tgen, topo) + + +def setup_module(mod): + """ + Sets up the pytest environment + + * `mod`: module name + """ + global topo + testsuite_run_time = time.asctime(time.localtime(time.time())) + logger.info("Testsuite start time: {}".format(testsuite_run_time)) + logger.info("=" * 40) + + logger.info("Running setup_module to create topology") + + # This function initiates the topology build with Topogen... + tgen = Topogen(CreateTopo, mod.__name__) + # ... and here it calls Mininet initialization functions. + + # get list of daemons needs to be started for this suite. + daemons = topo_daemons(tgen, topo) + + # Starting topology, create tmp files which are loaded to routers + # to start deamons and then start routers + start_topology(tgen, daemons) + + # Creating configuration from JSON + build_config_from_json(tgen, topo) + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + ospf_covergence = verify_ospf6_neighbor(tgen, topo) + assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( + ospf_covergence + ) + + logger.info("Running setup_module() done") + + +def teardown_module(mod): + """ + Teardown the pytest environment. + + * `mod`: module name + """ + + logger.info("Running teardown_module to delete topology") + + tgen = get_topogen() + + # Stop toplogy and Remove tmp files + tgen.stop_topology() + + logger.info( + "Testsuite end time: {}".format(time.asctime(time.localtime(time.time()))) + ) + logger.info("=" * 40) + + +def get_llip(onrouter, intf): + """ + API to get the link local ipv6 address of a perticular interface + + Parameters + ---------- + * `fromnode`: Source node + * `tonode` : interface for which link local ip needs to be returned. + + Usage + ----- + result = get_llip('r1', 'r2-link0') + + Returns + ------- + 1) link local ipv6 address from the interface. + 2) errormsg - when link local ip not found. + """ + tgen = get_topogen() + intf = topo["routers"][onrouter]["links"][intf]["interface"] + llip = get_frr_ipv6_linklocal(tgen, onrouter, intf) + if llip: + logger.info("llip ipv6 address to be set as NH is %s", llip) + return llip + return None + + +def get_glipv6(onrouter, intf): + """ + API to get the global ipv6 address of a perticular interface + + Parameters + ---------- + * `onrouter`: Source node + * `intf` : interface for which link local ip needs to be returned. + + Usage + ----- + result = get_glipv6('r1', 'r2-link0') + + Returns + ------- + 1) global ipv6 address from the interface. + 2) errormsg - when link local ip not found. + """ + glipv6 = (topo["routers"][onrouter]["links"][intf]["ipv6"]).split("/")[0] + if glipv6: + logger.info("Global ipv6 address to be set as NH is %s", glipv6) + return glipv6 + return None + + +def red_static(dut, config=True): + """Local def for Redstribute static routes inside ospf.""" + global topo + tgen = get_topogen() + if config: + ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "static"}]}}} + else: + ospf_red = { + dut: { + "ospf6": { + "redistribute": [{"redist_type": "static", "del_action": True}] + } + } + } + result = create_router_ospf(tgen, topo, ospf_red) + assert result is True, "Testcase : Failed \n Error: {}".format(result) + + +def red_connected(dut, config=True): + """Local def for Redstribute connected routes inside ospf.""" + global topo + tgen = get_topogen() + if config: + ospf_red = {dut: {"ospf6": {"redistribute": [{"redist_type": "connected"}]}}} + else: + ospf_red = { + dut: { + "ospf6": { + "redistribute": [{"redist_type": "connected", "del_action": True}] + } + } + } + result = create_router_ospf(tgen, topo, ospf_red) + assert result is True, "Testcase: Failed \n Error: {}".format(result) + + +# ################################## +# Test cases start here. +# ################################## +def test_ospfv3_cost_tc52_p0(request): + """OSPF Cost - verifying ospf interface cost functionality""" + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + global topo + step("Bring up the base config.") + reset_config_on_routers(tgen) + + step( + "Configure ospf cost as 20 on interface between R0 and R1. " + "Configure ospf cost as 30 between interface between R0 and R2." + ) + + r0_ospf_cost = { + "r0": {"links": {"r1": {"ospf6": {"cost": 20}}, "r2": {"ospf6": {"cost": 30}}}} + } + result = config_ospf6_interface(tgen, topo, r0_ospf_cost) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + step( + "Verify that cost is updated in the ospf interface between" + " r0 and r1 as 30 and r0 and r2 as 20" + ) + dut = "r0" + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=r0_ospf_cost) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step( + "Swap the costs between interfaces on r0, between r0 and r1 to 30" + ", r0 and r2 to 20" + ) + + r0_ospf_cost = { + "r0": {"links": {"r1": {"ospf6": {"cost": 30}}, "r2": {"ospf6": {"cost": 20}}}} + } + result = config_ospf6_interface(tgen, topo, r0_ospf_cost) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + step( + "Verify that cost is updated in the ospf interface between r0 " + "and r1 as 30 and r0 and r2 as 20." + ) + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=r0_ospf_cost) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step(" Un configure cost from the interface r0 - r1.") + + r0_ospf_cost = { + "r0": {"links": {"r1": {"ospf6": {"cost": 30, "del_action": True}}}} + } + result = config_ospf6_interface(tgen, topo, r0_ospf_cost) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + input_dict = { + "r0": {"links": {"r1": {"ospf6": {"cost": 10}}, "r2": {"ospf6": {"cost": 20}}}} + } + step( + "Verify that cost is updated in the ospf interface between r0" + " and r1 as 10 and r0 and r2 as 20." + ) + + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step(" Un configure cost from the interface r0 - r2.") + + r0_ospf_cost = { + "r0": {"links": {"r2": {"ospf6": {"cost": 20, "del_action": True}}}} + } + result = config_ospf6_interface(tgen, topo, r0_ospf_cost) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + step( + "Verify that cost is updated in the ospf interface between r0" + "and r1 as 10 and r0 and r2 as 10" + ) + + input_dict = { + "r0": {"links": {"r1": {"ospf6": {"cost": 10}}, "r2": {"ospf6": {"cost": 10}}}} + } + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + write_test_footer(tc_name) + + + +if __name__ == "__main__": + args = ["-s"] + sys.argv[1:] + sys.exit(pytest.main(args)) diff --git a/tests/topotests/ospfv3_basic_functionality/test_ospfv3_single_area.py b/tests/topotests/ospfv3_basic_functionality/test_ospfv3_single_area.py new file mode 100644 index 0000000000..a84f1a1eb6 --- /dev/null +++ b/tests/topotests/ospfv3_basic_functionality/test_ospfv3_single_area.py @@ -0,0 +1,417 @@ +#!/usr/bin/python + +# +# Copyright (c) 2021 by VMware, Inc. ("VMware") +# Used Copyright (c) 2018 by Network Device Education Foundation, Inc. +# ("NetDEF") in this file. +# +# Permission to use, copy, modify, and/or distribute this software +# for any purpose with or without fee is hereby granted, provided +# that the above copyright notice and this permission notice appear +# in all copies. +# +# THE SOFTWARE IS PROVIDED "AS IS" AND VMWARE DISCLAIMS ALL WARRANTIES +# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF +# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL VMWARE BE LIABLE FOR +# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY +# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, +# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS +# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE +# OF THIS SOFTWARE. +# + + +"""OSPF Basic Functionality Automation.""" +import os +import sys +import time +import pytest +import json +from copy import deepcopy +from ipaddress import IPv4Address +from lib.topotest import frr_unicode + +# Save the Current Working Directory to find configuration files. +CWD = os.path.dirname(os.path.realpath(__file__)) +sys.path.append(os.path.join(CWD, "../")) +sys.path.append(os.path.join(CWD, "../lib/")) + +# pylint: disable=C0413 +# Import topogen and topotest helpers +from mininet.topo import Topo +from lib.topogen import Topogen, get_topogen +import ipaddress + +# Import topoJson from lib, to create topology and initial configuration +from lib.common_config import ( + start_topology, + write_test_header, + write_test_footer, + reset_config_on_routers, + verify_rib, + create_static_routes, + step, + create_route_maps, + shutdown_bringup_interface, + create_interfaces_cfg, + topo_daemons, +) +from lib.topolog import logger +from lib.topojson import build_topo_from_json, build_config_from_json + +from lib.ospf import ( + verify_ospf6_neighbor, + config_ospf_interface, + clear_ospf, + verify_ospf6_rib, + create_router_ospf, + verify_ospf6_interface, + verify_ospf6_database, + config_ospf6_interface, +) + +from ipaddress import IPv6Address + +# Global variables +topo = None + +# Reading the data from JSON File for topology creation +jsonFile = "{}/ospfv3_single_area.json".format(CWD) +try: + with open(jsonFile, "r") as topoJson: + topo = json.load(topoJson) +except IOError: + assert False, "Could not read file {}".format(jsonFile) + +""" +TOPOOLOGY = + Please view in a fixed-width font such as Courier. + +---+ A1 +---+ + +R1 +------------+R2 | + +-+-+- +--++ + | -- -- | + | -- A0 -- | + A0| ---- | + | ---- | A2 + | -- -- | + | -- -- | + +-+-+- +-+-+ + +R0 +-------------+R3 | + +---+ A3 +---+ + +TESTCASES = +1. OSPF IFSM -Verify state change events on p2p network. +2. OSPF Timers - Verify OSPF interface timer hello interval functionality +3. OSPF Timers - Verify OSPF interface timer dead interval functionality +4. Verify ospf show commands with json output. +5. Verify NFSM events when ospf nbr changes with different MTU values. + """ + + +class CreateTopo(Topo): + """ + Test topology builder. + + * `Topo`: Topology object + """ + + def build(self, *_args, **_opts): + """Build function.""" + tgen = get_topogen(self) + + # Building topology from json file + build_topo_from_json(tgen, topo) + + +def setup_module(mod): + """ + Sets up the pytest environment + + * `mod`: module name + """ + global topo + testsuite_run_time = time.asctime(time.localtime(time.time())) + logger.info("Testsuite start time: {}".format(testsuite_run_time)) + logger.info("=" * 40) + + logger.info("Running setup_module to create topology") + + # This function initiates the topology build with Topogen... + tgen = Topogen(CreateTopo, mod.__name__) + # ... and here it calls Mininet initialization functions. + + # get list of daemons needs to be started for this suite. + daemons = topo_daemons(tgen, topo) + + # Starting topology, create tmp files which are loaded to routers + # to start deamons and then start routers + start_topology(tgen, daemons) + + # Creating configuration from JSON + build_config_from_json(tgen, topo) + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + ospf_covergence = verify_ospf6_neighbor(tgen, topo) + assert ospf_covergence is True, "setup_module :Failed \n Error:" " {}".format( + ospf_covergence + ) + + logger.info("Running setup_module() done") + + +def teardown_module(mod): + """ + Teardown the pytest environment. + + * `mod`: module name + """ + + logger.info("Running teardown_module to delete topology") + + tgen = get_topogen() + + # Stop toplogy and Remove tmp files + tgen.stop_topology() + + logger.info( + "Testsuite end time: {}".format(time.asctime(time.localtime(time.time()))) + ) + logger.info("=" * 40) + + +# ################################## +# Test cases start here. +# ################################## + + +def test_ospfv3_p2p_tc3_p0(request): + """OSPF IFSM -Verify state change events on p2p network.""" + tc_name = request.node.name + write_test_header(tc_name) + tgen = get_topogen() + + # Don't run this test if we have any failure. + if tgen.routers_have_failure(): + pytest.skip(tgen.errors) + + global topo + step("Bring up the base config as per the topology") + reset_config_on_routers(tgen) + step( + "Verify that OSPF is subscribed to multi cast services " + "(All SPF, all DR Routers)." + ) + step("Verify that interface is enabled in ospf.") + step("Verify that config is successful.") + dut = "r0" + input_dict = {"r0": {"links": {"r3": {"ospf6": {"ospf6Enabled": True}}}}} + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Delete the ip address") + topo1 = { + "r0": { + "links": { + "r3": { + "ipv6": topo["routers"]["r0"]["links"]["r3"]["ipv6"], + "interface": topo["routers"]["r0"]["links"]["r3"]["interface"], + "delete": True, + } + } + } + } + + result = create_interfaces_cfg(tgen, topo1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Change the ip on the R0 interface") + + topo_modify_change_ip = deepcopy(topo) + intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv6"] + topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv6"] = str( + IPv6Address(frr_unicode(intf_ip.split("/")[0])) + 3 + ) + "/{}".format(intf_ip.split("/")[1]) + + build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False) + step("Verify that interface is enabled in ospf.") + dut = "r0" + input_dict = { + "r0": { + "links": { + "r3": { + "ospf6": { + "internetAddress": [ + { + "type": "inet6", + "address": topo_modify_change_ip["routers"]["r0"][ + "links" + ]["r3"]["ipv6"].split("/")[0], + } + ], + } + } + } + } + } + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Modify the mask on the R0 interface") + ip_addr = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv6"] + mask = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv6"] + step("Delete the ip address") + topo1 = { + "r0": { + "links": { + "r3": { + "ipv6": ip_addr, + "interface": topo["routers"]["r0"]["links"]["r3"]["interface"], + "delete": True, + } + } + } + } + + result = create_interfaces_cfg(tgen, topo1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("Change the ip on the R0 interface") + + topo_modify_change_ip = deepcopy(topo) + intf_ip = topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv6"] + topo_modify_change_ip["routers"]["r0"]["links"]["r3"]["ipv6"] = str( + IPv6Address(frr_unicode(intf_ip.split("/")[0])) + 3 + ) + "/{}".format(int(intf_ip.split("/")[1]) + 1) + + build_config_from_json(tgen, topo_modify_change_ip, save_bkup=False) + step("Verify that interface is enabled in ospf.") + dut = "r0" + input_dict = { + "r0": { + "links": { + "r3": { + "ospf6": { + "internetAddress": [ + { + "type": "inet6", + "address": topo_modify_change_ip["routers"]["r0"][ + "links" + ]["r3"]["ipv6"].split("/")[0], + } + ], + } + } + } + } + } + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + topo1 = { + "r0": { + "links": { + "r3": { + "ipv6": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][ + "ipv6" + ], + "interface": topo_modify_change_ip["routers"]["r0"]["links"]["r3"][ + "interface" + ], + "delete": True, + } + } + } + } + + result = create_interfaces_cfg(tgen, topo1) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + build_config_from_json(tgen, topo, save_bkup=False) + + step("Change the area id on the interface") + input_dict = { + "r0": { + "links": { + "r3": { + "interface": topo["routers"]["r0"]["links"]["r3"]["interface"], + "ospf6": {"area": "0.0.0.0"}, + "delete": True, + } + } + } + } + + result = create_interfaces_cfg(tgen, input_dict) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + input_dict = { + "r0": { + "links": { + "r3": { + "interface": topo["routers"]["r0"]["links"]["r3"]["interface"], + "ospf6": {"area": "0.0.0.1"}, + } + } + } + } + + result = create_interfaces_cfg(tgen, input_dict) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + step("Verify that interface is enabled in ospf.") + dut = "r0" + input_dict = {"r0": {"links": {"r3": {"ospf6": {"ospf6Enabled": True}}}}} + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + input_dict = { + "r0": { + "links": { + "r3": { + "interface": topo["routers"]["r0"]["links"]["r3"]["interface"], + "ospf6": {"area": "0.0.0.1"}, + "delete": True, + } + } + } + } + + result = create_interfaces_cfg(tgen, input_dict) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + input_dict = { + "r0": { + "links": { + "r3": { + "interface": topo["routers"]["r0"]["links"]["r3"]["interface"], + "ospf6": {"area": "0.0.0.0"}, + } + } + } + } + + result = create_interfaces_cfg(tgen, input_dict) + assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result) + + step("Verify that interface is enabled in ospf.") + dut = "r0" + input_dict = {"r0": {"links": {"r3": {"ospf6": {"ospf6Enabled": True}}}}} + result = verify_ospf6_interface(tgen, topo, dut=dut, input_dict=input_dict) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + step("verify the all neighbors are up after clearing the process.") + for rtr in topo["routers"]: + clear_ospf(tgen, rtr, ospf="ospf6") + + ospf_covergence = verify_ospf6_neighbor(tgen, topo) + assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result) + + write_test_footer(tc_name) + + +if __name__ == "__main__": + args = ["-s"] + sys.argv[1:] + sys.exit(pytest.main(args)) -- 2.39.5