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
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
#!/usr/bin/env python
# SPDX-License-Identifier: ISC
#
# Copyright (c) 2023 by
# Adriano Marto Reis <adrianomarto@gmail.com>
#
import os
import sys
import json
import subprocess
from functools import partial
import pytest
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from lib.common_config import verify_rib
from lib.ospf import verify_ospf_rib
from _ast import Try
pytestmark = pytest.mark.ospfd
"""
A large set of routers are connected to the same switch. Each router shares a
single network. All shared networks must be reachable from all routers.
"""
TOPOLOGY = """
net1 net2 netN
--- --- ---
| | OSPF-passive |
| | |
+---+---+ +---+---+ +---+---+
| | | | | |
| r1 | | r2 | (...) | rN |
| | | | | |
+---+---+ +---+---+ +---+---+
| | OSPF-active |
| | |
+----------------+--------------------------------+
switch
"""
N_ROUTERS = 8
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
def build_topo(tgen):
"Build the topology"
# Create a single switch to connect all the routers
switch = tgen.add_switch("switch")
# Create routers
for router_id in range(1, N_ROUTERS + 1):
router = tgen.add_router(f"r{router_id}")
switch.add_link(router)
# The shared network needs to be connected to something
dummy = tgen.add_switch(f"s{router_id}")
dummy.add_link(router)
def setup_module(mod):
logger.info("OSPF single switch:\n {}".format(TOPOLOGY))
tgen = Topogen(build_topo, mod.__name__)
tgen.start_topology()
# Starting Routers
router_list = tgen.routers()
for rname, router in router_list.items():
logger.info("Loading router %s" % rname)
router.load_frr_config(os.path.join(CWD, "{}_frr.conf".format(rname)))
# Initialize all routers.
tgen.start_router()
def teardown_module(mod):
"Tear-down the test environment"
tgen = get_topogen()
tgen.stop_topology()
def is_iproute2_json_supported():
"""
Checks if the command 'ip -j route' is supported.
"""
try:
output = subprocess.run(
["ip", "-j", "route", "get", "0.0.0.0"], stdout=subprocess.PIPE
).stdout.decode()
json.loads(output)
return True
except json.decoder.JSONDecodeError:
return False
@pytest.mark.skipif(
not is_iproute2_json_supported(), reason="'ip -j route' not supported"
)
def test_all_routes_advertised():
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip("Skipped because of router(s) failure")
# networks advertised by each router and the expected next hops
networks = {
"r1": ("10.0.1.0/24", "203.0.113.1"),
"r2": ("10.0.2.0/24", "203.0.113.2"),
"r3": ("10.0.3.0/24", "203.0.113.3"),
"r4": ("10.0.4.0/24", "203.0.113.4"),
"r5": ("10.0.5.0/24", "203.0.113.5"),
"r6": ("10.0.6.0/24", "203.0.113.6"),
"r7": ("10.0.7.0/24", "203.0.113.7"),
"r8": ("10.0.8.0/24", "203.0.113.8"),
}
for router_orig in tgen.routers().keys():
for router_dest, network in networks.items():
if router_orig != router_dest:
input_dict = {
router_orig: {
"static_routes": [
{
"network": network[0],
}
]
}
}
result = verify_ospf_rib(
tgen, router_orig, input_dict, next_hop=network[1]
)
assert result is True, "Error: {}".format(result)
result = verify_rib(
tgen, "ipv4", router_orig, input_dict, next_hop=network[1]
)
assert result is True, "Error: {}".format(result)
check_route(router_orig, network[0], network[1])
def check_route(router_name, network, expected_nexthop):
"""
Checks if the given network is present on the given router and has the
expected next hop.
"""
tgen = get_topogen()
router = tgen.gears[router_name]
expected_response = {
network: [
{
"prefix": network,
"protocol": "ospf",
"nexthops": [
{
"ip": expected_nexthop,
"active": True,
},
],
},
],
}
test_func = partial(
topotest.router_json_cmp,
router,
f"show ip route {network} json",
expected_response,
)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=1)
assert (
result is None
), f"{router_name} (ospfd): no route {network} via {expected_nexthop}"
address = network.split("/")[0]
output = router.cmd(f"ip -j route get {address}")
logger.info(output)
routes = json.loads(output)
assert (
routes[0]["gateway"] == expected_nexthop
), f"{router_name} (kernel): no route {address} via {expected_nexthop}"
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))
|
