[mirror_frr.git] / tests / topotests / isis-topo1 / test_isis_topo1.py

#!/usr/bin/env python

#
# test_isis_topo1.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2017 by
# Network Device Education Foundation, Inc. ("NetDEF")
#
# 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 NETDEF DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF 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.
#

"""
test_isis_topo1.py: Test ISIS topology.
"""

import collections
import json
import os
import re
import sys
import pytest

CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, '../'))

# pylint: disable=C0413
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger

from mininet.topo import Topo


class ISISTopo1(Topo):
    "Simple two layer ISIS topology"
    def build(self, *_args, **_opts):
        "Build function"
        tgen = get_topogen(self)

        # Add ISIS routers:
        # r1      r2
        #  | sw1  | sw2
        # r3     r4
        #  |      |
        # sw3    sw4
        #   \    /
        #     r5
        for routern in range(1, 6):
            tgen.add_router('r{}'.format(routern))

        # r1 <- sw1 -> r3
        sw = tgen.add_switch('sw1')
        sw.add_link(tgen.gears['r1'])
        sw.add_link(tgen.gears['r3'])

        # r2 <- sw2 -> r4
        sw = tgen.add_switch('sw2')
        sw.add_link(tgen.gears['r2'])
        sw.add_link(tgen.gears['r4'])

        # r3 <- sw3 -> r5
        sw = tgen.add_switch('sw3')
        sw.add_link(tgen.gears['r3'])
        sw.add_link(tgen.gears['r5'])

        # r4 <- sw4 -> r5
        sw = tgen.add_switch('sw4')
        sw.add_link(tgen.gears['r4'])
        sw.add_link(tgen.gears['r5'])


def setup_module(mod):
    "Sets up the pytest environment"
    tgen = Topogen(ISISTopo1, mod.__name__)
    tgen.start_topology()

    # For all registered routers, load the zebra configuration file
    for rname, router in tgen.routers().iteritems():
        router.load_config(
            TopoRouter.RD_ZEBRA,
            os.path.join(CWD, '{}/zebra.conf'.format(rname))
        )
        router.load_config(
            TopoRouter.RD_ISIS,
            os.path.join(CWD, '{}/isisd.conf'.format(rname))
        )

    # After loading the configurations, this function loads configured daemons.
    tgen.start_router()


def teardown_module(mod):
    "Teardown the pytest environment"
    tgen = get_topogen()

    # This function tears down the whole topology.
    tgen.stop_topology()


def test_isis_convergence():
    "Wait for the protocol to converge before starting to test"
    tgen = get_topogen()
    # Don't run this test if we have any failure.
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    topotest.sleep(45, "waiting for ISIS protocol to converge")

    # Code to generate the json files.
    # for rname, router in tgen.routers().iteritems():
    #     open('/tmp/{}_topology.json'.format(rname), 'w').write(
    #         json.dumps(show_isis_topology(router), indent=2, sort_keys=True)
    #     )

    for rname, router in tgen.routers().iteritems():
        filename = '{0}/{1}/{1}_topology.json'.format(CWD, rname)
        expected = json.loads(open(filename, 'r').read())
        actual = show_isis_topology(router)
        assertmsg = "Router '{}' topology mismatch".format(rname)
        assert topotest.json_cmp(actual, expected) is None, assertmsg


def test_isis_route_installation():
    "Check whether all expected routes are present"
    tgen = get_topogen()
    # Don't run this test if we have any failure.
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    logger.info('Checking routers for installed ISIS routes')

    # Check for routes in 'show ip route json'
    for rname, router in tgen.routers().iteritems():
        filename = '{0}/{1}/{1}_route.json'.format(CWD, rname)
        expected = json.loads(open(filename, 'r').read())
        actual = router.vtysh_cmd('show ip route json', isjson=True)

        # Older FRR versions don't list interfaces in some ISIS routes
        if router.has_version('<', '3.1'):
            for network, routes in expected.iteritems():
                for route in routes:
                    if route['protocol'] != 'isis':
                        continue

                    for nexthop in route['nexthops']:
                        try:
                            nexthop.pop('interfaceIndex')
                        except KeyError:
                            pass

                        try:
                            nexthop.pop('interfaceName')
                        except KeyError:
                            pass

        assertmsg = "Router '{}' routes mismatch".format(rname)
        assert topotest.json_cmp(actual, expected) is None, assertmsg


