1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
#!/usr/bin/env python
# SPDX-License-Identifier: ISC
#
# test_bfd_static_vrf.py
# Part of NetDEF Topology Tests
#
# Copyright 2025 6WIND S.A.
#
"""
test_bfd_static_vrf.py: Test the FRR static routes with BFD tracking.
"""
import os
import sys
import json
import platform
import functools
import pytest
pytestmark = [pytest.mark.staticd, pytest.mark.bfdd]
# 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
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
def build_topo(tgen):
"Build function"
# Create 3 routers
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["r1"])
switch.add_link(tgen.gears["r3"])
def setup_module(mod):
"Sets up the pytest environment"
tgen = Topogen(build_topo, mod.__name__)
tgen.start_topology()
router_list = tgen.routers()
for rname, router in router_list.items():
tgen.net[rname].cmd(
f"""
ip link add cust1 type vrf table 10
ip link set dev cust1 up
ip link set dev {rname}-eth0 master cust1
sysctl net.ipv6.conf.{rname}-eth0.keep_addr_on_down=1
"""
)
tgen.net["r1"].cmd(
"""
ip link set dev r1-eth1 master cust1
sysctl net.ipv6.conf.r1-eth1.keep_addr_on_down=1
"""
)
for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_frr_config(
os.path.join(CWD, "{}/frr.conf".format(rname)),
[
(TopoRouter.RD_ZEBRA, None),
(TopoRouter.RD_MGMTD, None),
(TopoRouter.RD_BFD, None),
(TopoRouter.RD_STATIC, None),
],
)
# Initialize all routers.
tgen.start_router()
def teardown_module(_mod):
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
def check_bfd_state(step=None):
tgen = get_topogen()
r1 = tgen.gears["r1"]
step_suffix = f"_step{step}" if step else ""
logger.info("Check BFD entries")
reffile = os.path.join(CWD, f"r1/show_bfd_peers{step_suffix}.json")
expected = json.loads(open(reffile).read())
cmd = "show bfd peers json"
test_func = functools.partial(topotest.router_json_cmp, r1, cmd, expected)
_, res = topotest.run_and_expect(test_func, None, count=30, wait=1)
assertmsg = f"BFD did not converge. Error on r1 {cmd}"
assert res is None, assertmsg
logger.info("Check IPv4 default route")
reffile = os.path.join(CWD, f"r1/show_ip_route{step_suffix}.json")
expected = json.loads(open(reffile).read())
cmd = "show ip route vrf cust1 0.0.0.0/0 json"
test_func = functools.partial(topotest.router_json_cmp, r1, cmd, expected)
_, res = topotest.run_and_expect(test_func, None, count=30, wait=1)
assertmsg = f"BFD did not converge. Error on r1 {cmd}"
assert res is None, assertmsg
logger.info("Check IPv6 default route")
reffile = os.path.join(CWD, f"r1/show_ipv6_route{step_suffix}.json")
expected = json.loads(open(reffile).read())
cmd = "show ipv6 route vrf cust1 ::/0 json"
test_func = functools.partial(topotest.router_json_cmp, r1, cmd, expected)
_, res = topotest.run_and_expect(test_func, None, count=30, wait=1)
assertmsg = f"BFD did not converge. Error on r1 {cmd}"
assert res is None, assertmsg
def test_bfd_convergence():
"Assert that the BFD peers can find themselves."
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
check_bfd_state()
def test_bfd_static_vrf_step1():
"""
Assert that BFD notices the link down after simulating network
failure.
"""
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
logger.info("Set r3-eth0 down")
tgen.gears["r3"].link_enable("r3-eth0", enabled=False)
check_bfd_state(step=1)
def test_bfd_static_vrf_step2():
"""
Assert that BFD goes back to the nominal stater after links are back up.
"""
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
logger.info("Set r3-eth0 up")
tgen.gears["r3"].link_enable("r3-eth0", enabled=True)
check_bfd_state()
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))
|
