tests: Adding new test suite bgp_as_allow_in

1. Added 5 test cases to verify BGP AS-allow-in behavior in FRR
2. Enhanced framework to support BGP AS-allow-in config(lib/bgp.py)
3. Added API in bgp.py to verify BGP RIB table(lib/bgp.py)

Signed-off-by: Kuldeep Kashyap <kashyapk@vmware.com>
This commit is contained in:
Kuldeep Kashyap 2020-04-01 04:41:45 +00:00
parent fdf44b9038
commit 089f79edd4
2 changed files with 588 additions and 364 deletions

View File

@ -46,8 +46,8 @@ import pytest
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, '../'))
sys.path.append(os.path.join(CWD, '../lib/'))
sys.path.append(os.path.join(CWD, "../"))
sys.path.append(os.path.join(CWD, "../lib/"))
# pylint: disable=C0413
# Import topogen and topotest helpers
@ -65,13 +65,29 @@ from lib.topolog import logger
from lib.bgp import (
verify_bgp_convergence, create_router_bgp,
clear_bgp_and_verify, verify_bgp_rib
start_topology,
write_test_header,
write_test_footer,
reset_config_on_routers,
verify_rib,
create_static_routes,
create_route_maps,
check_address_types,
step,
)
from lib.topolog import logger
from lib.bgp import (
verify_bgp_convergence,
create_router_bgp,
clear_bgp_and_verify,
verify_bgp_rib,
)
from lib.topojson import build_topo_from_json, build_config_from_json
# Reading the data from JSON File for topology creation
jsonFile = "{}/bgp_as_allow_in.json".format(CWD)
try:
with open(jsonFile, 'r') as topoJson:
with open(jsonFile, "r") as topoJson:
topo = json.load(topoJson)
except IOError:
assert False, "Could not read file {}".format(jsonFile)
@ -128,8 +144,9 @@ def setup_module(mod):
# Api call verify whether BGP is converged
BGP_CONVERGENCE = verify_bgp_convergence(tgen, topo)
assert BGP_CONVERGENCE is True, ("setup_module :Failed \n Error:"
" {}".format(BGP_CONVERGENCE))
assert BGP_CONVERGENCE is True, "setup_module :Failed \n Error:" " {}".format(
BGP_CONVERGENCE
)
logger.info("Running setup_module() done")
@ -148,8 +165,9 @@ def teardown_module(mod):
# Stop toplogy and Remove tmp files
tgen.stop_topology()
logger.info("Testsuite end time: {}".
format(time.asctime(time.localtime(time.time()))))
logger.info(
"Testsuite end time: {}".format(time.asctime(time.localtime(time.time())))
)
logger.info("=" * 40)
@ -186,10 +204,7 @@ def test_bgp_allowas_in_p0(request):
input_dict_4 = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
@ -197,7 +212,8 @@ def test_bgp_allowas_in_p0(request):
logger.info("Configure static routes")
result = create_static_routes(tgen, input_dict_4)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("configure redistribute static in Router BGP in R1")
@ -206,11 +222,7 @@ def test_bgp_allowas_in_p0(request):
"bgp": {
"address_family": {
addr_type: {
"unicast": {
"redistribute": [{
"redist_type": "static"
}]
}
"unicast": {"redistribute": [{"redist_type": "static"}]}
}
}
}
@ -218,21 +230,31 @@ def test_bgp_allowas_in_p0(request):
}
result = create_router_bgp(tgen, topo, input_dict_2)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
step('Check BGP table of router R3 using "sh bgp ipv4" and "sh bgp '
'ipv6" command.')
step("We should not see prefix advertised from R1 in R3's BGP "
"table without allowas-in.")
logger.info("Verifying %s routes on r3, route should not be present",
addr_type)
result = verify_rib(tgen, addr_type, dut, input_dict_4,
tc_name, result
)
step(
'Check BGP table of router R3 using "sh bgp ipv4" and "sh bgp '
'ipv6" command.'
)
step(
"We should not see prefix advertised from R1 in R3's BGP "
"table without allowas-in."
)
logger.info("Verifying %s routes on r3, route should not be present", addr_type)
result = verify_rib(
tgen,
addr_type,
dut,
input_dict_4,
next_hop=NEXT_HOP_IP[addr_type],
protocol=protocol, expected=False)
protocol=protocol,
expected=False,
)
assert result is not True, "Testcase {} : Failed \n"
"Expected behavior: routes should not present in rib \n"
"Error: {}".format(tc_name, result)
step('Configure allowas-in on R3 for R2.')
step("Configure allowas-in on R3 for R2.")
step("We should see the prefix advertised from R1 in R3's BGP table.")
# Api call to enable allowas-in in bgp process.
input_dict_1 = {
@ -245,9 +267,7 @@ def test_bgp_allowas_in_p0(request):
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 1
}
"allowas-in": {"number_occurences": 1}
}
}
}
@ -260,11 +280,12 @@ def test_bgp_allowas_in_p0(request):
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, addr_type, dut, input_dict_4,
protocol=protocol)
tc_name, result
)
result = verify_rib(tgen, addr_type, dut, input_dict_4, protocol=protocol)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
write_test_footer(tc_name)
@ -300,10 +321,7 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
input_dict_4 = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
@ -311,7 +329,8 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
logger.info("Configure static routes")
result = create_static_routes(tgen, input_dict_4)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("configure redistribute static in Router BGP in R1")
@ -320,11 +339,7 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
"bgp": {
"address_family": {
addr_type: {
"unicast": {
"redistribute": [{
"redist_type": "static"
}]
}
"unicast": {"redistribute": [{"redist_type": "static"}]}
}
}
}
@ -332,24 +347,21 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
}
result = create_router_bgp(tgen, topo, input_dict_2)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step('Configure allowas-in on R3 for R2 under IPv4 addr-family only')
step("Configure allowas-in on R3 for R2 under IPv4 addr-family only")
# Api call to enable allowas-in in bgp process.
input_dict_1 = {
"r3": {
"bgp": {
"address_family": {
'ipv4': {
"ipv4": {
"unicast": {
"neighbor": {
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 1
}
}
"r3": {"allowas-in": {"number_occurences": 1}}
}
}
}
@ -360,16 +372,12 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
}
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
static_route_ipv4 = {
"r1": {
"static_routes": [
{
"network": NETWORK['ipv4'],
"next_hop": NEXT_HOP_IP['ipv4']
}
{"network": NETWORK["ipv4"], "next_hop": NEXT_HOP_IP["ipv4"]}
]
}
}
@ -377,33 +385,28 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
static_route_ipv6 = {
"r1": {
"static_routes": [
{
"network": NETWORK['ipv6'],
"next_hop": NEXT_HOP_IP['ipv6']
}
{"network": NETWORK["ipv6"], "next_hop": NEXT_HOP_IP["ipv6"]}
]
}
}
step("We should see R1 advertised prefix only in IPv4 AFI "
"not in IPv6 AFI.")
result = verify_rib(tgen, 'ipv4', dut, static_route_ipv4,
protocol=protocol)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, 'ipv6', dut, static_route_ipv6,
protocol=protocol, expected=False)
step("We should see R1 advertised prefix only in IPv4 AFI " "not in IPv6 AFI.")
result = verify_rib(tgen, "ipv4", dut, static_route_ipv4, protocol=protocol)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
result = verify_rib(
tgen, "ipv6", dut, static_route_ipv6, protocol=protocol, expected=False
)
assert result is not True, "Testcase {} : Failed \n"
"Expected behavior: routes are should not be present in ipv6 rib\n"
" Error: {}".format(tc_name, result)
step("Repeat the same test for IPv6 AFI.")
step('Configure allowas-in on R3 for R2 under IPv6 addr-family only')
step("Configure allowas-in on R3 for R2 under IPv6 addr-family only")
# Api call to enable allowas-in in bgp process.
input_dict_1 = {
"r3": {
"bgp": {
"address_family": {
'ipv4': {
"ipv4": {
"unicast": {
"neighbor": {
"r2": {
@ -411,7 +414,7 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
"r3": {
"allowas-in": {
"number_occurences": 2,
"delete": True
"delete": True,
}
}
}
@ -419,39 +422,32 @@ def test_bgp_allowas_in_per_addr_family_p0(request):
}
}
},
'ipv6': {
"ipv6": {
"unicast": {
"neighbor": {
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 2
}
}
}
"r3": {"allowas-in": {"number_occurences": 2}}
}
}
}
}
},
}
}
}
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
step("We should see R1 advertised prefix only in IPv6 AFI "
"not in IPv4 AFI.")
result = verify_rib(tgen, 'ipv4', dut, static_route_ipv4,
protocol=protocol, expected=False)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
step("We should see R1 advertised prefix only in IPv6 AFI " "not in IPv4 AFI.")
result = verify_rib(
tgen, "ipv4", dut, static_route_ipv4, protocol=protocol, expected=False
)
assert result is not True, "Testcase {} : Failed \n"
"Expected behavior: routes should not be present in ipv4 rib\n"
" Error: {}".format(tc_name, result)
result = verify_rib(tgen, 'ipv6', dut, static_route_ipv6,
protocol=protocol)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, "ipv6", dut, static_route_ipv6, protocol=protocol)
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
write_test_footer(tc_name)
@ -481,10 +477,7 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
static_routes = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
@ -492,7 +485,8 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
logger.info("Configure static routes")
result = create_static_routes(tgen, static_routes)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("configure redistribute static in Router BGP in R1")
@ -501,11 +495,7 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
"bgp": {
"address_family": {
addr_type: {
"unicast": {
"redistribute": [{
"redist_type": "static"
}]
}
"unicast": {"redistribute": [{"redist_type": "static"}]}
}
}
}
@ -513,28 +503,32 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
}
result = create_router_bgp(tgen, topo, input_dict_2)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step('Configure a route-map on R1 to prepend AS 4 times.')
step("Configure a route-map on R1 to prepend AS 4 times.")
for addr_type in ADDR_TYPES:
input_dict_4 = {
"r1": {
"route_maps": {
"ASP_{}".format(addr_type): [{
"ASP_{}".format(addr_type): [
{
"action": "permit",
"set": {
"aspath": {
"path": {
"as_num": "200 200 200 200",
"as_action": "prepend"
"as_action": "prepend",
}
},
}
}]
]
}
}
}
result = create_route_maps(tgen, input_dict_4)
assert result is True, 'Testcase {} : Failed \n Error: {}'.format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result
)
step("configure route map in out direction on R1")
# Configure neighbor for route map
@ -550,10 +544,10 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
"r1": {
"route_maps": [
{
"name":
"ASP_{}".format(
addr_type),
"direction": "out"
"name": "ASP_{}".format(
addr_type
),
"direction": "out",
}
]
}
@ -569,7 +563,8 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
result = create_router_bgp(tgen, topo, input_dict_7)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
for addr_type in ADDR_TYPES:
step('Configure "allowas-in 4" on R3 for R2.')
@ -584,9 +579,7 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 4
}
"allowas-in": {"number_occurences": 4}
}
}
}
@ -599,13 +592,14 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
result = verify_rib(tgen, addr_type, dut, static_routes,
protocol=protocol, expected=False)
tc_name, result
)
result = verify_rib(
tgen, addr_type, dut, static_routes, protocol=protocol, expected=False
)
assert result is not True, "Testcase {} : Failed \n "
"Expected behavior: routes are should not be present in rib\n"
"Error: {}".format(
tc_name, result)
"Error: {}".format(tc_name, result)
for addr_type in ADDR_TYPES:
step('Configure "allowas-in 5" on R3 for R2.')
@ -619,9 +613,7 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 5
}
"allowas-in": {"number_occurences": 5}
}
}
}
@ -634,21 +626,19 @@ def test_bgp_allowas_in_no_of_occurrences_p0(request):
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
static_routes = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
result = verify_rib(tgen, addr_type, dut, static_routes,
protocol=protocol)
result = verify_rib(tgen, addr_type, dut, static_routes, protocol=protocol)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
write_test_footer(tc_name)
@ -678,10 +668,7 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
static_routes = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
@ -689,7 +676,8 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
logger.info("Configure static routes")
result = create_static_routes(tgen, static_routes)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("configure redistribute static in Router BGP in R1")
@ -698,11 +686,7 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
"bgp": {
"address_family": {
addr_type: {
"unicast": {
"redistribute": [{
"redist_type": "static"
}]
}
"unicast": {"redistribute": [{"redist_type": "static"}]}
}
}
}
@ -710,28 +694,29 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
}
result = create_router_bgp(tgen, topo, input_dict_2)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step('Configure a route-map on R2 to prepend AS 2 times.')
step("Configure a route-map on R2 to prepend AS 2 times.")
for addr_type in ADDR_TYPES:
input_dict_4 = {
"r2": {
"route_maps": {
"ASP_{}".format(addr_type): [{
"ASP_{}".format(addr_type): [
{
"action": "permit",
"set": {
"aspath": {
"as_num": "200 200",
"as_action": "prepend"
"path": {"as_num": "200 200", "as_action": "prepend"}
},
}
}
}]
]
}
}
}
result = create_route_maps(tgen, input_dict_4)
assert result is True, 'Testcase {} : Failed \n Error: {}'.format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result
)
step("configure route map in out direction on R2")
# Configure neighbor for route map
@ -747,10 +732,11 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
"r2": {
"route_maps": [
{
"name":
"ASP_{}".format(
addr_type),
"direction": "out"}
"name": "ASP_{}".format(
addr_type
),
"direction": "out",
}
]
}
}
@ -765,7 +751,8 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
result = create_router_bgp(tgen, topo, input_dict_7)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step('Configure "allowas-in 3" on R3 for R1.')
input_dict_1 = {
@ -778,9 +765,7 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 3
}
"allowas-in": {"number_occurences": 3}
}
}
}
@ -793,7 +778,8 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
input_dict_1 = {
"r4": {
@ -805,9 +791,7 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
"r3": {
"dest_link": {
"r4": {
"allowas-in": {
"number_occurences": 3
}
"allowas-in": {"number_occurences": 3}
}
}
}
@ -820,24 +804,22 @@ def test_bgp_allowas_in_sameastoibgp_p1(request):
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
static_routes = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
dut = "r4"
aspath = "100 200 200 200"
result = verify_bgp_rib(tgen, addr_type, dut, static_routes,
aspath=aspath)
path = "100 200 200 200"
result = verify_bgp_rib(tgen, addr_type, dut, static_routes, aspath=path)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
write_test_footer(tc_name)
@ -867,10 +849,7 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
static_routes = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
@ -878,7 +857,8 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
logger.info("Configure static routes")
result = create_static_routes(tgen, static_routes)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step("configure redistribute static in Router BGP in R1")
@ -887,11 +867,7 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
"bgp": {
"address_family": {
addr_type: {
"unicast": {
"redistribute": [{
"redist_type": "static"
}]
}
"unicast": {"redistribute": [{"redist_type": "static"}]}
}
}
}
@ -899,28 +875,29 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
}
result = create_router_bgp(tgen, topo, input_dict_2)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
step('Configure a route-map on R2 to prepend AS 2 times.')
step("Configure a route-map on R2 to prepend AS 2 times.")
for addr_type in ADDR_TYPES:
input_dict_4 = {
"r2": {
"route_maps": {
"ASP_{}".format(addr_type): [{
"ASP_{}".format(addr_type): [
{
"action": "permit",
"set": {
"aspath": {
"as_num": "200 200",
"as_action": "prepend"
"path": {"as_num": "200 200", "as_action": "prepend"}
},
}
}
}]
]
}
}
}
result = create_route_maps(tgen, input_dict_4)
assert result is True, 'Testcase {} : Failed \n Error: {}'.format(
tc_name, result)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result
)
step("configure route map in out direction on R2")
# Configure neighbor for route map
@ -936,10 +913,11 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
"r2": {
"route_maps": [
{
"name":
"ASP_{}".format(
addr_type),
"direction": "out"}
"name": "ASP_{}".format(
addr_type
),
"direction": "out",
}
]
}
}
@ -954,7 +932,8 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
result = create_router_bgp(tgen, topo, input_dict_7)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
for addr_type in ADDR_TYPES:
step('Configure "allowas-in 3" on R3 for R1.')
@ -968,9 +947,7 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
"r2": {
"dest_link": {
"r3": {
"allowas-in": {
"number_occurences": 3
}
"allowas-in": {"number_occurences": 3}
}
}
}
@ -983,23 +960,21 @@ def test_bgp_allowas_in_sameastoebgp_p1(request):
}
result = create_router_bgp(tgen, topo, input_dict_1)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
static_routes = {
"r1": {
"static_routes": [
{
"network": NETWORK[addr_type],
"next_hop": NEXT_HOP_IP[addr_type]
}
{"network": NETWORK[addr_type], "next_hop": NEXT_HOP_IP[addr_type]}
]
}
}
dut = "r5"
aspath = "200 100 200 200 200"
result = verify_bgp_rib(tgen, addr_type, dut, static_routes,
aspath=aspath)
path = "200 100 200 200 200"
result = verify_bgp_rib(tgen, addr_type, dut, static_routes, aspath=path)
assert result is True, "Testcase {} : Failed \n Error: {}".format(
tc_name, result)
tc_name, result
)
write_test_footer(tc_name)

