# 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,
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
config_data.append(cmd)
+
# router id
router_id = ospf_data.setdefault("router_id", None)
del_router_id = ospf_data.setdefault("del_router_id", False)
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:
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", {})
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:
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,
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()")
################################
# 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
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
--- /dev/null
+#!/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))
--- /dev/null
+#!/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))
projects / matthieu / frr.git / commitdiff