]>
Commit | Line | Data |
---|---|---|
a1e0142d RZ |
1 | #!/usr/bin/env python |
2 | ||
3 | # | |
4 | # test_bfd_profiles_topo1.py | |
5 | # Part of NetDEF Topology Tests | |
6 | # | |
7 | # Copyright (c) 2020 by | |
8 | # Network Device Education Foundation, Inc. ("NetDEF") | |
9 | # | |
10 | # Permission to use, copy, modify, and/or distribute this software | |
11 | # for any purpose with or without fee is hereby granted, provided | |
12 | # that the above copyright notice and this permission notice appear | |
13 | # in all copies. | |
14 | # | |
15 | # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES | |
16 | # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF | |
17 | # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR | |
18 | # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY | |
19 | # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, | |
20 | # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS | |
21 | # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE | |
22 | # OF THIS SOFTWARE. | |
23 | # | |
24 | ||
25 | """ | |
26 | test_bfd_profiles_topo1.py: Test the FRR BFD profile protocol integration. | |
27 | """ | |
28 | ||
29 | import os | |
30 | import sys | |
31 | import json | |
32 | from functools import partial | |
33 | import pytest | |
34 | ||
35 | # Save the Current Working Directory to find configuration files. | |
36 | CWD = os.path.dirname(os.path.realpath(__file__)) | |
37 | sys.path.append(os.path.join(CWD, "../")) | |
38 | ||
39 | # pylint: disable=C0413 | |
40 | # Import topogen and topotest helpers | |
41 | from lib import topotest | |
42 | from lib.topogen import Topogen, TopoRouter, get_topogen | |
43 | from lib.topolog import logger | |
44 | ||
5ad1fd54 DS |
45 | pytestmark = [pytest.mark.bfdd, pytest.mark.bgpd, pytest.mark.isisd, pytest.mark.ospfd] |
46 | ||
a53c08bc | 47 | |
a1e0142d RZ |
48 | def setup_module(mod): |
49 | "Sets up the pytest environment" | |
8db751b8 CH |
50 | |
51 | topodef = { | |
52 | "s1": ("r1", "r2"), | |
53 | "s2": ("r2", "r3"), | |
54 | "s3": ("r3", "r4"), | |
55 | "s4": ("r4", "r5"), | |
56 | "s5": ("r1", "r6"), | |
57 | } | |
58 | tgen = Topogen(topodef, mod.__name__) | |
a1e0142d RZ |
59 | tgen.start_topology() |
60 | ||
61 | router_list = tgen.routers() | |
e5f0ed14 | 62 | for rname, router in router_list.items(): |
a1e0142d RZ |
63 | daemon_file = "{}/{}/bfdd.conf".format(CWD, rname) |
64 | if os.path.isfile(daemon_file): | |
65 | router.load_config(TopoRouter.RD_BFD, daemon_file) | |
66 | ||
67 | daemon_file = "{}/{}/bgpd.conf".format(CWD, rname) | |
68 | if os.path.isfile(daemon_file): | |
69 | router.load_config(TopoRouter.RD_BGP, daemon_file) | |
70 | ||
71 | daemon_file = "{}/{}/isisd.conf".format(CWD, rname) | |
72 | if os.path.isfile(daemon_file): | |
73 | router.load_config(TopoRouter.RD_ISIS, daemon_file) | |
74 | ||
75 | daemon_file = "{}/{}/ospfd.conf".format(CWD, rname) | |
76 | if os.path.isfile(daemon_file): | |
77 | router.load_config(TopoRouter.RD_OSPF, daemon_file) | |
78 | ||
79 | daemon_file = "{}/{}/ospf6d.conf".format(CWD, rname) | |
80 | if os.path.isfile(daemon_file): | |
81 | router.load_config(TopoRouter.RD_OSPF6, daemon_file) | |
82 | ||
83 | daemon_file = "{}/{}/zebra.conf".format(CWD, rname) | |
84 | if os.path.isfile(daemon_file): | |
85 | router.load_config(TopoRouter.RD_ZEBRA, daemon_file) | |
86 | ||
87 | # Initialize all routers. | |
88 | tgen.start_router() | |
89 | ||
90 | ||
91 | def teardown_module(_mod): | |
92 | "Teardown the pytest environment" | |
93 | tgen = get_topogen() | |
94 | tgen.stop_topology() | |
95 | ||
701a0192 | 96 | |
a1e0142d RZ |
97 | def test_wait_protocols_convergence(): |
98 | "Wait for all protocols to converge" | |
99 | tgen = get_topogen() | |
100 | if tgen.routers_have_failure(): | |
101 | pytest.skip(tgen.errors) | |
102 | ||
103 | logger.info("waiting for protocols to converge") | |
104 | ||
105 | def expect_loopback_route(router, iptype, route, proto): | |
106 | "Wait until route is present on RIB for protocol." | |
701a0192 | 107 | logger.info("waiting route {} in {}".format(route, router)) |
a1e0142d RZ |
108 | test_func = partial( |
109 | topotest.router_json_cmp, | |
110 | tgen.gears[router], | |
701a0192 | 111 | "show {} route json".format(iptype), |
112 | {route: [{"protocol": proto}]}, | |
a1e0142d RZ |
113 | ) |
114 | _, result = topotest.run_and_expect(test_func, None, count=130, wait=1) | |
115 | assertmsg = '"{}" OSPF convergence failure'.format(router) | |
116 | assert result is None, assertmsg | |
117 | ||
a1e0142d | 118 | # Wait for R1 <-> R6 convergence. |
701a0192 | 119 | expect_loopback_route("r1", "ip", "10.254.254.6/32", "ospf") |
a1e0142d RZ |
120 | |
121 | # Wait for R6 <-> R1 convergence. | |
701a0192 | 122 | expect_loopback_route("r6", "ip", "10.254.254.1/32", "ospf") |
a1e0142d RZ |
123 | |
124 | # Wait for R2 <-> R3 convergence. | |
701a0192 | 125 | expect_loopback_route("r2", "ip", "10.254.254.3/32", "bgp") |
a1e0142d RZ |
126 | |
127 | # Wait for R3 <-> R2 convergence. | |
701a0192 | 128 | expect_loopback_route("r3", "ip", "10.254.254.2/32", "bgp") |
a1e0142d RZ |
129 | |
130 | # Wait for R3 <-> R4 convergence. | |
701a0192 | 131 | expect_loopback_route("r3", "ipv6", "2001:db8:3::/64", "isis") |
a1e0142d RZ |
132 | |
133 | # Wait for R4 <-> R3 convergence. | |
701a0192 | 134 | expect_loopback_route("r4", "ipv6", "2001:db8:1::/64", "isis") |
a1e0142d RZ |
135 | |
136 | # Wait for R4 <-> R5 convergence. | |
701a0192 | 137 | expect_loopback_route("r4", "ipv6", "2001:db8:3::/64", "ospf6") |
a1e0142d RZ |
138 | |
139 | # Wait for R5 <-> R4 convergence. | |
701a0192 | 140 | expect_loopback_route("r5", "ipv6", "2001:db8:2::/64", "ospf6") |
a1e0142d RZ |
141 | |
142 | ||
143 | def test_bfd_profile_values(): | |
144 | "Assert that the BFD peers can find themselves." | |
145 | tgen = get_topogen() | |
146 | if tgen.routers_have_failure(): | |
147 | pytest.skip(tgen.errors) | |
148 | ||
149 | logger.info("waiting for bfd peers to go up and checking profile values") | |
150 | ||
151 | for router in tgen.routers().values(): | |
152 | json_file = "{}/{}/bfd-peers-initial.json".format(CWD, router.name) | |
153 | expected = json.loads(open(json_file).read()) | |
154 | test_func = partial( | |
155 | topotest.router_json_cmp, router, "show bfd peers json", expected | |
156 | ) | |
157 | _, result = topotest.run_and_expect(test_func, None, count=12, wait=0.5) | |
158 | assertmsg = '"{}" JSON output mismatches'.format(router.name) | |
159 | assert result is None, assertmsg | |
160 | ||
161 | ||
162 | def test_memory_leak(): | |
163 | "Run the memory leak test and report results." | |
164 | tgen = get_topogen() | |
165 | if not tgen.is_memleak_enabled(): | |
166 | pytest.skip("Memory leak test/report is disabled") | |
167 | ||
168 | tgen.report_memory_leaks() | |
169 | ||
170 | ||
171 | if __name__ == "__main__": | |
172 | args = ["-s"] + sys.argv[1:] | |
173 | sys.exit(pytest.main(args)) |