View File

@ -1953,3 +1953,252 @@ def verify_bgp_rib(tgen, addr_type, dut, input_dict, next_hop=None, aspath=None)
logger.debug("Exiting lib API: verify_bgp_rib()")
return True
@retry(attempts=5, wait=2, return_is_str=True, initial_wait=2)
def verify_bgp_rib(tgen, addr_type, dut, input_dict, next_hop=None, aspath=None):
"""
This API is to verify whether bgp rib has any
matching route for a nexthop.
Parameters
----------
* `tgen`: topogen object
* `dut`: input dut router name
* `addr_type` : ip type ipv4/ipv6
* `input_dict` : input dict, has details of static routes
* `next_hop`[optional]: next_hop which needs to be verified,
default = static
* 'aspath'[optional]: aspath which needs to be verified
Usage
-----
dut = 'r1'
next_hop = "192.168.1.10"
input_dict = topo['routers']
aspath = "100 200 300"
result = verify_bgp_rib(tgen, addr_type, dut, tgen, input_dict,
next_hop, aspath)
Returns
-------
errormsg(str) or True
"""
logger.debug("Entering lib API: verify_bgp_rib()")
router_list = tgen.routers()
additional_nexthops_in_required_nhs = []
list1 = []
list2 = []
for routerInput in input_dict.keys():
for router, rnode in router_list.iteritems():
if router != dut:
continue
# Verifying RIB routes
command = "show bgp"
# Static routes
sleep(2)
logger.info("Checking router {} BGP RIB:".format(dut))
if "static_routes" in input_dict[routerInput]:
static_routes = input_dict[routerInput]["static_routes"]
for static_route in static_routes:
found_routes = []
missing_routes = []
st_found = False
nh_found = False
vrf = static_route.setdefault("vrf", None)
if vrf:
cmd = "{} vrf {} {}".format(command, vrf, addr_type)
else:
cmd = "{} {}".format(command, addr_type)
cmd = "{} json".format(cmd)
rib_routes_json = run_frr_cmd(rnode, cmd, isjson=True)
# Verifying output dictionary rib_routes_json is not empty
if bool(rib_routes_json) == False:
errormsg = "No route found in rib of router {}..".format(router)
return errormsg
network = static_route["network"]
if "no_of_ip" in static_route:
no_of_ip = static_route["no_of_ip"]
else:
no_of_ip = 1
# Generating IPs for verification
ip_list = generate_ips(network, no_of_ip)
for st_rt in ip_list:
st_rt = str(ipaddr.IPNetwork(unicode(st_rt)))
_addr_type = validate_ip_address(st_rt)
if _addr_type != addr_type:
continue
if st_rt in rib_routes_json["routes"]:
st_found = True
found_routes.append(st_rt)
if next_hop:
if not isinstance(next_hop, list):
next_hop = [next_hop]
list1 = next_hop
found_hops = [
rib_r["ip"]
for rib_r in rib_routes_json["routes"][st_rt][0][
"nexthops"
]
]
list2 = found_hops
missing_list_of_nexthops = set(list2).difference(list1)
additional_nexthops_in_required_nhs = set(
list1
).difference(list2)
if list2:
if additional_nexthops_in_required_nhs:
logger.info(
"Missing nexthop %s for route"
" %s in RIB of router %s\n",
additional_nexthops_in_required_nhs,
st_rt,
dut,
)
errormsg = (
"Nexthop {} is Missing for "
"route {} in RIB of router {}\n".format(
additional_nexthops_in_required_nhs,
st_rt,
dut,
)
)
return errormsg
else:
nh_found = True
if aspath:
found_paths = rib_routes_json["routes"][st_rt][0][
"path"
]
if aspath == found_paths:
aspath_found = True
logger.info(
"Found AS path {} for route"
" {} in RIB of router "
"{}\n".format(aspath, st_rt, dut)
)
else:
errormsg = (
"AS Path {} is missing for route"
"for route {} in RIB of router {}\n".format(
aspath, st_rt, dut
)
)
return errormsg
else:
missing_routes.append(st_rt)
if nh_found:
logger.info(
"Found next_hop {} for all bgp"
" routes in RIB of"
" router {}\n".format(next_hop, router)
)
if len(missing_routes) > 0:
errormsg = (
"Missing route in RIB of router {}, "
"routes: {}\n".format(dut, missing_routes)
)
return errormsg
if found_routes:
logger.info(
"Verified routes in router {} BGP RIB, "
"found routes are: {} \n".format(dut, found_routes)
)
continue
if "bgp" not in input_dict[routerInput]:
continue
# Advertise networks
bgp_data_list = input_dict[routerInput]["bgp"]
if type(bgp_data_list) is not list:
bgp_data_list = [bgp_data_list]
for bgp_data in bgp_data_list:
vrf_id = bgp_data.setdefault("vrf", None)
if vrf_id:
cmd = "{} vrf {} {}".format(command, vrf_id, addr_type)
else:
cmd = "{} {}".format(command, addr_type)
cmd = "{} json".format(cmd)
rib_routes_json = run_frr_cmd(rnode, cmd, isjson=True)
# Verifying output dictionary rib_routes_json is not empty
if bool(rib_routes_json) == False:
errormsg = "No route found in rib of router {}..".format(router)
return errormsg
bgp_net_advertise = bgp_data["address_family"][addr_type]["unicast"]
advertise_network = bgp_net_advertise.setdefault(
"advertise_networks", []
)
for advertise_network_dict in advertise_network:
found_routes = []
missing_routes = []
found = False
network = advertise_network_dict["network"]
if "no_of_network" in advertise_network_dict:
no_of_network = advertise_network_dict["no_of_network"]
else:
no_of_network = 1
# Generating IPs for verification
ip_list = generate_ips(network, no_of_network)
for st_rt in ip_list:
st_rt = str(ipaddr.IPNetwork(unicode(st_rt)))
_addr_type = validate_ip_address(st_rt)
if _addr_type != addr_type:
continue
if st_rt in rib_routes_json["routes"]:
found = True
found_routes.append(st_rt)
else:
found = False
missing_routes.append(st_rt)
if len(missing_routes) > 0:
errormsg = (
"Missing route in BGP RIB of router {},"
" are: {}\n".format(dut, missing_routes)
)
return errormsg
if found_routes:
logger.info(
"Verified routes in router {} BGP RIB, found "
"routes are: {}\n".format(dut, found_routes)
)
logger.debug("Exiting lib API: verify_bgp_rib()")
return True