proxmox-mirrors/mirror_frr
1
0
Fork 0
mirror of https://git.proxmox.com/git/mirror_frr synced 2025-07-25 11:04:22 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

tests: PIM AutoRP tests expanded

Browse Source 
Now with a full AutoRP implementation, we can test AutoRP in a full network setup
beginning with candidate RP announcements all the way through discovery and active RP
selection.

Signed-off-by: Nathan Bahr <nbahr@atcorp.com>
This commit is contained in:
Nathan Bahr 2024-11-01 19:15:52 +00:00
parent b6aa71b59b
commit 13c0722b5c
6 changed files with 716 additions and 240 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
tests/topotests/lib/pim.py
View File

@ -1720,26 +1720,26 @@ def verify_pim_rp_info(
)
return errormsg
if not iamRP:
if rp_json["iAmRP"] == False:
logger.info(
"[DUT %s]: Verifying group "
"and iAmNotRP [PASSED]!!"
" Found Expected: (%s, %s:%s)",
dut,
grp_addr,
"iAmRP",
rp_json["iAmRP"],
)
else:
errormsg = (
"[DUT %s]: Verifying group"
"%s and iAmRP [FAILED]!! "
"Expected: (iAmRP: %s),"
" Found: (iAmRP: %s)"
% (dut, grp_addr, "false", rp_json["iAmRP"])
)
return errormsg
if not iamRP:
if rp_json["iAmRP"] == False:
logger.info(
"[DUT %s]: Verifying group "
"and iAmNotRP [PASSED]!!"
" Found Expected: (%s, %s:%s)",
dut,
grp_addr,
"iAmRP",
rp_json["iAmRP"],
)
else:
errormsg = (
"[DUT %s]: Verifying group"
"%s and iAmRP [FAILED]!! "
"Expected: (iAmRP: %s),"
" Found: (iAmRP: %s)"
% (dut, grp_addr, "false", rp_json["iAmRP"])
)
return errormsg
logger.debug("Exiting lib API: {}".format(sys._getframe().f_code.co_name))
return True

12
tests/topotests/pim_autorp/r1/frr.conf
View File

@ -2,15 +2,25 @@
hostname r1
password zebra
log file /tmp/r1-frr.log
!
debug pim autorp
!
interface r1-eth0
ip address 10.10.76.1/24
ip address 10.0.0.1/24
ip igmp
ip pim
!
interface r1-eth1
ip address 10.0.1.1/24
ip igmp
ip pim
!
ip forwarding
!
ip route 10.0.2.0/24 10.0.0.2 50
ip route 10.0.3.0/24 10.0.0.2 50
!
router pim
autorp discovery
rp 10.0.3.4 224.0.1.0/24
!

12
tests/topotests/pim_autorp/r2/frr.conf
View File

@ -2,15 +2,25 @@
hostname r2
password zebra
log file /tmp/r2-frr.log
!
debug pim autorp
!
interface r2-eth0
ip address 10.10.76.2/24
ip address 10.0.0.2/24
ip igmp
ip pim
!
interface r2-eth1
ip address 10.0.2.2/24
ip igmp
ip pim
!
ip forwarding
!
ip route 10.0.1.0/24 10.0.0.1 50
ip route 10.0.3.0/24 10.0.2.4 50
!
router pim
autorp discovery
rp 10.0.3.4 224.0.1.0/24
!

26
tests/topotests/pim_autorp/r3/frr.conf Normal file
View File

@ -0,0 +1,26 @@
!
hostname r3
password zebra
log file /tmp/r3-frr.log
!
debug pim autorp
!
interface r3-eth0
ip address 10.0.1.3/24
ip igmp
ip pim
!
interface r3-eth1
ip address 10.0.3.3/24
ip igmp
ip pim
!
ip forwarding
!
ip route 10.0.0.0/24 10.0.1.1 50
ip route 10.0.2.0/24 10.0.3.4 50
!
router pim
autorp discovery
rp 10.0.3.4 224.0.1.0/24
!

26
tests/topotests/pim_autorp/r4/frr.conf Normal file
View File

@ -0,0 +1,26 @@
!
hostname r4
password zebra
log file /tmp/r4-frr.log
!
debug pim autorp
!
interface r4-eth0
ip address 10.0.2.4/24
ip igmp
ip pim
!
interface r4-eth1
ip address 10.0.3.4/24
ip igmp
ip pim
!
ip forwarding
!
ip route 10.0.0.0/24 10.0.2.2 50
ip route 10.0.1.0/24 10.0.2.2 50
!
router pim
autorp discovery
rp 10.0.3.4 224.0.1.0/24
!

