def bmp_check_for_peer_message(
- expected_peers, bmp_log_type, bmp_collector, bmp_log_file, is_rd_instance=False
+ expected_peers,
+ bmp_log_type,
+ bmp_collector,
+ bmp_log_file,
+ is_rd_instance=False,
+ peer_bgp_id=None,
):
"""
Check for the presence of a peer up message for the peer
for m in messages:
if is_rd_instance and m["peer_distinguisher"] == "0:0":
continue
+ if peer_bgp_id and m["peer_bgp_id"] != peer_bgp_id:
+ continue
if (
"peer_ip" in m.keys()
and m["peer_ip"] != "0.0.0.0"
assert success, "Checking the BMP peer up LOC-RIB message failed !."
+def test_bgp_routerid_changed():
+ """
+ Checking for BGP loc-rib up messages with new router-id
+ """
+ tgen = get_topogen()
+
+ tgen.gears["r1vrf"].vtysh_cmd(
+ """
+ configure terminal
+ router bgp 65501 vrf vrf1
+ bgp router-id 192.168.1.77
+ """
+ )
+
+ peers = ["0.0.0.0"]
+
+ logger.info(
+ "checking for BMP peer down LOC-RIB message with router-id set to 192.168.0.1."
+ )
+ test_func = partial(
+ bmp_check_for_peer_message,
+ peers,
+ "peer down",
+ tgen.gears["bmp1vrf"],
+ os.path.join(tgen.logdir, "bmp1vrf", "bmp.log"),
+ is_rd_instance=True,
+ )
+ success, _ = topotest.run_and_expect(test_func, True, count=30, wait=1)
+ assert (
+ success
+ ), "Checking the BMP peer down LOC-RIB message with router-id set to 192.168.0.1 failed !."
+
+ logger.info(
+ "checking for BMP peer up LOC-RIB message with router-id set to 192.168.1.77."
+ )
+ test_func = partial(
+ bmp_check_for_peer_message,
+ peers,
+ "peer up",
+ tgen.gears["bmp1vrf"],
+ os.path.join(tgen.logdir, "bmp1vrf", "bmp.log"),
+ is_rd_instance=True,
+ peer_bgp_id="192.168.1.77",
+ )
+ success, _ = topotest.run_and_expect(test_func, True, count=30, wait=1)
+ assert (
+ success
+ ), "Checking the BMP peer up LOC-RIB message with router-id set to 192.168.1.77 failed !."
+
+
if __name__ == "__main__":
args = ["-s"] + sys.argv[1:]
sys.exit(pytest.main(args))