proxmox-mirrors/mirror_frr
1
0
Fork 0
mirror of https://git.proxmox.com/git/mirror_frr synced 2025-08-13 21:10:28 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #12418 from opensourcerouting/fix/tests_for_while_true

Browse Source 
Fix old tests with `while true`
This commit is contained in:
Donald Sharp 2022-12-01 11:00:23 -05:00 committed by GitHub
commit 0ec0fee279
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 111 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
bgpd/bgp_route.c
View File

@ -4281,14 +4281,6 @@ int bgp_update(struct peer *peer, const struct prefix *p, uint32_t addpath_id,
&& (!CHECK_FLAG(dest->flags, BGP_NODE_FIB_INSTALLED)))
SET_FLAG(dest->flags, BGP_NODE_FIB_INSTALL_PENDING);
/* If maximum prefix count is configured and current prefix
* count exeed it.
*/
if (bgp_maximum_prefix_overflow(peer, afi, safi, 0)) {
bgp_attr_flush(&new_attr);
return -1;
}
/* If neighbor soo is configured, tag all incoming routes with
* this SoO tag and then filter out advertisements in
* subgroup_announce_check() if it matches the configured SoO
@ -4803,6 +4795,17 @@ int bgp_update(struct peer *peer, const struct prefix *p, uint32_t addpath_id,
bgp_path_info_set_flag(dest, new, BGP_PATH_VALID);
}
/* If maximum prefix count is configured and current prefix
* count exeed it.
*/
if (bgp_maximum_prefix_overflow(peer, afi, safi, 0)) {
reason = "maximum-prefix overflow";
bgp_attr_flush(&new_attr);
bgp_unlink_nexthop(new);
bgp_path_info_delete(dest, new);
goto filtered;
}
/* Addpath ID */
new->addpath_rx_id = addpath_id;

50
tests/topotests/bgp_comm_list_delete/test_bgp_comm-list_delete.py
View File

@ -34,11 +34,13 @@ import os
import sys
import json
import pytest
import functools
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
# pylint: disable=C0413
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
pytestmark = [pytest.mark.bgpd]
@ -81,30 +83,34 @@ def test_bgp_maximum_prefix_invalid():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
def _bgp_converge(router):
while True:
output = json.loads(
tgen.gears[router].vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")
)
if output["192.168.255.1"]["bgpState"] == "Established":
if (
output["192.168.255.1"]["addressFamilyInfo"]["ipv4Unicast"][
"acceptedPrefixCounter"
]
== 2
):
return True
r2 = tgen.gears["r2"]
def _bgp_comm_list_delete(router):
output = json.loads(
tgen.gears[router].vtysh_cmd("show ip bgp 172.16.255.254/32 json")
)
if "333:333" in output["paths"][0]["community"]["list"]:
return False
return True
def _bgp_converge():
output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
expected = {
"192.168.255.1": {
"bgpState": "Established",
"addressFamilyInfo": {
"ipv4Unicast": {
"acceptedPrefixCounter": 2,
}
},
}
}
return topotest.json_cmp(output, expected)
if _bgp_converge("r2"):
assert _bgp_comm_list_delete("r2") == True
test_func = functools.partial(_bgp_converge)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Can't converge initially"
def _bgp_comm_list_delete():
output = json.loads(r2.vtysh_cmd("show ip bgp 172.16.255.254/32 json"))
expected = {"paths": [{"community": {"list": ["333:333"]}}]}
return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_comm_list_delete)
_, result = topotest.run_and_expect(test_func, not None, count=60, wait=0.5)
assert result is not None, "333:333 community SHOULD be stripped from r1"
if __name__ == "__main__":

57
tests/topotests/bgp_local_as_private_remove/test_bgp_local_as_private_remove.py
View File

