topotests: new test for multicast route maps

Implement tests for the new multicast route-map filtering.

Signed-off-by: Rafael Zalamena <[email protected]>

Author: rzalamena

tests/topotests/multicast_gmp_route_map_topo1/r1/frr.conf

@@ -0,0 +1,9 @@
+log commands
+!
+interface r1-eth0
+ ip address 192.168.100.1/24
+ ip igmp
+exit
+!
+router pim
+exit

tests/topotests/multicast_gmp_route_map_topo1/test_multicast_gmp_route_map_topo1.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python
+# SPDX-License-Identifier: ISC
+
+#
+# test_multicast_gmp_route_map_topo1.py
+# Part of NetDEF Topology Tests
+#
+# Copyright (c) 2025 by
+# Network Device Education Foundation, Inc. ("NetDEF")
+#
+
+"""
+test_multicast_gmp_route_map_topo1.py: Test the FRR PIM multicast route map.
+"""
+
+import os
+import sys
+import json
+from functools import partial
+import re
+import pytest
+
+# Save the Current Working Directory to find configuration files.
+CWD = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(CWD, "../"))
+
+# pylint: disable=C0413
+# Import topogen and topotest helpers
+from lib import topotest
+
+# Required to instantiate the topology builder class.
+from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topolog import logger
+
+from lib.pim import McastTesterHelper
+
+pytestmark = [pytest.mark.bgpd, pytest.mark.pimd]
+
+app_helper = McastTesterHelper()
+
+
+def build_topo(tgen):
+    """
+    +----+     +----+
+    | h1 | <-> | r1 |
+    +----+     +----+
+                 ^
+    +----+       |
+    | h2 | <-----+
+    +----+
+    """
+
+    tgen.add_router(f"r1")
+    tgen.add_host("h1", "192.168.100.100/24", "via 192.168.100.1")
+    tgen.add_host("h2", "192.168.101.100/24", "via 192.168.101.1")
+
+    switch = tgen.add_switch("s1")
+    switch.add_link(tgen.gears["r1"])
+    switch.add_link(tgen.gears["h1"])
+
+    switch = tgen.add_switch("s2")
+    switch.add_link(tgen.gears["r1"])
+    switch.add_link(tgen.gears["h2"])
+
+
+def setup_module(mod):
+    "Sets up the pytest environment"
+    tgen = Topogen(build_topo, mod.__name__)
+    tgen.start_topology()
+
+    tgen.gears["r1"].load_frr_config(os.path.join(CWD, "r1/frr.conf"))
+    tgen.start_router()
+
+    app_helper.init(tgen)
+
+
+def teardown_module():
+    "Teardown the pytest environment"
+    tgen = get_topogen()
+    app_helper.cleanup()
+    tgen.stop_topology()
+
+
+def test_igmp_route_map():
+    "Test IGMP route map filtering"
+
+    tgen = get_topogen()
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    def expect_loopback_route(router, iptype, route, proto):
+        "Wait until route is present on RIB for protocol."
+        logger.info("waiting route {} in {}".format(route, router))
+        test_func = partial(
+            topotest.router_json_cmp,
+            tgen.gears[router],
+            "show {} route json".format(iptype),
+            {route: [{"protocol": proto}]},
+        )
+        _, result = topotest.run_and_expect(test_func, None, count=130, wait=1)
+        assertmsg = '"{}" convergence failure'.format(router)
+        assert result is None, assertmsg
+
+    # Wait for R1
+    expect_loopback_route("r1", "ip", "10.254.254.2/32", "bgp")
+
+    # Wait for R2
+    expect_loopback_route("r2", "ip", "10.254.254.1/32", "bgp")
+
+
+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))