]>
git.proxmox.com Git - mirror_frr.git/blob - tests/topotests/ospfapi/test_ospf_clientapi.py
2 # -*- coding: utf-8 eval: (blacken-mode 1) -*-
4 # Copyright (c) 2021-2022, LabN Consulting, L.L.C.
6 # This program is free software; you can redistribute it and/or
7 # modify it under the terms of the GNU General Public License
8 # as published by the Free Software Foundation; either version 2
9 # of the License, or (at your option) any later version.
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU General Public License for more details.
16 # You should have received a copy of the GNU General Public License along
17 # with this program; see the file COPYING; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
22 test_ospf_clientapi.py: Test the OSPF client API.
32 from datetime
import datetime
, timedelta
36 from lib
.common_config
import retry
, run_frr_cmd
, step
37 from lib
.micronet
import comm_error
38 from lib
.topogen
import Topogen
, TopoRouter
39 from lib
.topotest
import interface_set_status
, json_cmp
41 pytestmark
= [pytest
.mark
.ospfd
]
43 CWD
= os
.path
.dirname(os
.path
.realpath(__file__
))
44 TESTDIR
= os
.path
.abspath(CWD
)
46 CLIENTDIR
= os
.path
.abspath(os
.path
.join(CWD
, "../../../ospfclient"))
47 if not os
.path
.exists(CLIENTDIR
):
48 CLIENTDIR
= os
.path
.join(CWD
, "/usr/lib/frr")
50 assert os
.path
.exists(
51 os
.path
.join(CLIENTDIR
, "ospfclient.py")
52 ), "can't locate ospfclient.py"
60 @pytest.fixture(scope
="function", name
="tgen")
62 "Setup/Teardown the environment and provide tgen argument to tests"
63 nrouters
= request
.param
65 topodef
= {"sw1:": ("r1",)}
67 topodef
= {f
"sw{i}": (f
"r{i}", f
"r{i+1}") for i
in range(1, nrouters
)}
69 tgen
= Topogen(topodef
, request
.module
.__name
__)
72 router_list
= tgen
.routers()
73 for _
, router
in router_list
.items():
74 router
.load_config(TopoRouter
.RD_ZEBRA
, "zebra.conf")
75 router
.load_config(TopoRouter
.RD_OSPF
, "ospfd.conf")
76 router
.net
.daemons_options
["ospfd"] = "--apiserver"
85 # Fixture that executes before each test
86 @pytest.fixture(autouse
=True)
87 def skip_on_failure(tgen
):
88 if tgen
.routers_have_failure():
89 pytest
.skip("skipped because of previous test failure")
97 @retry(retry_timeout
=45)
98 def verify_ospf_database(tgen
, dut
, input_dict
, cmd
="show ip ospf database json"):
100 show_ospf_json
= run_frr_cmd(dut
, cmd
, isjson
=True)
101 if not bool(show_ospf_json
):
102 return "ospf is not running"
103 result
= json_cmp(show_ospf_json
, input_dict
)
104 return str(result
) if result
else None
110 # logging.info("READING 1 CHAR")
113 return buf
if buf
else None
115 # logging.info("READ CHAR: '%s'", c)
120 def _wait_output(p
, regex
, timeout
=120):
121 retry_until
= datetime
.now() + timedelta(seconds
=timeout
)
122 while datetime
.now() < retry_until
:
123 # line = p.stdout.readline()
124 line
= myreadline(p
.stdout
)
126 assert None, "Timeout waiting for '{}'".format(regex
)
127 line
= line
.decode("utf-8")
130 logging
.debug("GOT LINE: '%s'", line
)
131 m
= re
.search(regex
, line
)
134 assert None, "Failed to get output withint {}s".format(timeout
)
142 def _test_reachability(tgen
, testbin
):
144 "192.168.0.1,192.168.0.2,192.168.0.4",
145 "192.168.0.2,192.168.0.4",
146 "192.168.0.1,192.168.0.2,192.168.0.4",
148 r2
= tgen
.gears
["r2"]
149 r3
= tgen
.gears
["r3"]
151 wait_args
= [f
"--wait={x}" for x
in waitlist
]
155 step("reachable: check for initial reachability")
157 ["/usr/bin/timeout", "120", testbin
, "-v", *wait_args
],
158 encoding
=None, # don't buffer
159 stdin
=subprocess
.DEVNULL
,
160 stdout
=subprocess
.PIPE
,
161 stderr
=subprocess
.STDOUT
,
163 _wait_output(p
, "SUCCESS: {}".format(waitlist
[0]))
165 step("reachable: check for modified reachability")
166 interface_set_status(r2
, "r2-eth0", False)
167 _wait_output(p
, "SUCCESS: {}".format(waitlist
[1]))
169 step("reachable: check for restored reachability")
170 interface_set_status(r2
, "r2-eth0", True)
171 _wait_output(p
, "SUCCESS: {}".format(waitlist
[2]))
172 except Exception as error
:
173 logging
.error("ERROR: %s", error
)
181 @pytest.mark
.parametrize("tgen", [4], indirect
=True)
182 def test_ospf_reachability(tgen
):
183 testbin
= os
.path
.join(TESTDIR
, "ctester.py")
184 rc
, o
, e
= tgen
.gears
["r2"].net
.cmd_status([testbin
, "--help"])
185 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", testbin
, rc
, o
, e
)
186 _test_reachability(tgen
, testbin
)
189 def _test_router_id(tgen
, testbin
):
190 r1
= tgen
.gears
["r1"]
197 mon_args
= [f
"--monitor={x}" for x
in waitlist
]
201 step("router id: check for initial router id")
203 ["/usr/bin/timeout", "120", testbin
, "-v", *mon_args
],
204 encoding
=None, # don't buffer
205 stdin
=subprocess
.DEVNULL
,
206 stdout
=subprocess
.PIPE
,
207 stderr
=subprocess
.STDOUT
,
209 _wait_output(p
, "SUCCESS: {}".format(waitlist
[0]))
211 step("router id: check for modified router id")
212 r1
.vtysh_multicmd("conf t\nrouter ospf\nospf router-id 1.1.1.1")
213 _wait_output(p
, "SUCCESS: {}".format(waitlist
[1]))
215 step("router id: check for restored router id")
216 r1
.vtysh_multicmd("conf t\nrouter ospf\nospf router-id 192.168.0.1")
217 _wait_output(p
, "SUCCESS: {}".format(waitlist
[2]))
218 except Exception as error
:
219 logging
.error("ERROR: %s", error
)
227 @pytest.mark
.parametrize("tgen", [2], indirect
=True)
228 def test_ospf_router_id(tgen
):
229 testbin
= os
.path
.join(TESTDIR
, "ctester.py")
230 rc
, o
, e
= tgen
.gears
["r1"].net
.cmd_status([testbin
, "--help"])
231 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", testbin
, rc
, o
, e
)
232 _test_router_id(tgen
, testbin
)
235 def _test_add_data(tgen
, apibin
):
236 "Test adding opaque data to domain"
238 r1
= tgen
.gears
["r1"]
240 step("add opaque: add opaque link local")
244 p
= r1
.popen([apibin
, "-v", "add,9,10.0.1.1,230,2,00000202"])
246 "routerId": "192.168.0.1",
249 "linkLocalOpaqueLsa": [
252 "advertisedRouter": "192.168.0.1",
253 "sequenceNumber": "80000001",
259 # Wait for it to show up
260 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
263 "linkLocalOpaqueLsa": {
267 "linkStateId": "230.0.0.2",
268 "advertisingRouter": "192.168.0.1",
269 "lsaSeqNumber": "80000001",
270 "opaqueData": "00000202",
277 json_cmd
= "show ip ospf da opaque-link json"
278 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
280 step("reset client, add opaque area, verify link local flushing")
282 p
.send_signal(signal
.SIGINT
)
286 p
= r1
.popen([apibin
, "-v", "add,10,1.2.3.4,231,1,00010101"])
288 "routerId": "192.168.0.1",
291 "linkLocalOpaqueLsa": [
294 "advertisedRouter": "192.168.0.1",
295 "sequenceNumber": "80000001",
299 "areaLocalOpaqueLsa": [
302 "advertisedRouter": "192.168.0.1",
303 "sequenceNumber": "80000001",
309 # Wait for it to show up
310 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
313 "areaLocalOpaqueLsa": {
317 "linkStateId": "231.0.0.1",
318 "advertisingRouter": "192.168.0.1",
319 "lsaSeqNumber": "80000001",
320 "opaqueData": "00010101",
327 json_cmd
= "show ip ospf da opaque-area json"
328 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
330 step("reset client, add opaque AS, verify area flushing")
332 p
.send_signal(signal
.SIGINT
)
337 p
= r1
.popen([apibin
, "-v", "add,11,232,3,deadbeaf01234567"])
339 "routerId": "192.168.0.1",
342 "areaLocalOpaqueLsa": [
345 "advertisedRouter": "192.168.0.1",
346 "sequenceNumber": "80000001",
352 "asExternalOpaqueLsa": [
355 "advertisedRouter": "192.168.0.1",
356 "sequenceNumber": "80000001",
360 # Wait for it to show up
361 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
364 "asExternalOpaqueLsa": [
366 "linkStateId": "232.0.0.3",
367 "advertisingRouter": "192.168.0.1",
368 "lsaSeqNumber": "80000001",
369 "opaqueData": "deadbeaf01234567",
374 json_cmd
= "show ip ospf da opaque-as json"
375 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
377 step("stop client, verify AS flushing")
379 p
.send_signal(signal
.SIGINT
)
385 "routerId": "192.168.0.1",
386 "asExternalOpaqueLsa": [
389 "advertisedRouter": "192.168.0.1",
390 "sequenceNumber": "80000001",
395 # Wait for it to be flushed
396 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
398 step("start client adding opaque domain, verify new sequence number and data")
401 p
= r1
.popen([apibin
, "-v", "add,11,232,3,ebadf00d"])
403 "routerId": "192.168.0.1",
404 "asExternalOpaqueLsa": [
407 "advertisedRouter": "192.168.0.1",
408 "sequenceNumber": "80000002",
412 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
415 "asExternalOpaqueLsa": [
417 "linkStateId": "232.0.0.3",
418 "advertisingRouter": "192.168.0.1",
419 "lsaSeqNumber": "80000002",
420 "opaqueData": "ebadf00d",
425 json_cmd
= "show ip ospf da opaque-as json"
426 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
428 p
.send_signal(signal
.SIGINT
)
446 @pytest.mark
.parametrize("tgen", [2], indirect
=True)
447 def test_ospf_opaque_add_data3(tgen
):
448 apibin
= os
.path
.join(CLIENTDIR
, "ospfclient.py")
449 rc
, o
, e
= tgen
.gears
["r2"].net
.cmd_status([apibin
, "--help"])
450 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", apibin
, rc
, o
, e
)
451 _test_add_data(tgen
, apibin
)
454 def _test_opaque_add_del(tgen
, apibin
):
455 "Test adding opaque data to domain"
457 r1
= tgen
.gears
["r1"]
458 r2
= tgen
.gears
["r2"]
463 step("reachable: check for add notification")
465 ["/usr/bin/timeout", "120", apibin
, "-v"],
466 encoding
=None, # don't buffer
467 stdin
=subprocess
.DEVNULL
,
468 stdout
=subprocess
.PIPE
,
469 stderr
=subprocess
.STDOUT
,
471 p
= r1
.popen([apibin
, "-v", "add,11,232,3,ebadf00d"])
473 # Wait for add notification
474 # RECV: LSA update msg for LSA 232.0.0.3 in area 0.0.0.0 seq 0x80000001 len 24 age 9
477 waitfor
= "RECV:.*update msg.*LSA {}.*age ([0-9]+)".format(ls_id
)
478 _
= _wait_output(pread
, waitfor
)
484 # step("reachable: check for flush/age out")
485 # # Wait for max age notification
486 # waitfor = "RECV:.*update msg.*LSA {}.*age 3600".format(ls_id)
487 # _wait_output(pread, waitfor)
489 step("reachable: check for delete")
490 # Wait for delete notification
491 waitfor
= "RECV:.*delete msg.*LSA {}.*".format(ls_id
)
492 _wait_output(pread
, waitfor
)
509 @pytest.mark
.parametrize("tgen", [2], indirect
=True)
510 def test_ospf_opaque_delete_data3(tgen
):
511 apibin
= os
.path
.join(CLIENTDIR
, "ospfclient.py")
512 rc
, o
, e
= tgen
.gears
["r2"].net
.cmd_status([apibin
, "--help"])
513 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", apibin
, rc
, o
, e
)
514 _test_opaque_add_del(tgen
, apibin
)
517 if __name__
== "__main__":
518 args
= ["-s"] + sys
.argv
[1:]
519 sys
.exit(pytest
.main(args
))