@ -31,13 +31,14 @@ used together with `remove-private-AS`.
import os
import sys
import json
import time
import pytest
import functools
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
# pylint: disable=C0413
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
pytestmark = [pytest.mark.bgpd]
@ -84,29 +85,43 @@ def test_bgp_remove_private_as():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
def _bgp_converge(router):
while True:
output = json.loads(
tgen.gears[router].vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")
)
if output["192.168.255.1"]["bgpState"] == "Established":
time.sleep(1)
return True
r2 = tgen.gears["r2"]
r4 = tgen.gears["r4"]
def _bgp_as_path(router):
output = json.loads(
tgen.gears[router].vtysh_cmd("show ip bgp 172.16.255.254/32 json")
)
if output["prefix"] == "172.16.255.254/32":
return output["paths"][0]["aspath"]["segments"][0]["list"]
def _bgp_converge():
output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
expected = {
"192.168.255.1": {
"bgpState": "Established",
}
}
return topotest.json_cmp(output, expected)
if _bgp_converge("r2"):
assert len(_bgp_as_path("r2")) == 1
assert 65000 not in _bgp_as_path("r2")
test_func = functools.partial(_bgp_converge)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Can't converge initially"
if _bgp_converge("r4"):
assert len(_bgp_as_path("r4")) == 2
assert 3000 in _bgp_as_path("r4")
def _bgp_as_path(router, asn_path, asn_length):
output = json.loads(router.vtysh_cmd("show ip bgp 172.16.255.254/32 json"))
expected = {
"paths": [
{
"aspath": {
"string": asn_path,
"length": asn_length,
}
}
]
}
return topotest.json_cmp(output, expected)
test_func = functools.partial(_bgp_as_path, r2, "500", 1)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Private ASNs not stripped"
test_func = functools.partial(_bgp_as_path, r4, "500 3000", 2)
_, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, "Private ASNs not stripped"
if __name__ == "__main__":

17
tests/topotests/bgp_maximum_prefix_invalid_update/r1/bgpd.conf
View File

@ -1,6 +1,13 @@
router bgp 65000
no bgp ebgp-requires-policy
neighbor 192.168.255.2 remote-as 65001
neighbor 192.168.255.2 timers 3 10
address-family ipv4 unicast
redistribute connected
no bgp ebgp-requires-policy
neighbor 192.168.255.2 remote-as 65001
neighbor 192.168.255.2 timers 3 10
address-family ipv4 unicast
redistribute connected
neighbor 192.168.255.2 prefix-list r2 out
exit-address-family
!
!
ip prefix-list r2 seq 5 permit 172.16.255.253/32
ip prefix-list r2 seq 10 permit 172.16.255.254/32
!

1
tests/topotests/bgp_maximum_prefix_invalid_update/r1/zebra.conf
View File

@ -1,6 +1,7 @@
!
interface lo
ip address 172.16.255.254/32
ip address 172.16.255.253/32
!
interface r1-eth0
ip address 192.168.255.1/24

13
tests/topotests/bgp_maximum_prefix_invalid_update/r2/bgpd.conf
View File

@ -1,6 +1,9 @@
router bgp 65001
no bgp ebgp-requires-policy
neighbor 192.168.255.1 remote-as 65000
neighbor 192.168.255.1 timers 3 10
address-family ipv4
neighbor 192.168.255.1 maximum-prefix 1
no bgp ebgp-requires-policy
neighbor 192.168.255.1 remote-as 65000
neighbor 192.168.255.1 timers 3 10
address-family ipv4
neighbor 192.168.255.1 maximum-prefix 1
exit-address-family
!
!

36
tests/topotests/bgp_maximum_prefix_invalid_update/test_bgp_maximum_prefix_invalid_update.py
View File

@ -36,11 +36,13 @@ import os
import sys
import json
import pytest
import functools
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
# pylint: disable=C0413
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
pytestmark = [pytest.mark.bgpd]
@ -83,29 +85,21 @@ def test_bgp_maximum_prefix_invalid():
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
def _bgp_converge(router):
while True:
output = json.loads(
tgen.gears[router].vtysh_cmd("show ip bgp neighbor 192.168.255.1 json")
)
if output["192.168.255.1"]["connectionsEstablished"] > 0:
return True
r2 = tgen.gears["r2"]
def _bgp_parsing_nlri(router):
cmd_max_exceeded = (
'grep "%MAXPFXEXCEED: No. of IPv4 Unicast prefix received" bgpd.log'
)
cmdt_error_parsing_nlri = 'grep "Error parsing NLRI" bgpd.log'
output_max_exceeded = tgen.gears[router].run(cmd_max_exceeded)
output_error_parsing_nlri = tgen.gears[router].run(cmdt_error_parsing_nlri)
def _bgp_parsing_nlri():
output = json.loads(r2.vtysh_cmd("show ip bgp neighbor 192.168.255.1 json"))
expected = {
"192.168.255.1": {
"lastNotificationReason": "Cease/Maximum Number of Prefixes Reached",
"lastResetDueTo": "BGP Notification send",
}
}
return topotest.json_cmp(output, expected)
if len(output_max_exceeded) > 0:
if len(output_error_parsing_nlri) > 0:
return False
return True
if _bgp_converge("r2"):
assert _bgp_parsing_nlri("r2") == True
test_func = functools.partial(_bgp_parsing_nlri)
_, result = topotest.run_and_expect(test_func, None, count=30, wait=0.5)
assert result is None, "Didn't send NOTIFICATION when hitting maximum-prefix"
if __name__ == "__main__":