import os
import re
import sys
+import json
from functools import partial
# Save the Current Working Directory to find configuration files.
logger.info("\n\n---- OSPF Segment Routing tests End ----\n")
-# Shared test function to validate expected output.
-def compare_ospf_srdb(rname, expected):
- """
- Calls 'show ip ospf database segment-routing json' for router `rname`
- and compare the obtained result with the expected output.
- """
- tgen = get_topogen()
- current = tgen.gears[rname].vtysh_cmd("show ip ospf database segment-routing json")
- # Filter Adjacency SID allocation
- current = re.sub(r'"sid":5000[0-9],', '"sid":"XX",', current)
- current = re.sub(r'"inputLabel":5000[0-9],', '"inputLabel":"XX",', current)
- return topotest.difflines(
- current, expected, title1="Current output", title2="Expected output"
- )
-
-
-def compare_mpls_table(rname, expected):
- """
- Calls 'show mpls table json' for router `rname` and compare the obtained
- result with the expected output.
- """
- tgen = get_topogen()
- current = tgen.gears[rname].vtysh_cmd("show mpls table json")
- # Filter Adjacency SID allocation
- current = re.sub(r'"5000[0-9]":', '"XX":', current)
- current = re.sub(r'"inLabel":5000[0-9],', '"inLabel":"XX",', current)
- return topotest.difflines(
- current, expected, title1="Current output", title2="Expected output"
- )
-
-
def test_ospf_sr():
"Test OSPF daemon Segment Routing"
tgen = get_topogen()
# Load expected results from the command
reffile = os.path.join(CWD, "{}/ospf_srdb.json".format(router))
- expected = open(reffile).read()
+ expected = json.loads(open(reffile).read())
# Run test function until we get an result. Wait at most 60 seconds.
- test_func = partial(compare_ospf_srdb, router, expected)
- result, diff = topotest.run_and_expect(test_func, "", count=25, wait=3)
- assert result, ("OSPF did not start Segment Routing on {}:\n{}").format(
+ rt = tgen.gears[router]
+ test_func = partial(
+ topotest.router_json_cmp, rt, 'show ip ospf database segment-routing json', expected
+ )
+ rv, diff = topotest.run_and_expect(test_func, None, count=25, wait=3)
+ assert rv, "OSPF did not start Segment Routing on {}:\n{}".format(
router, diff
)
logger.info("--- test OSPF Segment Routing MPLS tables ---")
+ def show_mpls_table_json_cmp(rt, expected):
+ "Removes random label and use `label-X` instead."
+ text = rt.vtysh_cmd('show mpls table json')
+
+ # Substitue random labels with fixed label value.
+ for label in range(1, 10):
+ text = re.sub(r'"5000[0-9]"', '"label-{}"'.format(label), text,
+ count=1)
+
+ print '\n{}\n'.format(text)
+
+ output = json.loads(text)
+ return topotest.json_cmp(output, expected)
+
for rnum in range(1, 5):
router = "r{}".format(rnum)
# Load expected results from the command
reffile = os.path.join(CWD, "{}/zebra_mpls.json".format(router))
- expected = open(reffile).read()
+ expected = json.loads(open(reffile).read())
# Run test function until we get an result. Wait at most 60 seconds.
- test_func = partial(compare_mpls_table, router, expected)
- result, diff = topotest.run_and_expect(test_func, "", count=25, wait=3)
- assert result, ("OSPF did not properly instal MPLS table on {}:\n{}").format(
+ rt = tgen.gears[router]
+ test_func = partial(show_mpls_table_json_cmp, rt, expected)
+ rv, diff = topotest.run_and_expect(test_func, None, count=25, wait=3)
+ assert rv, "OSPF did not properly instal MPLS table on {}:\n{}".format(
router, diff
)
projects / mirror / frr.git / commitdiff