diff options
Diffstat (limited to 'tests/topotests/lib/topojson.py')
| -rw-r--r-- | tests/topotests/lib/topojson.py | 153 |
1 files changed, 111 insertions, 42 deletions
diff --git a/tests/topotests/lib/topojson.py b/tests/topotests/lib/topojson.py index fcc6c19868..4f23e1ace0 100644 --- a/tests/topotests/lib/topojson.py +++ b/tests/topotests/lib/topojson.py @@ -18,67 +18,64 @@ # OF THIS SOFTWARE. # -from collections import OrderedDict -from json import dumps as json_dumps -from re import search as re_search +import json import ipaddress -import pytest -import ipaddr +import os +from collections import OrderedDict from copy import deepcopy +from re import search as re_search +import pytest -# Import topogen and topotest helpers -from lib.topolog import logger - -# Required to instantiate the topology builder class. +from lib.bgp import create_router_bgp from lib.common_config import ( - number_to_row, - number_to_column, - load_config_to_router, + create_bgp_community_lists, create_interfaces_cfg, - create_static_routes, create_prefix_lists, create_route_maps, - create_bgp_community_lists, + create_static_routes, create_vrf_cfg, + load_config_to_routers, + start_topology, + topo_daemons, + number_to_column, ) - -from lib.pim import create_pim_config, create_igmp_config -from lib.bgp import create_router_bgp from lib.ospf import create_router_ospf, create_router_ospf6 - -ROUTER_LIST = [] +from lib.pim import create_igmp_config, create_pim_config +from lib.topolog import logger -def build_topo_from_json(tgen, topo): +def build_topo_from_json(tgen, topo=None): """ Reads configuration from JSON file. Adds routers, creates interface names dynamically and link routers as defined in JSON to create topology. Assigns IPs dynamically to all interfaces of each router. * `tgen`: Topogen object - * `topo`: json file data + * `topo`: json file data, or use tgen.json_topo if None """ + if topo is None: + topo = tgen.json_topo - ROUTER_LIST = sorted( - topo["routers"].keys(), key=lambda x: int(re_search("\d+", x).group(0)) + router_list = sorted( + topo["routers"].keys(), key=lambda x: int(re_search(r"\d+", x).group(0)) ) - SWITCH_LIST = [] + switch_list = [] if "switches" in topo: - SWITCH_LIST = sorted( - topo["switches"].keys(), key=lambda x: int(re_search("\d+", x).group(0)) + switch_list = sorted( + topo["switches"].keys(), key=lambda x: int(re_search(r"\d+", x).group(0)) ) - listRouters = sorted(ROUTER_LIST[:]) - listSwitches = sorted(SWITCH_LIST[:]) + listRouters = sorted(router_list[:]) + listSwitches = sorted(switch_list[:]) listAllRouters = deepcopy(listRouters) dictSwitches = {} - for routerN in ROUTER_LIST: + for routerN in router_list: logger.info("Topo: Add router {}".format(routerN)) tgen.add_router(routerN) - for switchN in SWITCH_LIST: + for switchN in switch_list: logger.info("Topo: Add switch {}".format(switchN)) dictSwitches[switchN] = tgen.add_switch(switchN) @@ -101,7 +98,7 @@ def build_topo_from_json(tgen, topo): # Physical Interfaces if "links" in topo["routers"][curRouter]: for destRouterLink, data in sorted( - topo["routers"][curRouter]["links"].iteritems() + topo["routers"][curRouter]["links"].items() ): currRouter_lo_json = topo["routers"][curRouter]["links"][destRouterLink] # Loopback interfaces @@ -204,7 +201,7 @@ def build_topo_from_json(tgen, topo): logger.debug( "Generated link data for router: %s\n%s", curRouter, - json_dumps( + json.dumps( topo["routers"][curRouter]["links"], indent=4, sort_keys=True ), ) @@ -282,24 +279,47 @@ def build_topo_from_json(tgen, topo): ] = "{}/{}".format( ipv6Next, topo["link_ip_start"]["v6mask"] ) - ipv6Next = ipaddr.IPv6Address(int(ipv6Next) + ipv6Step) + ipv6Next = ipaddress.IPv6Address(int(ipv6Next) + ipv6Step) logger.debug( "Generated link data for router: %s\n%s", curRouter, - json_dumps( + json.dumps( topo["routers"][curRouter]["links"], indent=4, sort_keys=True ), ) -def build_config_from_json(tgen, topo, save_bkup=True): +def linux_intf_config_from_json(tgen, topo=None): + """Configure interfaces from linux based on topo.""" + if topo is None: + topo = tgen.json_topo + + routers = topo["routers"] + for rname in routers: + router = tgen.net[rname] + links = routers[rname]["links"] + for rrname in links: + link = links[rrname] + if rrname == "lo": + lname = "lo" + else: + lname = link["interface"] + if "ipv4" in link: + router.cmd_raises("ip addr add {} dev {}".format(link["ipv4"], lname)) + if "ipv6" in link: + router.cmd_raises( + "ip -6 addr add {} dev {}".format(link["ipv6"], lname) + ) + + +def build_config_from_json(tgen, topo=None, save_bkup=True): """ Reads initial configuraiton from JSON for each router, builds configuration and loads its to router. * `tgen`: Topogen object - * `topo`: json file data + * `topo`: json file data, or use tgen.json_topo if None """ func_dict = OrderedDict( @@ -318,16 +338,65 @@ def build_config_from_json(tgen, topo, save_bkup=True): ] ) + if topo is None: + topo = tgen.json_topo + data = topo["routers"] for func_type in func_dict.keys(): logger.info("Checking for {} configuration in input data".format(func_type)) func_dict.get(func_type)(tgen, data, build=True) - for router in sorted(topo["routers"].keys()): - logger.debug("Configuring router {}...".format(router)) + routers = sorted(topo["routers"].keys()) + result = load_config_to_routers(tgen, routers, save_bkup) + if not result: + logger.info("build_config_from_json: failed to configure topology") + pytest.exit(1) + + +def create_tgen_from_json(testfile, json_file=None): + """Create a topogen object given a testfile. + + - `testfile` : The path to the testfile. + - `json_file` : The path to the json config file. If None the pathname is derived + from the `testfile` first by trying to replace `.py` by `.json` and if that isn't + present then by removing `test_` prefix as well. + """ + from lib.topogen import Topogen # Topogen imports this module too + + thisdir = os.path.dirname(os.path.realpath(testfile)) + basename = os.path.basename(testfile) + logger.debug("starting standard JSON based module setup for %s", basename) + + assert basename.startswith("test_") + assert basename.endswith(".py") + json_file = os.path.join(thisdir, basename[:-3] + ".json") + if not os.path.exists(json_file): + json_file = os.path.join(thisdir, basename[5:-3] + ".json") + assert os.path.exists(json_file) + with open(json_file, "r") as topof: + topo = json.load(topof) + + # Create topology + tgen = Topogen(lambda tgen: build_topo_from_json(tgen, topo), basename[:-3]) + tgen.json_topo = topo + return tgen + + +def setup_module_from_json(testfile, json_file=None): + """Do the standard module setup for JSON based test. + + * `testfile` : The path to the testfile. The name is used to derive the json config + file name as well (removing `test_` prefix and replacing `.py` suffix with `.json` + """ + # Create topology object + tgen = create_tgen_from_json(testfile, json_file) + + # Start routers (and their daemons) + start_topology(tgen, topo_daemons(tgen)) + + # Configure routers + build_config_from_json(tgen) + assert not tgen.routers_have_failure() - result = load_config_to_router(tgen, router, save_bkup) - if not result: - logger.info("Failed while configuring {}".format(router)) - pytest.exit(1) + return tgen |