def test_isis_linux_route_installation():
    "Check whether all expected routes are present and installed in the OS"
    tgen = get_topogen()
    # Don't run this test if we have any failure.
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    logger.info('Checking routers for installed ISIS routes in OS')

    # Check for routes in `ip route`
    for rname, router in tgen.routers().iteritems():
        filename = '{0}/{1}/{1}_route_linux.json'.format(CWD, rname)
        expected = json.loads(open(filename, 'r').read())
        actual = topotest.ip4_route(router)

        # Older FRR versions install routes using different proto
        if router.has_version('<', '3.1'):
            for network, netoptions in expected.iteritems():
                if 'proto' in netoptions and netoptions['proto'] == '187':
                    netoptions['proto'] = 'zebra'

        assertmsg = "Router '{}' OS routes mismatch".format(rname)
        assert topotest.json_cmp(actual, expected) is None, assertmsg


def test_memory_leak():
    "Run the memory leak test and report results."
    tgen = get_topogen()
    if not tgen.is_memleak_enabled():
        pytest.skip('Memory leak test/report is disabled')

    tgen.report_memory_leaks()


if __name__ == '__main__':
    args = ["-s"] + sys.argv[1:]
    sys.exit(pytest.main(args))


#
# Auxiliary functions
#


def dict_merge(dct, merge_dct):
    """
    Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
    updating only top-level keys, dict_merge recurses down into dicts nested
    to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
    ``dct``.
    :param dct: dict onto which the merge is executed
    :param merge_dct: dct merged into dct
    :return: None

    Source:
    https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
    """
    for k, v in merge_dct.iteritems():
        if (k in dct and isinstance(dct[k], dict)
                and isinstance(merge_dct[k], collections.Mapping)):
            dict_merge(dct[k], merge_dct[k])
        else:
            dct[k] = merge_dct[k]


def parse_topology(lines, level):
    """
    Parse the output of 'show isis topology level-X' into a Python dict.
    """
    areas = {}
    area = None
    ipv = None

    for line in lines:
        area_match = re.match(r"Area (.+):", line)
        if area_match:
            area = area_match.group(1)
            if area not in areas:
                areas[area] = {
                    level: {
                        'ipv4': [],
                        'ipv6': []
                    }
                }
            ipv = None
            continue
        elif area is None:
            continue

        if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
            ipv = 'ipv6'
            continue
        if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
            ipv = 'ipv4'
            continue

        item_match = re.match(
            r"([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+)", line)
        if item_match is not None:
            areas[area][level][ipv].append({
                'vertex': item_match.group(1),
                'type': item_match.group(2),
                'metric': item_match.group(3),
                'next-hop': item_match.group(4),
                'interface': item_match.group(5),
                'parent': item_match.group(6),
            })
            continue

        item_match = re.match(r"([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+)", line)
        if item_match is not None:
            areas[area][level][ipv].append({
                'vertex': item_match.group(1),
                'type': item_match.group(2),
                'metric': item_match.group(3),
                'parent': item_match.group(4),
            })
            continue

        item_match = re.match(r"([^ ]+)", line)
        if item_match is not None:
            areas[area][level][ipv].append({'vertex': item_match.group(1)})
            continue

    return areas


def show_isis_topology(router):
    """
    Get the ISIS topology in a dictionary format.

    Sample:
    {
      'area-name': {
        'level-1': [
          {
            'vertex': 'r1'
          }
        ],
        'level-2': [
          {
            'vertex': '10.0.0.1/24',
            'type': 'IP',
            'parent': '0',
            'metric': 'internal'
          }
        ]
      },
      'area-name-2': {
        'level-2': [
          {
            "interface": "rX-ethY",
            "metric": "Z",
            "next-hop": "rA",
            "parent": "rC(B)",
            "type": "TE-IS",
            "vertex": "rD"
          }
        ]
      }
    }
    """
    l1out = topotest.normalize_text(
        router.vtysh_cmd('show isis topology level-1')
    ).splitlines()
    l2out = topotest.normalize_text(
        router.vtysh_cmd('show isis topology level-2')
    ).splitlines()

    l1 = parse_topology(l1out, 'level-1')
    l2 = parse_topology(l2out, 'level-2')

    dict_merge(l1, l2)
    return l1
