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
|
#!/usr/bin/env python
# SPDX-License-Identifier: ISC
#
# test_ospf_unnumbered.py
#
# Copyright (c) 2019 by
# Cumulus Networks, Inc
# Donald Sharp
#
"""
test_ospf_unnumbered.py: Test the OSPF unnumbered.
"""
import os
import sys
from functools import partial
import pytest
import json
# 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
# Required to instantiate the topology builder class.
pytestmark = [pytest.mark.ospfd]
CWD = os.path.dirname(os.path.realpath(__file__))
def build_topo(tgen):
"Build function"
# Create 4 routers
for routern in range(1, 3):
tgen.add_router("r{}".format(routern))
# Create a empty network for router 1
switch = tgen.add_switch("s1")
switch.add_link(tgen.gears["r1"])
# Create a empty network for router 2
switch = tgen.add_switch("s2")
switch.add_link(tgen.gears["r2"])
# Interconect router 1, 2
switch = tgen.add_switch("s3")
switch.add_link(tgen.gears["r1"])
switch.add_link(tgen.gears["r2"])
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():
router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
)
router.load_config(
TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
)
# What is this? OSPF Unnumbered depends on the rp_filter
# being set appropriately( HA! )
# Effectively we are putting different /32's on the interface
# the multicast packet delivery is somewhat controlled by
# the rp_filter. Setting it to '0' allows the OS to pass
# up the mcast packet not destined for the local routers
# network.
topotest.sysctl_assure(tgen.net["r1"], "net.ipv4.conf.r1-eth1.rp_filter", 0)
topotest.sysctl_assure(tgen.net["r1"], "net.ipv4.conf.all.rp_filter", 0)
topotest.sysctl_assure(tgen.net["r2"], "net.ipv4.conf.r2-eth1.rp_filter", 0)
topotest.sysctl_assure(tgen.net["r2"], "net.ipv4.conf.all.rp_filter", 0)
# Initialize all routers.
tgen.start_router()
# tgen.mininet_cli()
def teardown_module(mod):
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
def test_ospf_convergence():
"Test OSPF daemon convergence and that we have received the ospf routes"
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip("skipped because of router(s) failure")
for router, rnode in tgen.routers().items():
logger.info('Waiting for router "%s" convergence', router)
json_file = "{}/{}/ospf-route.json".format(CWD, router)
expected = json.loads(open(json_file).read())
test_func = partial(
topotest.router_json_cmp, rnode, "show ip ospf route json", expected
)
_, result = topotest.run_and_expect(test_func, None, count=160, wait=0.5)
assertmsg = '"{}" JSON output mismatches'.format(router)
assert result is None, assertmsg
# tgen.mininet_cli()
def test_ospf_kernel_route():
"Test OSPF kernel route installation and we have the onlink success"
tgen = get_topogen()
if tgen.routers_have_failure():
pytest.skip("skipped because of router(s) failure")
rlist = tgen.routers().values()
for router in rlist:
logger.info('Checking OSPF IPv4 kernel routes in "%s"', router.name)
json_file = "{}/{}/v4_route.json".format(CWD, router.name)
expected = json.loads(open(json_file).read())
test_func = partial(
topotest.router_json_cmp, router, "show ip route json", expected
)
_, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
assertmsg = '"{}" JSON output mistmatches'.format(router)
assert result is None, assertmsg
# tgen.mininet_cli()
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))
|
