#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
#
# test_bgp_multiview_topo1.py
# Copyright (c) 2016 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.
-#
-"""
+r"""
test_ldp_topo1.py: Simple FRR LDP Test
+---------+
import re
import sys
import pytest
+import json
+from functools import partial
from time import sleep
+from lib.topolog import logger
+
+# 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.dirname(os.path.dirname(os.path.abspath(__file__))))
from lib import topotest
switch.add_link(tgen.gears["r3"])
+#####################################################
+##
+## Helper functions
+##
+#####################################################
+
+
+def router_compare_json_output(rname, command, reference, count=60, wait=1):
+ "Compare router JSON output"
+
+ logger.info('Comparing router "%s" "%s" output', rname, command)
+
+ tgen = get_topogen()
+ filename = "{}/{}/{}".format(CWD, rname, reference)
+ expected = json.loads(open(filename).read())
+
+ # Run test function until we get an result.
+ test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
+ _, diff = topotest.run_and_expect(test_func, None, count, wait)
+ assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
+ assert diff is None, assertmsg
+
+
#####################################################
##
## Tests starting
assert fatal_error == "", fatal_error
+def test_ospf_convergence():
+ logger.info("Test: check OSPF adjacencies")
+
+ # Skip if previous fatal error condition is raised
+ if fatal_error != "":
+ pytest.skip(fatal_error)
+
+ for rname in ["r1", "r2", "r3", "r4"]:
+ router_compare_json_output(
+ rname, "show ip ospf neighbor json", "show_ip_ospf_neighbor.json"
+ )
+
+
def test_mpls_ldp_neighbor_establish():
global fatal_error
net = get_topogen().net
if fatal_error != "":
pytest.skip(fatal_error)
+ neighbors_operational = {
+ 1: 1,
+ 2: 3,
+ 3: 2,
+ 4: 2,
+ }
+
# Wait for MPLS LDP neighbors to establish.
print("\n\n** Verify MPLS LDP neighbors to establish")
print("******************************************\n")
established = "" # Empty string shows NOT established
if re.search(operational, lines[j]):
found_operational += 1
+
+ logger.info("Found operational %d" % found_operational)
if found_operational < 1:
# Need at least one operational neighbor
established = "" # Empty string shows NOT established
+ else:
+ if found_operational != neighbors_operational[i]:
+ established = ""
if not established:
print("Waiting for r%s" % i)
sys.stdout.flush()
else:
print("r%s ok" % i)
- assert (
- failures == 0
- ), "MPLS LDP Interface binding output for router r%s:\n%s" % (i, diff)
+ assert failures == 0, "MPLS LDP binding output for router r%s:\n%s" % (
+ i,
+ diff,
+ )
# Make sure that all daemons are running
for i in range(1, 5):