Commit	Line	Data
3aefe207 RZ	1	#!/usr/bin/env python
	2
	3	#
	4	# test_isis_topo1.py
	5	# Part of NetDEF Topology Tests
	6	#
	7	# Copyright (c) 2017 by
	8	# Network Device Education Foundation, Inc. ("NetDEF")
	9	#
	10	# Permission to use, copy, modify, and/or distribute this software
	11	# for any purpose with or without fee is hereby granted, provided
	12	# that the above copyright notice and this permission notice appear
	13	# in all copies.
	14	#
	15	# THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
	16	# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
	17	# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
	18	# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
	19	# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
	20	# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
	21	# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
	22	# OF THIS SOFTWARE.
	23	#
	24
	25	"""
	26	test_isis_topo1.py: Test ISIS topology.
	27	"""
	28
67f1e9ed RZ	29	import collections
67f1e9ed RZ	30	import json
3aefe207	31	import os
67f1e9ed	32	import re
3aefe207 RZ	33	import sys
	34	import pytest
	35
	36	CWD = os.path.dirname(os.path.realpath(__file__))
	37	sys.path.append(os.path.join(CWD, '../'))
	38
	39	# pylint: disable=C0413
	40	from lib import topotest
	41	from lib.topogen import Topogen, TopoRouter, get_topogen
	42	from lib.topolog import logger
	43
	44	from mininet.topo import Topo
	45
	46
	47	class ISISTopo1(Topo):
	48	"Simple two layer ISIS topology"
	49	def build(self, _args, *_opts):
	50	"Build function"
	51	tgen = get_topogen(self)
	52
	53	# Add ISIS routers:
	54	# r1 r2
	55	# \| sw1 \| sw2
	56	# r3 r4
	57	# \| \|
	58	# sw3 sw4
	59	# \ /
	60	# r5
	61	for routern in range(1, 6):
	62	tgen.add_router('r{}'.format(routern))
	63
	64	# r1 <- sw1 -> r3
	65	sw = tgen.add_switch('sw1')
	66	sw.add_link(tgen.gears['r1'])
	67	sw.add_link(tgen.gears['r3'])
	68
	69	# r2 <- sw2 -> r4
	70	sw = tgen.add_switch('sw2')
	71	sw.add_link(tgen.gears['r2'])
	72	sw.add_link(tgen.gears['r4'])
	73
	74	# r3 <- sw3 -> r5
	75	sw = tgen.add_switch('sw3')
	76	sw.add_link(tgen.gears['r3'])
	77	sw.add_link(tgen.gears['r5'])
	78
	79	# r4 <- sw4 -> r5
	80	sw = tgen.add_switch('sw4')
	81	sw.add_link(tgen.gears['r4'])
	82	sw.add_link(tgen.gears['r5'])
	83
	84
	85	def setup_module(mod):
	86	"Sets up the pytest environment"
	87	tgen = Topogen(ISISTopo1, mod.__name__)
	88	tgen.start_topology()
	89
	90	# For all registered routers, load the zebra configuration file
	91	for rname, router in tgen.routers().iteritems():
	92	router.load_config(
	93	TopoRouter.RD_ZEBRA,
	94	os.path.join(CWD, '{}/zebra.conf'.format(rname))
	95	)
	96	router.load_config(
97	TopoRouter.RD_ISIS,
98	os.path.join(CWD, '{}/isisd.conf'.format(rname))
99	)
100
101	# After loading the configurations, this function loads configured daemons.
102	tgen.start_router()
103
104
105	def teardown_module(mod):
106	"Teardown the pytest environment"
107	tgen = get_topogen()
108
109	# This function tears down the whole topology.
110	tgen.stop_topology()
111
112
113	def test_isis_convergence():
114	"Wait for the protocol to converge before starting to test"
115	tgen = get_topogen()
116	# Don't run this test if we have any failure.
117	if tgen.routers_have_failure():
118	pytest.skip(tgen.errors)
119
67f1e9ed RZ	120	topotest.sleep(45, "waiting for ISIS protocol to converge")
	121
	122	# Code to generate the json files.
	123	# for rname, router in tgen.routers().iteritems():
	124	# open('/tmp/{}_topology.json'.format(rname), 'w').write(
	125	# json.dumps(show_isis_topology(router), indent=2, sort_keys=True)
	126	# )
	127
	128	for rname, router in tgen.routers().iteritems():
	129	filename = '{0}/{1}/{1}_topology.json'.format(CWD, rname)
	130	expected = json.loads(open(filename, 'r').read())
	131	actual = show_isis_topology(router)
	132	assertmsg = "Router '{}' topology mismatch".format(rname)
	133	assert topotest.json_cmp(actual, expected) is None, assertmsg
