]>
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, 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
64 topodef
= {f
"sw{i}": (f
"r{i}", f
"r{i+1}") for i
in range(1, nrouters
)}
66 tgen
= Topogen(topodef
, request
.module
.__name
__)
69 router_list
= tgen
.routers()
70 for _
, router
in router_list
.items():
71 router
.load_config(TopoRouter
.RD_ZEBRA
, "zebra.conf")
72 router
.load_config(TopoRouter
.RD_OSPF
, "ospfd.conf")
73 router
.net
.daemons_options
["ospfd"] = "--apiserver"
82 # Fixture that executes before each test
83 @pytest.fixture(autouse
=True)
84 def skip_on_failure(tgen
):
85 if tgen
.routers_have_failure():
86 pytest
.skip("skipped because of previous test failure")
94 @retry(retry_timeout
=45)
95 def verify_ospf_database(tgen
, dut
, input_dict
, cmd
="show ip ospf database json"):
97 show_ospf_json
= run_frr_cmd(dut
, cmd
, isjson
=True)
98 if not bool(show_ospf_json
):
99 return "ospf is not running"
100 result
= json_cmp(show_ospf_json
, input_dict
)
101 return str(result
) if result
else None
107 # logging.info("READING 1 CHAR")
110 return buf
if buf
else None
112 # logging.info("READ CHAR: '%s'", c)
117 def _wait_output(p
, regex
, timeout
=120):
118 retry_until
= datetime
.now() + timedelta(seconds
=timeout
)
119 while datetime
.now() < retry_until
:
120 # line = p.stdout.readline()
121 line
= myreadline(p
.stdout
)
123 assert None, "Timeout waiting for '{}'".format(regex
)
124 line
= line
.decode("utf-8")
127 logging
.debug("GOT LINE: '%s'", line
)
128 m
= re
.search(regex
, line
)
131 assert None, "Failed to get output withint {}s".format(timeout
)
139 def _test_reachability(tgen
, testbin
):
141 "192.168.0.1,192.168.0.2,192.168.0.4",
142 "192.168.0.2,192.168.0.4",
143 "192.168.0.1,192.168.0.2,192.168.0.4",
145 r2
= tgen
.gears
["r2"]
146 r3
= tgen
.gears
["r3"]
148 wait_args
= [f
"--wait={x}" for x
in waitlist
]
152 step("reachable: check for initial reachability")
154 ["/usr/bin/timeout", "120", testbin
, "-v", *wait_args
],
155 encoding
=None, # don't buffer
156 stdin
=subprocess
.DEVNULL
,
157 stdout
=subprocess
.PIPE
,
158 stderr
=subprocess
.STDOUT
,
160 _wait_output(p
, "SUCCESS: {}".format(waitlist
[0]))
162 step("reachable: check for modified reachability")
163 interface_set_status(r2
, "r2-eth0", False)
164 _wait_output(p
, "SUCCESS: {}".format(waitlist
[1]))
166 step("reachable: check for restored reachability")
167 interface_set_status(r2
, "r2-eth0", True)
168 _wait_output(p
, "SUCCESS: {}".format(waitlist
[2]))
169 except Exception as error
:
170 logging
.error("ERROR: %s", error
)
178 @pytest.mark
.parametrize("tgen", [4], indirect
=True)
179 def test_ospf_reachability(tgen
):
180 testbin
= os
.path
.join(TESTDIR
, "ctester.py")
181 rc
, o
, e
= tgen
.gears
["r2"].net
.cmd_status([testbin
, "--help"])
182 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", testbin
, rc
, o
, e
)
183 _test_reachability(tgen
, testbin
)
186 def _test_add_data(tgen
, apibin
):
187 "Test adding opaque data to domain"
189 r1
= tgen
.gears
["r1"]
191 step("add opaque: add opaque link local")
195 p
= r1
.popen([apibin
, "-v", "add,9,10.0.1.1,230,2,00000202"])
197 "routerId": "192.168.0.1",
200 "linkLocalOpaqueLsa": [
203 "advertisedRouter": "192.168.0.1",
204 "sequenceNumber": "80000001",
210 # Wait for it to show up
211 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
214 "linkLocalOpaqueLsa": {
218 "linkStateId": "230.0.0.2",
219 "advertisingRouter": "192.168.0.1",
220 "lsaSeqNumber": "80000001",
221 "opaqueData": "00000202",
228 json_cmd
= "show ip ospf da opaque-link json"
229 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
231 step("reset client, add opaque area, verify link local flushing")
233 p
.send_signal(signal
.SIGINT
)
237 p
= r1
.popen([apibin
, "-v", "add,10,1.2.3.4,231,1,00010101"])
239 "routerId": "192.168.0.1",
242 "linkLocalOpaqueLsa": [
245 "advertisedRouter": "192.168.0.1",
246 "sequenceNumber": "80000001",
250 "areaLocalOpaqueLsa": [
253 "advertisedRouter": "192.168.0.1",
254 "sequenceNumber": "80000001",
260 # Wait for it to show up
261 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
264 "areaLocalOpaqueLsa": {
268 "linkStateId": "231.0.0.1",
269 "advertisingRouter": "192.168.0.1",
270 "lsaSeqNumber": "80000001",
271 "opaqueData": "00010101",
278 json_cmd
= "show ip ospf da opaque-area json"
279 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
281 step("reset client, add opaque AS, verify area flushing")
283 p
.send_signal(signal
.SIGINT
)
288 p
= r1
.popen([apibin
, "-v", "add,11,232,3,deadbeaf01234567"])
290 "routerId": "192.168.0.1",
293 "areaLocalOpaqueLsa": [
296 "advertisedRouter": "192.168.0.1",
297 "sequenceNumber": "80000001",
303 "asExternalOpaqueLsa": [
306 "advertisedRouter": "192.168.0.1",
307 "sequenceNumber": "80000001",
311 # Wait for it to show up
312 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
315 "asExternalOpaqueLsa": [
317 "linkStateId": "232.0.0.3",
318 "advertisingRouter": "192.168.0.1",
319 "lsaSeqNumber": "80000001",
320 "opaqueData": "deadbeaf01234567",
325 json_cmd
= "show ip ospf da opaque-as json"
326 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
328 step("stop client, verify AS flushing")
330 p
.send_signal(signal
.SIGINT
)
336 "routerId": "192.168.0.1",
337 "asExternalOpaqueLsa": [
340 "advertisedRouter": "192.168.0.1",
341 "sequenceNumber": "80000001",
346 # Wait for it to be flushed
347 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
349 step("start client adding opaque domain, verify new sequence number and data")
352 p
= r1
.popen([apibin
, "-v", "add,11,232,3,ebadf00d"])
354 "routerId": "192.168.0.1",
355 "asExternalOpaqueLsa": [
358 "advertisedRouter": "192.168.0.1",
359 "sequenceNumber": "80000002",
363 assert verify_ospf_database(tgen
, r1
, input_dict
) is None
366 "asExternalOpaqueLsa": [
368 "linkStateId": "232.0.0.3",
369 "advertisingRouter": "192.168.0.1",
370 "lsaSeqNumber": "80000002",
371 "opaqueData": "ebadf00d",
376 json_cmd
= "show ip ospf da opaque-as json"
377 assert verify_ospf_database(tgen
, r1
, input_dict
, json_cmd
) is None
379 p
.send_signal(signal
.SIGINT
)
397 @pytest.mark
.parametrize("tgen", [2], indirect
=True)
398 def test_ospf_opaque_add_data3(tgen
):
399 apibin
= os
.path
.join(CLIENTDIR
, "ospfclient.py")
400 rc
, o
, e
= tgen
.gears
["r2"].net
.cmd_status([apibin
, "--help"])
401 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", apibin
, rc
, o
, e
)
402 _test_add_data(tgen
, apibin
)
405 def _test_opaque_add_del(tgen
, apibin
):
406 "Test adding opaque data to domain"
408 r1
= tgen
.gears
["r1"]
409 r2
= tgen
.gears
["r2"]
414 step("reachable: check for add notification")
416 ["/usr/bin/timeout", "120", apibin
, "-v"],
417 encoding
=None, # don't buffer
418 stdin
=subprocess
.DEVNULL
,
419 stdout
=subprocess
.PIPE
,
420 stderr
=subprocess
.STDOUT
,
422 p
= r1
.popen([apibin
, "-v", "add,11,232,3,ebadf00d"])
424 # Wait for add notification
425 # RECV: LSA update msg for LSA 232.0.0.3 in area 0.0.0.0 seq 0x80000001 len 24 age 9
428 waitfor
= "RECV:.*update msg.*LSA {}.*age ([0-9]+)".format(ls_id
)
429 _
= _wait_output(pread
, waitfor
)
435 # step("reachable: check for flush/age out")
436 # # Wait for max age notification
437 # waitfor = "RECV:.*update msg.*LSA {}.*age 3600".format(ls_id)
438 # _wait_output(pread, waitfor)
440 step("reachable: check for delete")
441 # Wait for delete notification
442 waitfor
= "RECV:.*delete msg.*LSA {}.*".format(ls_id
)
443 _wait_output(pread
, waitfor
)
460 @pytest.mark
.parametrize("tgen", [2], indirect
=True)
461 def test_ospf_opaque_delete_data3(tgen
):
462 apibin
= os
.path
.join(CLIENTDIR
, "ospfclient.py")
463 rc
, o
, e
= tgen
.gears
["r2"].net
.cmd_status([apibin
, "--help"])
464 logging
.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", apibin
, rc
, o
, e
)
465 _test_opaque_add_del(tgen
, apibin
)
468 if __name__
== "__main__":
469 args
= ["-s"] + sys
.argv
[1:]
470 sys
.exit(pytest
.main(args
))