topotests: add vrf test to bgp_rpki_topo1

Add vrf test to bgp_rpki_topo1

Signed-off-by: Louis Scalbert <louis.scalbert@6wind.com>
This commit is contained in:
Louis Scalbert 2023-12-18 13:04:11 +01:00
parent 98e965565b
commit 2b30c4ae92
8 changed files with 162 additions and 14 deletions

View File

@ -1 +1,5 @@
ip route 192.0.2.1/32 192.168.1.1
!
vrf vrf10
ip route 192.0.2.3/32 192.168.2.3
!

View File

@ -7,3 +7,6 @@ interface vrf10 vrf vrf10
interface r2-eth0
ip address 192.168.1.2/24
!
interface r2-eth1 vrf vrf10
ip address 192.168.2.2/24
!

View File

@ -0,0 +1,14 @@
router bgp 65530
no bgp ebgp-requires-policy
no bgp network import-check
neighbor 192.0.2.2 remote-as 65002
neighbor 192.0.2.2 timers 1 3
neighbor 192.0.2.2 timers connect 1
neighbor 192.0.2.2 ebgp-multihop 3
neighbor 192.0.2.2 update-source 192.0.2.3
address-family ipv4 unicast
network 198.51.100.0/24
network 203.0.113.0/24
network 10.0.0.0/24
exit-address-family
!

View File

@ -0,0 +1 @@
../r1/rtrd.py

View File

@ -0,0 +1 @@
ip route 192.0.2.2/32 192.168.2.2

View File

@ -0,0 +1 @@
../r1/vrps.csv
1 ../r1/vrps.csv

View File

@ -0,0 +1,5 @@
interface lo
ip address 192.0.2.3/32
!
interface r3-eth0
ip address 192.168.2.3/24

View File