3aefe207 RZ	134
3aefe207 RZ	135
e4d08d5b RZ	136	def test_isis_route_installation():
	137	"Check whether all expected routes are present"
	138	tgen = get_topogen()
	139	# Don't run this test if we have any failure.
	140	if tgen.routers_have_failure():
	141	pytest.skip(tgen.errors)
	142
	143	logger.info('Checking routers for installed ISIS routes')
	144
	145	# Check for routes in 'show ip route json'
	146	for rname, router in tgen.routers().iteritems():
	147	filename = '{0}/{1}/{1}_route.json'.format(CWD, rname)
	148	expected = json.loads(open(filename, 'r').read())
	149	actual = router.vtysh_cmd('show ip route json', isjson=True)
29614f56 RZ	150
	151	# Older FRR versions don't list interfaces in some ISIS routes
	152	if router.has_version('<', '3.1'):
	153	for network, routes in expected.iteritems():
	154	for route in routes:
	155	if route['protocol'] != 'isis':
	156	continue
	157
	158	for nexthop in route['nexthops']:
	159	try:
	160	nexthop.pop('interfaceIndex')
	161	except KeyError:
	162	pass
	163
	164	try:
	165	nexthop.pop('interfaceName')
	166	except KeyError:
	167	pass
	168
e4d08d5b RZ	169	assertmsg = "Router '{}' routes mismatch".format(rname)
	170	assert topotest.json_cmp(actual, expected) is None, assertmsg
	171
	172
2d013cda RZ	173	def test_isis_linux_route_installation():
	174	"Check whether all expected routes are present and installed in the OS"
	175	tgen = get_topogen()
	176	# Don't run this test if we have any failure.
	177	if tgen.routers_have_failure():
	178	pytest.skip(tgen.errors)
	179
	180	logger.info('Checking routers for installed ISIS routes in OS')
	181
	182	# Check for routes in `ip route`
	183	for rname, router in tgen.routers().iteritems():
	184	filename = '{0}/{1}/{1}_route_linux.json'.format(CWD, rname)
	185	expected = json.loads(open(filename, 'r').read())
	186	actual = topotest.ip4_route(router)
29614f56 RZ	187
	188	# Older FRR versions install routes using different proto
	189	if router.has_version('<', '3.1'):
	190	for network, netoptions in expected.iteritems():
	191	if 'proto' in netoptions and netoptions['proto'] == '187':
	192	netoptions['proto'] = 'zebra'
	193
2d013cda RZ	194	assertmsg = "Router '{}' OS routes mismatch".format(rname)
	195	assert topotest.json_cmp(actual, expected) is None, assertmsg
	196
	197
3aefe207 RZ	198	def test_memory_leak():
	199	"Run the memory leak test and report results."
	200	tgen = get_topogen()
	201	if not tgen.is_memleak_enabled():
	202	pytest.skip('Memory leak test/report is disabled')
	203
	204	tgen.report_memory_leaks()
	205
	206
	207	if __name__ == '__main__':
	208	args = ["-s"] + sys.argv[1:]
	209	sys.exit(pytest.main(args))
