2 # SPDX-License-Identifier: ISC
7 # Copyright (c) 2020 by
8 # Cumulus Networks, Inc.
13 test_pbr_topo1.py: Testing PBR
22 from functools
import partial
24 # Save the Current Working Directory to find configuration files.
25 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
26 sys
.path
.append(os
.path
.join(CWD
, "../"))
28 # pylint: disable=C0413
29 # Import topogen and topotest helpers
30 from lib
import topotest
31 from lib
.topogen
import Topogen
, TopoRouter
, get_topogen
32 from lib
.topolog
import logger
33 from lib
.common_config
import shutdown_bringup_interface
35 # Required to instantiate the topology builder class.
37 pytestmark
= [pytest
.mark
.pbrd
]
39 #####################################################
41 ## Network Topology Definition
43 #####################################################
50 for routern
in range(1, 2):
51 tgen
.add_router("r{}".format(routern
))
54 for switchn
in range(1, 6):
55 switch
= tgen
.add_switch("sw{}".format(switchn
))
56 switch
.add_link(tgen
.gears
["r1"])
59 #####################################################
63 #####################################################
66 def setup_module(module
):
68 tgen
= Topogen(build_topo
, module
.__name
__)
71 krel
= platform
.release()
72 if topotest
.version_cmp(krel
, "4.10") < 0:
73 tgen
.errors
= "Newer kernel than 4.9 needed for pbr tests"
74 pytest
.skip(tgen
.errors
)
76 router_list
= tgen
.routers()
77 for rname
, router
in router_list
.items():
78 # Install vrf into the kernel and slave eth3
79 router
.run("ip link add vrf-chiyoda type vrf table 1000")
80 router
.run("ip link set dev {}-eth3 master vrf-chiyoda".format(rname
))
81 router
.run("ip link set vrf-chiyoda up")
84 TopoRouter
.RD_ZEBRA
, os
.path
.join(CWD
, "{}/zebra.conf".format(rname
))
87 TopoRouter
.RD_PBRD
, os
.path
.join(CWD
, "{}/pbrd.conf".format(rname
))
93 def teardown_module(_mod
):
94 "Teardown the pytest environment"
97 # This function tears down the whole topology.
101 def test_converge_protocols():
102 "Wait for protocol convergence"
105 # Don't run this test if we have any failure.
106 if tgen
.routers_have_failure():
107 pytest
.skip(tgen
.errors
)
109 topotest
.sleep(5, "Waiting for PBR convergence")
113 "Test PBR 'show ip eigrp'"
116 # Don't run this test if we have any failure.
117 if tgen
.routers_have_failure():
118 pytest
.skip(tgen
.errors
)
121 logger
.info("Verifying PBR routes")
123 router_list
= tgen
.routers().values()
124 for router
in router_list
:
125 intf_file
= "{}/{}/pbr-interface.json".format(CWD
, router
.name
)
126 logger
.info(intf_file
)
128 # Read expected result from file
129 expected
= json
.loads(open(intf_file
).read())
131 # Actual output from router
133 topotest
.router_json_cmp
, router
, "show pbr interface json", expected
135 _
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
136 assertmsg
= '"show pbr interface" mismatches on {}'.format(router
.name
)
137 if result
is not None:
138 gather_pbr_data_on_error(router
)
139 assert result
is None, assertmsg
141 map_file
= "{}/{}/pbr-map.json".format(CWD
, router
.name
)
142 logger
.info(map_file
)
144 # Read expected result from file
145 expected
= json
.loads(open(map_file
).read())
147 # Actual output from router
149 topotest
.router_json_cmp
, router
, "show pbr map json", expected
151 _
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
152 assertmsg
= '"show pbr map" mismatches on {}'.format(router
.name
)
153 if result
is not None:
154 gather_pbr_data_on_error(router
)
155 assert result
is None, assertmsg
157 nexthop_file
= "{}/{}/pbr-nexthop-groups.json".format(CWD
, router
.name
)
158 logger
.info(nexthop_file
)
160 # Read expected result from file
161 expected
= json
.loads(open(nexthop_file
).read())
163 # Actual output from router
165 topotest
.router_json_cmp
, router
, "show pbr nexthop-groups json", expected
167 _
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
168 assertmsg
= '"show pbr nexthop-groups" mismatches on {}'.format(router
.name
)
169 if result
is not None:
170 gather_pbr_data_on_error(router
)
171 assert result
is None, assertmsg
175 "Test PBR interface flapping"
178 # Don't run this test if we have any failure.
179 if tgen
.routers_have_failure():
180 pytest
.skip(tgen
.errors
)
183 logger
.info("Flapping PBR Interfaces")
185 router_list
= tgen
.routers().values()
186 for router
in router_list
:
187 # Flap interface to see if route-map properties are intact
191 intf
= "r1-eth{}".format(i
)
193 # Down and back again
194 shutdown_bringup_interface(tgen
, router
.name
, intf
, False)
195 shutdown_bringup_interface(tgen
, router
.name
, intf
, True)
197 intf_file
= "{}/{}/pbr-interface.json".format(CWD
, router
.name
)
198 logger
.info(intf_file
)
200 # Read expected result from file
201 expected
= json
.loads(open(intf_file
).read())
203 # Actual output from router
205 topotest
.router_json_cmp
, router
, "show pbr interface json", expected
207 _
, result
= topotest
.run_and_expect(test_func
, None, count
=30, wait
=1)
208 assertmsg
= '"show pbr interface" mismatches on {}'.format(router
.name
)
209 if result
is not None:
210 gather_pbr_data_on_error(router
)
211 assert result
is None, assertmsg
214 def test_rule_linux_installation():
215 "Ensure that rule is installed in the kernel"
218 # Don't run this test if we have any failure.
219 if tgen
.routers_have_failure():
220 pytest
.skip(tgen
.errors
)
222 logger
.info("Checking for installed PBR rules in OS")
224 def _get_router_rules(router
, expected
):
225 actual
= topotest
.ip_rules(router
)
228 return topotest
.json_cmp(actual
, expected
)
230 router_list
= tgen
.routers().values()
231 for router
in router_list
:
232 rules_file
= "{}/{}/linux-rules.json".format(CWD
, router
.name
)
234 expected
= json
.loads(open(rules_file
).read())
236 test_func
= partial(_get_router_rules
, router
, expected
)
238 _
, result
= topotest
.run_and_expect(test_func
, None, count
=20, wait
=1)
239 assertmsg
= "Router {} OS rules mismatch".format(router
.name
)
240 assert result
is None, assertmsg
243 if __name__
== "__main__":
244 args
= ["-s"] + sys
.argv
[1:]
245 sys
.exit(pytest
.main(args
))
250 def gather_pbr_data_on_error(router
):
251 logger
.info(router
.vtysh_cmd("show ip route"))
252 logger
.info(router
.vtysh_cmd("show ip route vrf vrf-chiyoda"))
253 logger
.info(router
.vtysh_cmd("show ip nht"))
254 logger
.info(router
.vtysh_cmd("show pbr interface"))
255 logger
.info(router
.vtysh_cmd("show pbr map"))
256 logger
.info(router
.vtysh_cmd("show pbr nexthop-groups"))
257 logger
.info(router
.vtysh_cmd("show nexthop-group rib singleton ip"))
258 logger
.info(router
.vtysh_cmd("show nexthop-group rib singleton ipv6"))
259 logger
.info(router
.vtysh_cmd("show nexthop-group rib"))
260 logger
.info(router
.run("ip nexthop show"))
261 logger
.info(router
.run("ip route show"))
262 logger
.info(router
.run("ip route show table 1000"))
263 logger
.info(router
.run("ip route show table 10000"))
264 logger
.info(router
.run("ip -6 route show table 10000"))
265 logger
.info(router
.run("ip route show table 10001"))
266 logger
.info(router
.run("ip -6 route show table 10001"))
267 logger
.info(router
.run("ip route show table 10002"))
268 logger
.info(router
.run("ip -6 route show table 10002"))
269 logger
.info(router
.run("ip route show table 10003"))
270 logger
.info(router
.run("ip -6 route show table 10003"))
271 logger
.info(router
.run("ip route show table 10004"))
272 logger
.info(router
.run("ip -6 route show table 10004"))
273 logger
.info(router
.run("ip route show table 10005"))
274 logger
.info(router
.run("ip -6 route show table 10005"))
275 logger
.info(router
.run("ip rule show"))