sys.path.append(os.path.join(CWD, "../"))
# Required to instantiate the topology builder class.
-pytestmark = [pytest.mark.pimd]
+pytestmark = [
+ pytest.mark.pimd,
+ pytest.mark.pim6d,
+ pytest.mark.ospfd,
+ pytest.mark.ospf6d,
+]
def build_topo(tgen):
sw1.add_link(tgen.gears["r4"])
sw3.add_link(tgen.gears["r6"])
+
def setup_module(mod):
logger.info("PIM Candidate RP/BSR:\n {}".format(TOPOLOGY))
tgen = get_topogen()
tgen.stop_topology()
+
def test_pim_bsr_election_r1(request):
"Test PIM BSR Election"
tgen = get_topogen()
r2 = tgen.gears["r2"]
# r1 should be the BSR winner because it has higher priority
expected = {
- "bsr":"10.0.0.1",
- "priority":200,
- "state":"ACCEPT_PREFERRED",
+ "bsr": "10.0.0.1",
+ "priority": 200,
+ "state": "ACCEPT_PREFERRED",
}
- test_func = partial(
- topotest.router_json_cmp, r2, "show ip pim bsr json", expected
- )
+ test_func = partial(topotest.router_json_cmp, r2, "show ip pim bsr json", expected)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
assertmsg = "r2: r1 was not elected, bsr election mismatch"
assert result is None, assertmsg
+
def test_pim_bsr_cand_bsr_r1(request):
"Test PIM BSR candidate BSR"
tgen = get_topogen()
r2 = tgen.gears["r2"]
# r2 is a candidate bsr with low priority: elected = False
- expected = {
- "address": "10.0.0.2",
- "priority": 100,
- "elected": False
- }
+ expected = {"address": "10.0.0.2", "priority": 100, "elected": False}
test_func = partial(
topotest.router_json_cmp, r2, "show ip pim bsr candidate-bsr json", expected
)
assertmsg = "r2: candidate bsr mismatch "
assert result is None, assertmsg
+
def test_pim_bsr_cand_rp(request):
"Test PIM BSR candidate RP"
tgen = get_topogen()
r3 = tgen.gears["r3"]
# r3 is a candidate rp
- expected = {
- "address":"10.0.3.3",
- "priority":10
- }
+ expected = {"address": "10.0.3.3", "priority": 10}
test_func = partial(
topotest.router_json_cmp, r3, "show ip pim bsr candidate-rp json", expected
)
# At this point, all nodes, including r5 should have synced the RP state
step("Verify rp-info on r5 from BSR")
- result = verify_pim_rp_info(tgen, None, "r5", "239.0.0.0/16", None, "10.0.3.3",
- "BSR", False, "ipv4", True, retry_timeout = 90)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "239.0.0.0/16",
+ None,
+ "10.0.3.3",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=90,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
- result = verify_pim_rp_info(tgen, None, "r5", "239.0.0.0/8", None, "10.0.3.4",
- "BSR", False, "ipv4", True, retry_timeout = 30)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "239.0.0.0/8",
+ None,
+ "10.0.3.4",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=30,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
- result = verify_pim_rp_info(tgen, None, "r5", "239.0.0.0/24", None, "10.0.3.4",
- "BSR", False, "ipv4", True, retry_timeout = 30)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "239.0.0.0/24",
+ None,
+ "10.0.3.4",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=30,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Verify rp-info on the BSR node itself r1")
- result = verify_pim_rp_info(tgen, None, "r1", "239.0.0.0/16", None, "10.0.3.3",
- "BSR", False, "ipv4", True, retry_timeout = 10)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r1",
+ "239.0.0.0/16",
+ None,
+ "10.0.3.3",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=10,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
- result = verify_pim_rp_info(tgen, None, "r1", "239.0.0.0/8", None, "10.0.3.4",
- "BSR", False, "ipv4", True, retry_timeout = 10)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r1",
+ "239.0.0.0/8",
+ None,
+ "10.0.3.4",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=10,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
- result = verify_pim_rp_info(tgen, None, "r1", "239.0.0.0/24", None, "10.0.3.4",
- "BSR", False, "ipv4", True, retry_timeout = 10)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r1",
+ "239.0.0.0/24",
+ None,
+ "10.0.3.4",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=10,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
configure
router pim
no bsr candidate-bsr priority 200 source address 10.0.0.1
- """)
+ """
+ )
step("Verify r1 is no longer a BSR candidate")
expected = {}
r2 = tgen.gears["r2"]
# We should fall back to r2 as the BSR
expected = {
- "bsr":"10.0.0.2",
- "priority":100,
- "state":"BSR_ELECTED",
+ "bsr": "10.0.0.2",
+ "priority": 100,
+ "state": "BSR_ELECTED",
}
step("Verify that we fallback to r2 as the new BSR")
- test_func = partial(
- topotest.router_json_cmp, r2, "show ip pim bsr json", expected
- )
+ test_func = partial(topotest.router_json_cmp, r2, "show ip pim bsr json", expected)
_, result = topotest.run_and_expect(test_func, None, count=180, wait=1)
assertmsg = "r2: failed to fallback to r2 as a BSR"
configure
router pim
no bsr candidate-rp group 239.0.0.0/16
- """)
+ """
+ )
step("Verify falling back to r4 as the new RP for 239.0.0.0/16")
- result = verify_pim_rp_info(tgen, None, "r5", "239.0.0.0/16", None, "10.0.3.4",
- "BSR", False, "ipv4", True, retry_timeout = 30)
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "239.0.0.0/16",
+ None,
+ "10.0.3.4",
+ "BSR",
+ False,
+ "ipv4",
+ True,
+ retry_timeout=30,
+ )
+ assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
+
+
+def test_pimv6_bsr_election_r1(request):
+ "Test PIMv6 BSR Election"
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ if tgen.routers_have_failure():
+ pytest.skip("skipped because of router(s) failure")
+
+ r2 = tgen.gears["r2"]
+ # r1 should be the BSR winner because it has higher priority
+ expected = {
+ "bsr": "fd00::1",
+ "priority": 200,
+ "state": "ACCEPT_PREFERRED",
+ }
+
+ test_func = partial(
+ topotest.router_json_cmp, r2, "show ipv6 pim bsr json", expected
+ )
+ _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
+
+ assertmsg = "r2: r1 was not elected, IPv6 bsr election mismatch"
+ assert result is None, assertmsg
+
+
+def test_pimv6_bsr_cand_rp(request):
+ "Test PIMv6 BSR candidate RP"
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ if tgen.routers_have_failure():
+ pytest.skip("skipped because of router(s) failure")
+
+ r3 = tgen.gears["r3"]
+
+ # r3 is a candidate rp
+ expected = {"address": "fd00:0:0:3::3", "priority": 10}
+ test_func = partial(
+ topotest.router_json_cmp, r3, "show ipv6 pim bsr candidate-rp json", expected
+ )
+ _, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
+
+ assertmsg = "r3: bsr candidate rp mismatch"
+ assert result is None, assertmsg
+
+
+def test_pimv6_bsr_rp_info(request):
+ "Test IPv6 RP info state"
+ tgen = get_topogen()
+ tc_name = request.node.name
+ write_test_header(tc_name)
+
+ if tgen.routers_have_failure():
+ pytest.skip("skipped because of router(s) failure")
+
+ # At this point, all nodes, including r5 should have synced the RP state
+ step("Verify rp-info on r5 from BSR")
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "ffbb::0/64",
+ None,
+ "fd00:0:0:3::3",
+ "BSR",
+ False,
+ "ipv6",
+ True,
+ retry_timeout=90,
+ )
+ assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
+
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "ffbb::0/124",
+ None,
+ "fd00:0:0:3::4",
+ "BSR",
+ False,
+ "ipv6",
+ True,
+ retry_timeout=30,
+ )
+ assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
+
+ result = verify_pim_rp_info(
+ tgen,
+ None,
+ "r5",
+ "ffbb::0/108",
+ None,
+ "fd00:0:0:3::4",
+ "BSR",
+ False,
+ "ipv6",
+ True,
+ retry_timeout=30,
+ )
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
projects / mirror / frr.git / commitdiff