@ -22,13 +22,17 @@ pytestmark = [pytest.mark.bgpd]
def build_topo(tgen):
for routern in range(1, 3):
for routern in range(1, 4):
tgen.add_router("r{}".format(routern))
switch = tgen.add_switch("s1")
switch.add_link(tgen.gears["r1"])
switch.add_link(tgen.gears["r2"])
switch = tgen.add_switch("s2")
switch.add_link(tgen.gears["r2"])
switch.add_link(tgen.gears["r3"])
def setup_module(mod):
tgen = Topogen(build_topo, mod.__name__)
@ -49,25 +53,34 @@ def setup_module(mod):
" -M bgpd_rpki" if rname == "r2" else "",
)
tgen.gears["r2"].run("ip link add vrf10 type vrf table 10")
tgen.gears["r2"].run("ip link set vrf10 up")
tgen.gears["r2"].run("ip link set r2-eth1 master vrf10")
tgen.start_router()
global rtrd_process
rtrd_process = {}
rname = "r1"
for rname in ["r1", "r3"]:
rtr_path = os.path.join(CWD, rname)
log_dir = os.path.join(tgen.logdir, rname)
log_file = os.path.join(log_dir, "rtrd.log")
rtr_path = os.path.join(CWD, rname)
log_dir = os.path.join(tgen.logdir, rname)
log_file = os.path.join(log_dir, "rtrd.log")
tgen.gears[rname].cmd("chmod u+x {}/rtrd.py".format(rtr_path))
rtrd_process = tgen.gears[rname].popen("{}/rtrd.py {}".format(rtr_path, log_file))
tgen.gears[rname].cmd("chmod u+x {}/rtrd.py".format(rtr_path))
rtrd_process[rname] = tgen.gears[rname].popen(
"{}/rtrd.py {}".format(rtr_path, log_file)
)
def teardown_module(mod):
tgen = get_topogen()
logger.info("r1: sending SIGTERM to rtrd RPKI server")
rtrd_process.kill()
for rname in ["r1", "r3"]:
logger.info("{}: sending SIGTERM to rtrd RPKI server".format(rname))
rtrd_process[rname].kill()
tgen.stop_topology()
@ -114,7 +127,7 @@ def test_show_bgp_rpki_prefixes():
for rname in ["r1", "r3"]:
logger.info("{}: checking if rtrd is running".format(rname))
if rtrd_process.poll() is not None:
if rtrd_process[rname].poll() is not None:
pytest.skip(tgen.errors)
rname = "r2"
@ -156,7 +169,7 @@ def test_show_bgp_rpki_prefixes_no_rpki_cache():
for rname in ["r1", "r3"]:
logger.info("{}: checking if rtrd is running".format(rname))
if rtrd_process.poll() is not None:
if rtrd_process[rname].poll() is not None:
pytest.skip(tgen.errors)
def _show_rpki_no_connection(rname):
@ -192,7 +205,7 @@ def test_show_bgp_rpki_prefixes_reconnect():
for rname in ["r1", "r3"]:
logger.info("{}: checking if rtrd is running".format(rname))
if rtrd_process.poll() is not None:
if rtrd_process[rname].poll() is not None:
pytest.skip(tgen.errors)
step("Restore RPKI server configuration")
@ -241,7 +254,7 @@ def test_show_bgp_rpki_route_map():
for rname in ["r1", "r3"]:
logger.info("{}: checking if rtrd is running".format(rname))
if rtrd_process.poll() is not None:
if rtrd_process[rname].poll() is not None:
pytest.skip(tgen.errors)
step("Apply RPKI valid route-map on neighbor")
@ -283,6 +296,112 @@ router bgp 65002
assert result is None, "Unexpected prefixes RPKI state on {}".format(rname)
def test_show_bgp_rpki_prefixes_vrf():
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ["r1", "r3"]:
logger.info("{}: checking if rtrd is running".format(rname))
if rtrd_process[rname].poll() is not None:
pytest.skip(tgen.errors)
step("Configure RPKI cache server on vrf10")
rname = "r2"
tgen.gears[rname].vtysh_cmd(
"""
configure
vrf vrf10
rpki
rpki cache 192.0.2.3 15432 preference 1
exit
exit
"""
)
step("Check vrf10 RPKI prefix table")
expected = open(os.path.join(CWD, "{}/rpki_prefix_table.json".format(rname))).read()
expected_json = json.loads(expected)
test_func = functools.partial(show_rpki_prefixes, rname, expected_json, vrf="vrf10")
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Failed to see RPKI prefixes on {}".format(rname)
for rpki_state in ["valid", "notfound", None]:
if rpki_state:
step(
"Check RPKI state of prefixes in vrf10 BGP table: {}".format(rpki_state)
)
else:
step("Check prefixes in vrf10 BGP table")
expected = open(
os.path.join(
CWD,
"{}/bgp_table_rpki_{}.json".format(
rname, rpki_state if rpki_state else "any"
),
)
).read()
expected_json = json.loads(expected)
test_func = functools.partial(
show_bgp_ipv4_table_rpki, rname, rpki_state, expected_json, vrf="vrf10"
)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Unexpected prefixes RPKI state on {}".format(rname)
def test_show_bgp_rpki_route_map_vrf():
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
for rname in ["r1", "r3"]:
logger.info("{}: checking if rtrd is running".format(rname))
if rtrd_process[rname].poll() is not None:
pytest.skip(tgen.errors)
step("Apply RPKI valid route-map on vrf10 neighbor")
rname = "r2"
tgen.gears[rname].vtysh_cmd(
"""
configure
router bgp 65002 vrf vrf10
address-family ipv4 unicast
neighbor 192.0.2.3 route-map RPKI in
"""
)
for rpki_state in ["valid", "notfound", None]:
if rpki_state:
step(
"Check RPKI state of prefixes in vrf10 BGP table: {}".format(rpki_state)
)
else:
step("Check prefixes in vrf10 BGP table")
expected = open(
os.path.join(
CWD,
"{}/bgp_table_rmap_rpki_{}.json".format(
rname, rpki_state if rpki_state else "any"
),
)
).read()
expected_json = json.loads(expected)
test_func = functools.partial(
show_bgp_ipv4_table_rpki,
rname,
rpki_state,
expected_json,
vrf="vrf10",
)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Unexpected prefixes RPKI state on {}".format(rname)
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))