topotests: OSPFv3 NSSA test LSA type 7

New OSPFv3 NSSA test:
 * When a static route is redistributed to an NSSA router it should be
   type 7 and should show up in OSPFv3 route database.
 * Test LSA Type 7 and route removal.

Co-authored-by: Soman K.S <somanks@gmail.com>
Signed-off-by: Rafael Zalamena <rzalamena@opensourcerouting.org>

1 files changed, 129 insertions, 26 deletions

diff --git a/tests/topotests/ospf6_topo2/test_ospf6_topo2.py b/tests/topotests/ospf6_topo2/test_ospf6_topo2.py
index 8289292632..0fe5228ce6 100644
--- a/tests/topotests/ospf6_topo2/test_ospf6_topo2.py
+++ b/tests/topotests/ospf6_topo2/test_ospf6_topo2.py
@@ -47,6 +47,49 @@ from mininet.topo import Topo
 pytestmark = [pytest.mark.ospf6d]
 
 
+def expect_lsas(router, area, lsas, wait=5, extra_params=""):
+    """
+    Run the OSPFv3 show LSA database command and expect the supplied LSAs.
+
+    Optional parameters:
+     * `wait`: amount of seconds to wait.
+     * `extra_params`: extra LSA database parameters.
+     * `inverse`: assert the inverse of the expected.
+    """
+    tgen = get_topogen()
+
+    command = "show ipv6 ospf6 database {} json".format(extra_params)
+
+    logger.info("waiting OSPFv3 router '{}' LSA".format(router))
+    test_func = partial(
+        topotest.router_json_cmp,
+        tgen.gears[router],
+        command,
+        {"areaScopedLinkStateDb": [{"areaId": area, "lsa": lsas}]},
+    )
+    _, result = topotest.run_and_expect(test_func, None, count=wait, wait=1)
+    assertmsg = '"{}" convergence failure'.format(router)
+
+    assert result is None, assertmsg
+
+
+def expect_ospfv3_routes(router, routes, wait=5):
+    "Run command `ipv6 ospf6 route` and expect route with type."
+    tgen = get_topogen()
+
+    logger.info("waiting OSPFv3 router '{}' route".format(router))
+    test_func = partial(
+        topotest.router_json_cmp,
+        tgen.gears[router],
+        "show ipv6 ospf6 route json",
+        {"routes": routes}
+    )
+    _, result = topotest.run_and_expect(test_func, None, count=wait, wait=1)
+    assertmsg = '"{}" convergence failure'.format(router)
+
+    assert result is None, assertmsg
+
+
 class OSPFv3Topo2(Topo):
     "Test topology builder"
 
@@ -183,36 +226,96 @@ def test_ospf6_default_route():
         assertmsg = '"{}" convergence failure'.format(router)
         assert result is None, assertmsg
 
-    def expect_lsa(router, area, prefix, metric):
-        "Test OSPF6 LSA existence."
-        logger.info("waiting OSPFv3 router '{}' LSA".format(router))
-        test_func = partial(
-            topotest.router_json_cmp,
-            tgen.gears[router],
-            "show ipv6 ospf6 database inter-prefix detail json",
-            {
-                "areaScopedLinkStateDb": [
-                    {
-                        "areaId": area,
-                        "lsa": [
-                            {
-                                "prefix": prefix,
-                                "metric": metric,
-                            }
-                        ],
-                    }
-                ]
-            },
-        )
-        _, result = topotest.run_and_expect(test_func, None, count=4, wait=1)
-        assertmsg = '"{}" convergence failure'.format(router)
-        assert result is None, assertmsg
-
     metric = 123
-    expect_lsa("r1", "0.0.0.1", "::/0", metric)
+    expect_lsas(
+        "r1",
+        "0.0.0.1",
+        [{"prefix": "::/0", "metric": metric}],
+        extra_params="inter-prefix detail",
+    )
     expect_route("r1", "::/0", metric + 10)
 
 
+def test_nssa_lsa_type7():
+    """
+    Test that static route gets announced as external route when redistributed
+    and gets removed when redistribution stops.
+    """
+    tgen = get_topogen()
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    #
+    # Add new static route and check if it gets announced as LSA Type-7.
+    #
+    config = """
+    configure terminal
+    ipv6 route 2001:db8:100::/64 Null0
+    """
+    tgen.gears["r2"].vtysh_cmd(config)
+
+    lsas = [
+        {
+            "type": "NSSA",
+            "advertisingRouter": "10.254.254.2",
+            "prefix": "2001:db8:100::/64",
+            "forwardingAddress": "2001:db8:3::1",
+        }
+    ]
+    route = {
+        "2001:db8:100::/64": {
+            "pathType": "E1",
+            "nextHops": [
+                {"nextHop": "::", "interfaceName": "r4-eth0"}
+            ]
+        }
+    }
+
+    logger.info("Expecting LSA type-7 and OSPFv3 route 2001:db8:100::/64 to show up")
+    expect_lsas("r4", "0.0.0.2", lsas, wait=30, extra_params="type-7 detail")
+    expect_ospfv3_routes("r4", route, wait=30)
+
+    #
+    # Remove static route and check for LSA Type-7 removal.
+    #
+    config = """
+    configure terminal
+    no ipv6 route 2001:db8:100::/64 Null0
+    """
+    tgen.gears["r2"].vtysh_cmd(config)
+
+    def dont_expect_lsa(unexpected_lsa):
+        "Specialized test function to expect LSA go missing"
+        output = tgen.gears["r4"].vtysh_cmd("show ipv6 ospf6 database type-7 detail json", isjson=True)
+        for lsa in output['areaScopedLinkStateDb'][0]['lsa']:
+            if lsa["prefix"] == unexpected_lsa["prefix"]:
+                if lsa["forwardingAddress"] == unexpected_lsa["forwardingAddress"]:
+                    return lsa
+        return None
+
+    def dont_expect_route(unexpected_route):
+        "Specialized test function to expect route go missing"
+        output = tgen.gears["r4"].vtysh_cmd("show ipv6 ospf6 route json", isjson=True)
+        if output["routes"].has_key(unexpected_route):
+            return output["routes"][unexpected_route]
+        return None
+
+
+    logger.info("Expecting LSA type-7 and OSPFv3 route 2001:db8:100::/64 to go away")
+
+    # Test that LSA doesn't exist.
+    test_func = partial(dont_expect_lsa, lsas[0])
+    _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
+    assertmsg = '"{}" LSA still exists'.format("r4")
+    assert result is None, assertmsg
+
+    # Test that route doesn't exist.
+    test_func = partial(dont_expect_route, "2001:db8:100::/64")
+    _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
+    assertmsg = '"{}" route still exists'.format("r4")
+    assert result is None, assertmsg
+
+
 def teardown_module(_mod):
     "Teardown the pytest environment"
     tgen = get_topogen()