840
tests/topotests/pim_autorp/test_pim_autorp.py
View File

@ -11,18 +11,14 @@
import os
import sys
import pytest
import json
from functools import partial
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
from lib.topogen import Topogen, get_topogen
from lib.topogen import Topogen, topotest, get_topogen
from lib.topolog import logger
from lib.pim import (
scapy_send_autorp_raw_packet,
verify_pim_rp_info,
verify_pim_rp_info_is_empty,
)
from lib.pim import verify_pim_rp_info
from lib.common_config import step, write_test_header
from time import sleep
@ -32,13 +28,26 @@ test_pim_autorp.py: Test general PIM AutoRP functionality
"""
TOPOLOGY = """
Basic AutoRP functionality
Test PIM AutoRP functionality:
AutoRP candidate RP announcements
Mapping agent announcement receive and send discovery
AutoRP discovery to active RP info
+---+---+ +---+---+
| | 10.10.76.0/24 | |
+ R1 + <------------------> + R2 |
| | .1 .2 | |
+---+---+ +---+---+
+---+---+ +---+---+
| | 10.0.0.0/24 | |
+ R1 +----------------------+ R2 |
| | .1 .2 | |
+---+---+ r1-eth0 r2-eth0 +---+---+
.1 | r1-eth1 r2-eth1 | .2
| |
10.0.1.0/24 | | 10.0.2.0/24
| |
.3 | r3-eth0 r4-eth0 | .4
+---+---+ r3-eth1 r4-eth1 +---+---+
| | .3 .4 | |
+ R3 +----------------------+ R4 |
| | 10.0.3.0/24 | |
+---+---+ +---+---+
"""
# Save the Current Working Directory to find configuration files.
@ -55,11 +64,14 @@ def build_topo(tgen):
# Create routers
tgen.add_router("r1")
tgen.add_router("r2")
tgen.add_router("r3")
tgen.add_router("r4")
# Create link between router 1 and 2
switch = tgen.add_switch("s1-2")
switch.add_link(tgen.gears["r1"])
switch.add_link(tgen.gears["r2"])
# Create topology links
tgen.add_link(tgen.gears["r1"], tgen.gears["r2"], "r1-eth0", "r2-eth0")
tgen.add_link(tgen.gears["r1"], tgen.gears["r3"], "r1-eth1", "r3-eth0")
tgen.add_link(tgen.gears["r2"], tgen.gears["r4"], "r2-eth1", "r4-eth0")
tgen.add_link(tgen.gears["r3"], tgen.gears["r4"], "r3-eth1", "r4-eth1")
def setup_module(mod):
@ -68,15 +80,6 @@ def setup_module(mod):
tgen = Topogen(build_topo, mod.__name__)
tgen.start_topology()
# Router 1 will be the router configured with "fake" autorp configuration, so give it a default route
# to router 2 so that routing to the RP address is not an issue
# r1_defrt_setup_cmds = [
# "ip route add default via 10.10.76.1 dev r1-eth0",
# ]
# for cmd in r1_defrt_setup_cmds:
# tgen.net["r1"].cmd(cmd)
logger.info("Testing PIM AutoRP support")
router_list = tgen.routers()
for rname, router in router_list.items():
logger.info("Loading router %s" % rname)
@ -95,8 +98,8 @@ def teardown_module(mod):
tgen.stop_topology()
def test_pim_autorp_discovery_single_rp(request):
"Test PIM AutoRP Discovery with single RP"
def test_pim_autorp_init(request):
"Test PIM AutoRP startup with only the static RP"
tgen = get_topogen()
tc_name = request.node.name
write_test_header(tc_name)
@ -104,37 +107,253 @@ def test_pim_autorp_discovery_single_rp(request):
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
step("Start with no RP configuration")
result = verify_pim_rp_info_is_empty(tgen, "r1")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Send AutoRP packet from r1 to r2")
# 1 RP(s), hold time 5 secs, 10.10.76.1, group(s) 224.0.0.0/4
data = "01005e00012800127f55cfb1080045c00030700c000008110abe0a0a4c01e000012801f001f0001c798b12010005000000000a0a4c0103010004e0000000"
scapy_send_autorp_raw_packet(tgen, "r1", "r1-eth0", data)
step("Verify rp-info from AutoRP packet")
result = verify_pim_rp_info(
tgen,
None,
"r2",
"224.0.0.0/4",
"r2-eth0",
"10.10.76.1",
"AutoRP",
False,
"ipv4",
True,
step("Verify start-up with no extra RP configuration")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
]
}"""
)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
step("Verify AutoRP configuration times out")
result = verify_pim_rp_info_is_empty(tgen, "r2")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Verify start-up with AutoRP only discovery enabled")
expected = json.loads(
"""
{
"discovery":{
"enabled": true
},
"announce": {
"enabled":false
},
"mapping-agent": {
"enabled":false
}
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim autorp json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format(
rtr
)
def test_pim_autorp_discovery_multiple_rp(request):
"Test PIM AutoRP Discovery with multiple RP's"
def test_pim_autorp_no_mapping_agent_rp(request):
"Test PIM AutoRP candidate with no mapping agent"
tgen = get_topogen()
tc_name = request.node.name
write_test_header(tc_name)
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
# Start a candidate RP on r2
tgen.routers()["r2"].vtysh_cmd(
"""
conf
router pim
autorp announce 10.0.0.2 224.0.0.0/4
autorp announce scope 31 interval 1 holdtime 5
"""
)
# Without a mapping agent, we should still have no RP
step("Verify no RP without mapping agent")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
]
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
step("Verify candidate RP in AutoRP on R2")
expected = json.loads(
"""
{
"discovery":{
"enabled": true
},
"announce": {
"enabled":true,
"scope":31,
"interval":1,
"holdtime":5,
"rpList":[
{
"rpAddress":"10.0.0.2",
"groupRange":"224.0.0.0/4",
"prefixList":"-"
}
]
},
"mapping-agent": {
"enabled":false
}
}"""
)
test_func = partial(
topotest.router_json_cmp, tgen.gears["r2"], "show ip pim autorp json", expected
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format("r2")
def test_pim_autorp_discovery_rp(request):
"Test PIM AutoRP candidate advertised by mapping agent"
tgen = get_topogen()
tc_name = request.node.name
write_test_header(tc_name)
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
# Start the mapping agent on R1
tgen.routers()["r1"].vtysh_cmd(
"""
conf
router pim
autorp send-rp-discovery source interface r1-eth0
autorp send-rp-discovery scope 31 interval 1 holdtime 5
"""
)
step("Verify rp-info of the only candidate RP")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
],
"10.0.0.2":[
{
"rpAddress":"10.0.0.2",
"group":"224.0.0.0/4",
"source":"AutoRP"
}
]
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
step("Verify mapping-agent in AutoRP on R1")
expected = json.loads(
"""
{
"announce": {
"enabled":false
},
"mapping-agent": {
"enabled":true,
"active":true,
"scope":31,
"interval":1,
"holdtime":5,
"source":"interface",
"interface":"r1-eth0",
"address":"10.0.0.1",
"rpList":{
"10.0.0.2":{
"rpAddress":"10.0.0.2",
"groupRanges":[
{
"negative":false,
"prefix":"224.0.0.0/4"
}
]
}
}
}
}"""
)
test_func = partial(
topotest.router_json_cmp, tgen.gears["r1"], "show ip pim autorp json", expected
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format("r1")
step("Verify AutoRP discovery RP's")
expected = json.loads(
"""
{
"discovery":{
"enabled": true,
"rpList":{
"10.0.0.2":{
"rpAddress":"10.0.0.2",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"224.0.0.0/4"
}
]
}
}
}
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim autorp json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format(
rtr
)
def test_pim_autorp_discovery_multiple_rp_same(request):
"Test PIM AutoRP Discovery with multiple RP's for same group prefix"
tgen = get_topogen()
tc_name = request.node.name
write_test_header(tc_name)
@ -142,42 +361,309 @@ def test_pim_autorp_discovery_multiple_rp(request):
if tgen.routers_have_failure():
pytest.skip("skipped because of router(s) failure")
step("Start with no RP configuration")
result = verify_pim_rp_info_is_empty(tgen, "r2")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Send AutoRP packet from r1 to r2")
# 2 RP(s), hold time 5 secs, 10.10.76.1, group(s) 224.0.0.0/8, 10.10.76.3, group(s) 225.0.0.0/8
data = "01005e00012800127f55cfb1080045c0003c700c000008110ab20a0a4c01e000012801f001f000283f5712020005000000000a0a4c0103010008e00000000a0a4c0303010008e1000000"
scapy_send_autorp_raw_packet(tgen, "r1", "r1-eth0", data)
step("Verify rp-info from AutoRP packet")
result = verify_pim_rp_info(
tgen,
None,
"r2",
"224.0.0.0/8",
"r2-eth0",
"10.10.76.1",
"AutoRP",
False,
"ipv4",
True,
# Start a candidate RP on r3
tgen.routers()["r3"].vtysh_cmd(
"""
conf
router pim
autorp announce 10.0.1.3 224.0.0.0/4
autorp announce scope 31 interval 1 holdtime 5
"""
)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
result = verify_pim_rp_info(
tgen,
None,
"r2",
"225.0.0.0/8",
"r2-eth0",
"10.10.76.3",
"AutoRP",
False,
"ipv4",
True,
# The new candidate RP has the same group range but a higher IP, they should all
# switch to this RP
step("Verify rp-info of the candidate RP with the higher IP")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
],
"10.0.1.3":[
{
"rpAddress":"10.0.1.3",
"group":"224.0.0.0/4",
"source":"AutoRP"
}
]
}"""
)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
step("Verify AutoRP discovery RP's")
expected = json.loads(
"""
{
"discovery":{
"enabled": true,
"rpList":{
"10.0.0.2":{
"rpAddress":"10.0.0.2",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"224.0.0.0/4"
}
]
},
"10.0.1.3":{
"rpAddress":"10.0.1.3",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"224.0.0.0/4"
}
]
}
}
}
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim autorp json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format(
rtr
)
def test_pim_autorp_discovery_multiple_rp_different(request):
"Test PIM AutoRP Discovery with multiple RP's for different group prefixes"
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")
# Switch R3 candidate to prefix list with different groups
step("Change R3 candidate to a prefix list")
tgen.routers()["r3"].vtysh_cmd(
"""
conf
ip prefix-list MYLIST permit 225.0.0.0/8
ip prefix-list MYLIST permit 226.0.0.0/8
router pim
autorp announce 10.0.1.3 group-list MYLIST
"""
)
# Now that R3 doesn't conflict, we should see both RP's
step("Verify rp-info of both candidate RP's")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
],
"10.0.0.2":[
{
"rpAddress":"10.0.0.2",
"group":"224.0.0.0/4",
"source":"AutoRP"
}
],
"10.0.1.3":[
{
"rpAddress":"10.0.1.3",
"prefixList":"__AUTORP_10.0.1.3__",
"source":"AutoRP"
}
]
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
step("Verify AutoRP discovery RP's")
expected = json.loads(
"""
{
"discovery":{
"enabled": true,
"rpList":{
"10.0.0.2":{
"rpAddress":"10.0.0.2",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"224.0.0.0/4"
}
]
},
"10.0.1.3":{
"rpAddress":"10.0.1.3",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"225.0.0.0/8"
},
{
"negative":false,
"prefix":"226.0.0.0/8"
}
]
}
}
}
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim autorp json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format(
rtr
)
def test_pim_autorp_discovery_neg_prefixes(request):
"Test PIM AutoRP Discovery with negative prefixes"
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")
# Add negative prefixes to the R3 candidate prefix list
step("Change R3 candidate prefix list to include overlapping negative prefixes")
tgen.routers()["r3"].vtysh_cmd(
"""
conf
ip prefix-list MYLIST deny 225.1.0.0/16
ip prefix-list MYLIST deny 226.1.0.0/16
"""
)
step("Verify rp-info stays the same")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
],
"10.0.0.2":[
{
"rpAddress":"10.0.0.2",
"group":"224.0.0.0/4",
"source":"AutoRP"
}
],
"10.0.1.3":[
{
"rpAddress":"10.0.1.3",
"prefixList":"__AUTORP_10.0.1.3__",
"source":"AutoRP"
}
]
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
step("Verify AutoRP discovery RP's")
expected = json.loads(
"""
{
"discovery":{
"enabled": true,
"rpList":{
"10.0.0.2":{
"rpAddress":"10.0.0.2",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"224.0.0.0/4"
}
]
},
"10.0.1.3":{
"rpAddress":"10.0.1.3",
"holdtime":5,
"groupRanges":[
{
"negative":false,
"prefix":"225.0.0.0/8"
},
{
"negative":false,
"prefix":"226.0.0.0/8"
},
{
"negative":true,
"prefix":"225.1.0.0/16"
},
{
"negative":true,
"prefix":"226.1.0.0/16"
}
]
}
}
}
}"""
)
for rtr in ["r1", "r2", "r3", "r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim autorp json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct autorp configuration".format(
rtr
)
def test_pim_autorp_discovery_static(request):
@ -189,142 +675,60 @@ def test_pim_autorp_discovery_static(request):
if tgen.routers_have_failure():
pytest.skip("skipped because of router(s) failure")
step("Start with no RP configuration")
result = verify_pim_rp_info_is_empty(tgen, "r2")
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Add static RP configuration to r2")
rnode = tgen.routers()["r2"]
rnode.cmd("vtysh -c 'conf t' -c 'router pim' -c 'rp 10.10.76.3 224.0.0.0/4'")
step("Verify static rp-info from r2")
result = verify_pim_rp_info(
tgen,
None,
"r2",
"224.0.0.0/4",
"r2-eth0",
"10.10.76.3",
"Static",
False,
"ipv4",
True,
)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
step("Send AutoRP packet from r1 to r2")
# 1 RP(s), hold time 5 secs, 10.10.76.1, group(s) 224.0.0.0/4
data = "01005e00012800127f55cfb1080045c00030700c000008110abe0a0a4c01e000012801f001f0001c798b12010005000000000a0a4c0103010004e0000000"
scapy_send_autorp_raw_packet(tgen, "r1", "r1-eth0", data)
step("Verify rp-info from AutoRP packet")
result = verify_pim_rp_info(
tgen,
None,
"r2",
"224.0.0.0/4",
"r2-eth0",
"10.10.76.1",
"AutoRP",
False,
"ipv4",
True,
)
assert result is True, "Testcase {} :Failed \n Error: {}".format(tc_name, result)
def test_pim_autorp_announce_cli(request):
"Test PIM AutoRP Announcement CLI commands"
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")
step("Add AutoRP announcement configuration to r1")
r1 = tgen.routers()["r1"]
r1.vtysh_cmd(
# Add in a static RP with a specific range and make sure both are used
step("Add static RP configuration to r4")
tgen.routers()["r4"].vtysh_cmd(
"""
conf
conf t
router pim
autorp announce holdtime 90
autorp announce interval 120
autorp announce scope 5
autorp announce 10.2.3.4 225.0.0.0/24
"""
)
expected = {
"discoveryEnabled": True,
"announce": {
"scope": 5,
"interval": 120,
"holdtime": 90,
"rpList": [
{"rpAddress": "10.2.3.4", "group": "225.0.0.0/24", "prefixList": ""}
],
},
}
test_func = partial(
topotest.router_json_cmp, r1, "show ip pim autorp json", expected
)
_, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assertmsg = '"{}" JSON output mismatches'.format(r1.name)
assert result is None, assertmsg
r1.vtysh_cmd(
rp 10.0.2.2 239.0.0.0/24
"""
conf
router pim
autorp announce 10.2.3.4 group-list ListA
"""
)
expected = {
"discoveryEnabled": True,
"announce": {
"scope": 5,
"interval": 120,
"holdtime": 90,
"rpList": [{"rpAddress": "10.2.3.4", "group": "", "prefixList": "ListA"}],
},
}
test_func = partial(
topotest.router_json_cmp, r1, "show ip pim autorp json", expected
step("Verify static rp-info from r4")
expected = json.loads(
"""
{
"10.0.3.4":[
{
"rpAddress":"10.0.3.4",
"group":"224.0.1.0/24",
"source":"Static"
}
],
"10.0.0.2":[
{
"rpAddress":"10.0.0.2",
"group":"224.0.0.0/4",
"source":"AutoRP"
}
],
"10.0.1.3":[
{
"rpAddress":"10.0.1.3",
"prefixList":"__AUTORP_10.0.1.3__",
"source":"AutoRP"
}
],
"10.0.2.2":[
{
"rpAddress":"10.0.2.2",
"group":"239.0.0.0/24",
"source":"Static"
}
]
}"""
)
_, result = topotest.run_and_expect(test_func, None, count=30, wait=1)
assertmsg = '"{}" JSON output mismatches'.format(r1.name)
assert result is None, assertmsg
def test_pim_autorp_announce_group(request):
"Test PIM AutoRP Announcement with a single group"
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")
step("Add candidate RP configuration to r1")
rnode = tgen.routers()["r1"]
rnode.cmd(
"vtysh -c 'conf t' -c 'router pim' -c 'send-rp-announce 10.10.76.1 224.0.0.0/4'"
)
step("Verify Announcement sent data")
# TODO: Verify AutoRP mapping agent receives candidate RP announcement
# Mapping agent is not yet implemented
# sleep(10)
step("Change AutoRP Announcement packet parameters")
rnode.cmd(
"vtysh -c 'conf t' -c 'router pim' -c 'send-rp-announce scope 8 interval 10 holdtime 60'"
)
step("Verify Announcement sent data")
# TODO: Verify AutoRP mapping agent receives updated candidate RP announcement
# Mapping agent is not yet implemented
# sleep(10)
for rtr in ["r4"]:
test_func = partial(
topotest.router_json_cmp,
tgen.gears[rtr],
"show ip pim rp-info json",
expected,
)
_, result = topotest.run_and_expect(test_func, None)
assert result is None, "{} does not have correct rp-info".format(rtr)
def test_memory_leak():