67f1e9ed RZ	210
	211
	212	#
	213	# Auxiliary functions
	214	#
	215
	216
	217	def dict_merge(dct, merge_dct):
	218	"""
	219	Recursive dict merge. Inspired by :meth:``dict.update()``, instead of
	220	updating only top-level keys, dict_merge recurses down into dicts nested
	221	to an arbitrary depth, updating keys. The ``merge_dct`` is merged into
	222	``dct``.
	223	:param dct: dict onto which the merge is executed
	224	:param merge_dct: dct merged into dct
	225	:return: None
	226
	227	Source:
	228	https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
	229	"""
	230	for k, v in merge_dct.iteritems():
	231	if (k in dct and isinstance(dct[k], dict)
	232	and isinstance(merge_dct[k], collections.Mapping)):
	233	dict_merge(dct[k], merge_dct[k])
	234	else:
	235	dct[k] = merge_dct[k]
	236
	237
	238	def parse_topology(lines, level):
	239	"""
	240	Parse the output of 'show isis topology level-X' into a Python dict.
	241	"""
	242	areas = {}
67f1e9ed	243	area = None
5836fac2	244	ipv = None
67f1e9ed RZ	245
67f1e9ed RZ	246	for line in lines:
5836fac2 RZ	247	area_match = re.match(r"Area (.+):", line)
5836fac2 RZ	248	if area_match:
67f1e9ed	249	area = area_match.group(1)
5836fac2 RZ	250	if area not in areas:
	251	areas[area] = {
	252	level: {
	253	'ipv4': [],
	254	'ipv6': []
	255	}
	256	}
	257	ipv = None
	258	continue
	259	elif area is None:
67f1e9ed RZ	260	continue
67f1e9ed RZ	261
5836fac2 RZ	262	if re.match(r"IS\-IS paths to level-. routers that speak IPv6", line):
5836fac2 RZ	263	ipv = 'ipv6'
67f1e9ed	264	continue
5836fac2 RZ	265	if re.match(r"IS\-IS paths to level-. routers that speak IP", line):
5836fac2 RZ	266	ipv = 'ipv4'
67f1e9ed RZ	267	continue
	268
	269	item_match = re.match(
	270	r"([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+)", line)
	271	if item_match is not None:
5836fac2	272	areas[area][level][ipv].append({
67f1e9ed RZ	273	'vertex': item_match.group(1),
	274	'type': item_match.group(2),
	275	'metric': item_match.group(3),
	276	'next-hop': item_match.group(4),
	277	'interface': item_match.group(5),
	278	'parent': item_match.group(6),
	279	})
	280	continue
	281
	282	item_match = re.match(r"([^ ]+) ([^ ]+) ([^ ]+) ([^ ]+)", line)
	283	if item_match is not None:
5836fac2	284	areas[area][level][ipv].append({
67f1e9ed RZ	285	'vertex': item_match.group(1),
	286	'type': item_match.group(2),
	287	'metric': item_match.group(3),
	288	'parent': item_match.group(4),
	289	})
	290	continue
	291
	292	item_match = re.match(r"([^ ]+)", line)
	293	if item_match is not None:
5836fac2	294	areas[area][level][ipv].append({'vertex': item_match.group(1)})
67f1e9ed RZ	295	continue
67f1e9ed RZ	296
67f1e9ed RZ	297	return areas
	298
	299
	300	def show_isis_topology(router):
	301	"""
	302	Get the ISIS topology in a dictionary format.
	303
	304	Sample:
	305	{
	306	'area-name': {
	307	'level-1': [
	308	{
	309	'vertex': 'r1'
	310	}
	311	],
	312	'level-2': [
	313	{
	314	'vertex': '10.0.0.1/24',
	315	'type': 'IP',
	316	'parent': '0',
	317	'metric': 'internal'
	318	}
	319	]
	320	},
	321	'area-name-2': {
	322	'level-2': [
	323	{
	324	"interface": "rX-ethY",
	325	"metric": "Z",
	326	"next-hop": "rA",
	327	"parent": "rC(B)",
	328	"type": "TE-IS",
	329	"vertex": "rD"
	330	}
	331	]
	332	}
	333	}
	334	"""
	335	l1out = topotest.normalize_text(
	336	router.vtysh_cmd('show isis topology level-1')
	337	).splitlines()
	338	l2out = topotest.normalize_text(
	339	router.vtysh_cmd('show isis topology level-2')
	340	).splitlines()
	341
	342	l1 = parse_topology(l1out, 'level-1')
	343	l2 = parse_topology(l2out, 'level-2')
	344
	345	dict_merge(l1, l2)
	346	return l1