test: update rib source protocol, infix-routing -> ietf-ospf

Latest YANG model change reverts back to using the IETF routing-protocol
type.  Update tests to check for infix-routing:ospf -> ietf-ospf:ospfv2.

Also, refactor and extend infamy/route.py:_exist_route() slightly for
readability and PEP-8, as well as to use an optional model prefix also
for next-hop-list addresses.

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2024-09-26 15:49:45 +02:00
parent 76e54420a2
commit 6e85ace4b0
4 changed files with 40 additions and 29 deletions
+2 -2
View File
@@ -171,8 +171,8 @@ with infamy.Test() as test:
config_target2(R2, R2link))
with test.step("Wait for OSPF routes"):
print("Waiting for OSPF routes..")
until(lambda: route.ipv4_route_exist(R1, "192.168.200.1/32", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "192.168.100.1/32", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "192.168.200.1/32", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "192.168.100.1/32", source_protocol="ietf-ospf:ospfv2"), attempts=200)
with test.step("Test connectivity from PC:data to 192.168.200.1"):
_, hport0 = env.ltop.xlate("PC", "data")
+12 -12
View File
@@ -549,17 +549,17 @@ with infamy.Test() as test:
with test.step("Wait for routes from OSPF on all routers"):
print("Waiting for routes from OSPF")
until(lambda: route.ipv4_route_exist(R1, "10.0.0.2/32", nexthop="10.0.12.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.0.3/32", nexthop="10.0.13.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.0.4/32", nexthop="10.0.41.1", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "192.168.4.0/24", nexthop="10.0.41.1", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.24.0/30", nexthop="10.0.41.1", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.0.1/32", nexthop="10.0.23.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.0.3/32", nexthop="10.0.23.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.0.4/32", nexthop="10.0.24.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R3, "0.0.0.0/0", nexthop="10.0.23.1", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R4, "10.0.0.3/32", nexthop="10.0.41.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.13.0/30", nexthop="10.0.23.2", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.0.2/32", nexthop="10.0.12.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.0.3/32", nexthop="10.0.13.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.0.4/32", nexthop="10.0.41.1", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "192.168.4.0/24", nexthop="10.0.41.1", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "10.0.24.0/30", nexthop="10.0.41.1", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.0.1/32", nexthop="10.0.23.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.0.3/32", nexthop="10.0.23.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.0.4/32", nexthop="10.0.24.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R3, "0.0.0.0/0", nexthop="10.0.23.1", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R4, "10.0.0.3/32", nexthop="10.0.41.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "10.0.13.0/30", nexthop="10.0.23.2", source_protocol="ietf-ospf:ospfv2"), attempts=200)
with test.step("Verify Area 0.0.0.1 on R3 is NSSA area"):
assert(route.ospf_is_area_nssa(R3, "0.0.0.1"))
@@ -598,7 +598,7 @@ with infamy.Test() as test:
disable_link(R1, R1ring2) # Here we should test with link breakers, to test BFD recouppling, for now disable the link
with test.step("Verify that the route to 10.0.0.3 from PC:data4, go through 10.0.24.1"):
until(lambda: route.ipv4_route_exist(R4, "10.0.0.3/32", nexthop="10.0.24.1", source_protocol = "infix-routing:ospf"), attempts=100)
until(lambda: route.ipv4_route_exist(R4, "10.0.0.3/32", nexthop="10.0.24.1", source_protocol="ietf-ospf:ospfv2"), attempts=100)
until(lambda: route.ipv4_route_exist(R4, "10.0.0.3/32", nexthop="10.0.41.2") == False, attempts = 10)
trace=ns0.traceroute("10.0.0.3")
assert(len(trace) == 3)
@@ -167,9 +167,9 @@ with infamy.Test() as test:
lambda: config_target2(R2, R2link))
with test.step("Wait for OSPF routes"):
print("Waiting for OSPF routes..")
until(lambda: route.ipv4_route_exist(R1, "192.168.200.1/32", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "192.168.100.1/32", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "192.168.10.0/24", source_protocol = "infix-routing:ospf"), attempts=200)
until(lambda: route.ipv4_route_exist(R1, "192.168.200.1/32", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "192.168.100.1/32", source_protocol="ietf-ospf:ospfv2"), attempts=200)
until(lambda: route.ipv4_route_exist(R2, "192.168.10.0/24", source_protocol="ietf-ospf:ospfv2"), attempts=200)
with test.step("Check interface type"):
assert(route.ospf_get_interface_type(R1, "0.0.0.0", R1link) == "point-to-point")
+23 -12
View File
@@ -7,34 +7,45 @@ def _get_routes(target, protocol):
return r.get("routes", {}).get("route",{})
return {}
def _exist_route(target, prefix, nexthop, version, source_protocol):
routes = _get_routes(target, version)
route_found = False
def _exist_route(target, prefix, nexthop, ip, source_protocol):
routes = _get_routes(target, ip)
for r in routes:
# netconf presents destination-prefix, restconf prefix with model
p=r.get("destination-prefix") or (version == "ipv4" and r.get("ietf-ipv4-unicast-routing:destination-prefix")) or (version == "ipv6" and r.get("ietf-ipv6-unicast-routing:destination-prefix"))
p = r.get("destination-prefix") or \
r.get(f"ietf-{ip}-unicast-routing:destination-prefix")
if p != prefix:
continue
if source_protocol and r.get("source-protocol") != source_protocol:
return False
route_found = True
nh = r["next-hop"]
next_hop_list = nh.get("next-hop-list")
if nexthop:
nh = r["next-hop"]
if not nh:
return False
next_hop_list = nh.get("next-hop-list")
if next_hop_list:
for nhl in next_hop_list["next-hop"]:
address = nhl.get("address")
# netconf presents address, restconf prefix with
# model ietf-ipv4-unicast-routing:address
address = nhl.get("address") or \
nhl.get(f"ietf-{ip}-unicast-routing:address")
if address == nexthop:
return True
return False
else:
# netconf presents next-hop-address, restconf prefix with model ietf-ipv4-unicast-routing:next-hop-address
nh_addr=nh.get("next-hop-address", None) or (version == "ipv4" and nh.get("ietf-ipv4-unicast-routing:next-hop-address", None)) or (version == "ipv6" and nh.get("ietf-ipv6-unicast-routing:next-hop-address", None))
# netconf presents next-hop-address, restconf prefix
# with model ietf-ipv4-unicast-routing:next-hop-address
nh_addr = nh.get("next-hop-address") or \
nh.get(f"ietf-{ip}-unicast-routing:next-hop-address")
if nh_addr == nexthop:
return True
return False
return route_found
return False
else:
return True
return False
def ipv4_route_exist(target, prefix, nexthop=None,source_protocol=None):
return _exist_route(target, prefix, nexthop, "ipv4",source_protocol)