test: minor, whitespace and PEP-8 fixes

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2024-09-26 15:49:45 +02:00
parent 6e85ace4b0
commit 0ea84cf557
4 changed files with 249 additions and 229 deletions
+109 -105
View File
@@ -197,30 +197,30 @@ def config_target2(target, ring1, ring2, cross):
"enabled": True,
"ipv4": {
"address": [{
"ip": "10.0.0.2",
"prefix-length": 32
},{
"ip": "11.0.9.1",
"prefix-length": 24
},{
"ip": "11.0.10.1",
"prefix-length": 24
},{
"ip": "11.0.11.1",
"prefix-length": 24
},{
"ip": "11.0.12.1",
"prefix-length": 24
},{
"ip": "11.0.13.1",
"prefix-length": 24
},{
"ip": "11.0.14.1",
"prefix-length": 24
},{
"ip": "11.0.15.1",
"prefix-length": 24
}]
"ip": "10.0.0.2",
"prefix-length": 32
}, {
"ip": "11.0.9.1",
"prefix-length": 24
}, {
"ip": "11.0.10.1",
"prefix-length": 24
}, {
"ip": "11.0.11.1",
"prefix-length": 24
}, {
"ip": "11.0.12.1",
"prefix-length": 24
}, {
"ip": "11.0.13.1",
"prefix-length": 24
}, {
"ip": "11.0.14.1",
"prefix-length": 24
}, {
"ip": "11.0.15.1",
"prefix-length": 24
}]
}
}
]
@@ -243,7 +243,7 @@ def config_target2(target, ring1, ring2, cross):
"areas": {
"area": [{
"area-id": "0.0.0.0",
"interfaces":{
"interfaces": {
"interface": [{
"bfd": {
"enabled": True
@@ -251,32 +251,30 @@ def config_target2(target, ring1, ring2, cross):
"name": ring2,
"hello-interval": 1,
"enabled": True
},{
}, {
"name": "lo",
"enabled": True
}]
}
},
{
}, {
"area-id": "0.0.0.1",
"area-type": "nssa-area",
"summary": False,
"interfaces":{
"interfaces": {
"interface": [{
"bfd": {
"enabled": True
},
"name": ring1,
"hello-interval": 1,
},{
}, {
"name": "lo",
"enabled": True
}]
}
},
{
}, {
"area-id": "0.0.0.2",
"interfaces":{
"interfaces": {
"interface": [{
"bfd": {
"enabled": True
@@ -284,7 +282,7 @@ def config_target2(target, ring1, ring2, cross):
"name": cross,
"hello-interval": 1,
"cost": 2000
},{
}, {
"name": "lo",
"enabled": True
}]
@@ -297,6 +295,7 @@ def config_target2(target, ring1, ring2, cross):
}
})
def config_target3(target, ring2, cross, link):
target.put_config_dict("ietf-interfaces", {
"interfaces": {
@@ -366,7 +365,7 @@ def config_target3(target, ring2, cross, link):
"area-id": "0.0.0.1",
"area-type": "nssa-area",
"summary": False,
"interfaces":{
"interfaces": {
"interface": [{
"bfd": {
"enabled": True
@@ -375,18 +374,18 @@ def config_target3(target, ring2, cross, link):
"hello-interval": 1,
"enabled": True,
"cost": 2000
},{
}, {
"bfd": {
"enabled": True
},
"name": ring2,
"hello-interval": 1,
"enabled": True
},{
}, {
"name": link,
"enabled": True,
"passive": True
},{
}, {
"name": "lo",
"enabled": True
}]
@@ -399,6 +398,7 @@ def config_target3(target, ring2, cross, link):
}
})
def config_target4(target, ring1, cross, link):
target.put_config_dict("ietf-interfaces", {
"interfaces": {
@@ -429,8 +429,8 @@ def config_target4(target, ring1, cross, link):
"ipv4": {
"forwarding": True,
"address": [{
"ip": "192.168.4.1",
"prefix-length": 24
"ip": "192.168.4.1",
"prefix-length": 24
}]
}
},
@@ -439,8 +439,8 @@ def config_target4(target, ring1, cross, link):
"enabled": True,
"ipv4": {
"address": [{
"ip": "10.0.0.4",
"prefix-length": 32
"ip": "10.0.0.4",
"prefix-length": 32
}]
}
}
@@ -459,48 +459,48 @@ def config_target4(target, ring1, cross, link):
"control-plane-protocols": {
"control-plane-protocol": [
{
"type": "infix-routing:ospfv2",
"name": "default",
"ospf": {
"redistribute": {
"redistribute": [
{
"protocol": "connected"
}]
},
"areas": {
"area": [{
"area-id": "0.0.0.2",
"interfaces":{
"interface": [{
"bfd": {
"enabled": True
},
"name": ring1,
"hello-interval": 1,
"enabled": True
},{
"bfd": {
"enabled": True
},
"name": cross,
"hello-interval": 1,
"enabled": True,
"cost": 5000
},{
"name": "lo",
"enabled": True
"type": "infix-routing:ospfv2",
"name": "default",
"ospf": {
"redistribute": {
"redistribute": [{
"protocol": "connected"
}]
}
}]
},
"areas": {
"area": [{
"area-id": "0.0.0.2",
"interfaces": {
"interface": [{
"bfd": {
"enabled": True
},
"name": ring1,
"hello-interval": 1,
"enabled": True
}, {
"bfd": {
"enabled": True
},
"name": cross,
"hello-interval": 1,
"enabled": True,
"cost": 5000
}, {
"name": "lo",
"enabled": True
}]
}
}]
}
}
}
}
]
}
}
})
def disable_link(target, link):
target.put_config_dict("ietf-interfaces", {
"interfaces": {
@@ -512,6 +512,7 @@ def disable_link(target, link):
}
})
with infamy.Test() as test:
with test.step("Configure targets"):
env = infamy.Env()
@@ -520,7 +521,6 @@ with infamy.Test() as test:
R3 = env.attach("R3", "mgmt")
R4 = env.attach("R4", "mgmt")
_, R1ring1 = env.ltop.xlate("R1", "ring1")
_, R1ring2 = env.ltop.xlate("R1", "ring2")
_, R2ring1 = env.ltop.xlate("R2", "ring1")
@@ -528,17 +528,18 @@ with infamy.Test() as test:
_, R3ring2 = env.ltop.xlate("R3", "ring2")
_, R4ring1 = env.ltop.xlate("R4", "ring1")
_, R3data = env.ltop.xlate("R3", "data")
_, R4data = env.ltop.xlate("R4", "data")
_, R3data = env.ltop.xlate("R3", "data")
_, R4data = env.ltop.xlate("R4", "data")
_, R1cross = env.ltop.xlate("R1", "cross")
_, R2cross = env.ltop.xlate("R2", "cross")
_, R3cross = env.ltop.xlate("R3", "cross")
_, R4cross = env.ltop.xlate("R4", "cross")
_, R1cross = env.ltop.xlate("R1", "cross")
_, R2cross = env.ltop.xlate("R2", "cross")
_, R3cross = env.ltop.xlate("R3", "cross")
_, R4cross = env.ltop.xlate("R4", "cross")
parallel(config_target1(R1, R1ring1, R1ring2, R1cross),
config_target2(R2, R2ring1, R2ring2, R2cross),
config_target3(R3, R3ring2, R3cross, R3data),
config_target4(R4, R4ring1, R4cross, R4data))
with test.step("Wait for all neighbor to peer"):
print("Waiting for neighbors to peer")
until(lambda: route.ospf_get_neighbor(R1, "0.0.0.0", R1ring1, "10.0.0.2"), attempts=200)
@@ -564,18 +565,19 @@ with infamy.Test() as test:
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"))
with test.step("Verify on R3, there are no routes beyond 10.0.23.1, just a default route"): # Should be only default route out of the area.
parallel(until(lambda: route.ipv4_route_exist(R3, "0.0.0.0/0"), attempts=200),
until(lambda: route.ipv4_route_exist(R3, "10.0.12.0/30") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "10.0.12.0/30") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.8.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.9.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.10.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.11.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.12.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.13.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.14.0/24") == False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.15.0/24") == False, attempts=5))
with test.step("Verify on R3, there are no routes beyond 10.0.23.1, just a default route"):
# Should be only default route out of the area.
parallel(until(lambda: route.ipv4_route_exist(R3, "0.0.0.0/0"), attempts=200),
until(lambda: route.ipv4_route_exist(R3, "10.0.12.0/30") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "10.0.12.0/30") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.8.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.9.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.10.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.11.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.12.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.13.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.14.0/24") is False, attempts=5),
until(lambda: route.ipv4_route_exist(R3, "11.0.15.0/24") is False, attempts=5))
_, hport0 = env.ltop.xlate("PC", "data3")
with infamy.IsolatedMacVlan(hport0) as ns0:
@@ -589,21 +591,23 @@ with infamy.Test() as test:
ns0.addip("192.168.4.2")
ns0.addroute("0.0.0.0/0", "192.168.4.1")
with test.step("Verify that the route to 10.0.0.3 from PC:data4, go through 10.0.41.2"):
trace=ns0.traceroute("10.0.0.3")
assert(len(trace) == 3)
assert(trace[1][1] == "10.0.41.2")
assert(trace[2][1] == "10.0.0.3")
trace = ns0.traceroute("10.0.0.3")
assert len(trace) == 3
assert trace[1][1] == "10.0.41.2"
assert trace[2][1] == "10.0.0.3"
with test.step("Break link R1:ring2 --- R4:ring1"):
disable_link(R1, R1ring2) # Here we should test with link breakers, to test BFD recouppling, for now disable the link
# Here we should test with link breakers, to test BFD
# recouppling, for now disable the link
disable_link(R1, R1ring2)
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="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)
assert(trace[1][1] == "10.0.24.1")
assert(trace[2][1] == "10.0.0.3")
until(lambda: route.ipv4_route_exist(R4, "10.0.0.3/32", nexthop="10.0.41.2") is False, attempts=10)
trace = ns0.traceroute("10.0.0.3")
assert len(trace) == 3
assert trace[1][1] == "10.0.24.1"
assert trace[2][1] == "10.0.0.3"
ns0.must_reach("10.0.0.3")
test.succeed()
@@ -37,8 +37,8 @@ def config_target1(target, data, link):
"ipv4": {
"forwarding": True,
"address": [{
"ip": "10.0.0.1",
"prefix-length": 32
"ip": "10.0.0.1",
"prefix-length": 32
}]
}
},
@@ -47,8 +47,8 @@ def config_target1(target, data, link):
"enabled": True,
"ipv4": {
"address": [{
"ip": "192.168.100.1",
"prefix-length": 32
"ip": "192.168.100.1",
"prefix-length": 32
}]
}
}
@@ -73,12 +73,10 @@ def config_target1(target, data, link):
"hello-interval": 1,
"dead-interval": 3,
"interface-type": "point-to-point"
},
{
}, {
"name": data,
"passive": True
},
{
}, {
"name": "lo",
"passive": True
}]
@@ -91,6 +89,7 @@ def config_target1(target, data, link):
}
})
def config_target2(target, link):
target.put_config_dict("ietf-interfaces", {
"interfaces": {
@@ -112,8 +111,8 @@ def config_target2(target, link):
"forwarding": True,
"ipv4": {
"address": [{
"ip": "192.168.200.1",
"prefix-length": 32
"ip": "192.168.200.1",
"prefix-length": 32
}]
}
}
@@ -126,33 +125,33 @@ def config_target2(target, link):
"control-plane-protocols": {
"control-plane-protocol": [
{
"type": "infix-routing:ospfv2",
"name": "default",
"ospf": {
"areas": {
"area": [{
"area-id": "0.0.0.0",
"interfaces":{
"interface": [{
"name": link,
"hello-interval": 1,
"dead-interval": 3,
"interface-type": "point-to-point"
},
{
"name": "lo",
"passive": True
}]
}
}]
"type": "infix-routing:ospfv2",
"name": "default",
"ospf": {
"areas": {
"area": [{
"area-id": "0.0.0.0",
"interfaces": {
"interface": [{
"name": link,
"hello-interval": 1,
"dead-interval": 3,
"interface-type": "point-to-point"
}, {
"name": "lo",
"passive": True
}]
}
}]
}
}
}
}
]
}
}
})
with infamy.Test() as test:
with test.step("Configure targets"):
env = infamy.Env()
+77 -76
View File
@@ -21,72 +21,70 @@ Static routing
Test that it is possible to use static routes (IPv4 and IPv6)
works as expected
"""
import copy
import infamy
import time
import infamy.iface as iface
import infamy.route as route
from infamy.util import until, parallel
def config_target1(target, data, link):
target.put_config_dict("ietf-interfaces", {
"interfaces": {
"interface": [
{
"name": data,
"enabled": True,
"ipv4": {
"forwarding": True,
"address": [{
"ip": "192.168.10.1",
"prefix-length": 24
}]},
"ipv6": {
"forwarding": True,
"address": [{
"ip": "2001:db8:3c4d:10::1",
"prefix-length": 64
}]
}
},
{
"name": link,
"enabled": True,
"ipv4": {
"forwarding": True,
"address": [{
"interfaces": {
"interface": [
{
"name": data,
"enabled": True,
"ipv4": {
"forwarding": True,
"address": [{
"ip": "192.168.10.1",
"prefix-length": 24
}]},
"ipv6": {
"forwarding": True,
"address": [{
"ip": "2001:db8:3c4d:10::1",
"prefix-length": 64
}]
}
},
{
"name": link,
"enabled": True,
"ipv4": {
"forwarding": True,
"address": [{
"ip": "192.168.50.1",
"prefix-length": 24
}]
},
"ipv6": {
"forwarding": True,
"address": [{
"ip": "2001:db8:3c4d:50::1",
"prefix-length": 64
}]
}
}]
},
{
"name": "lo",
"enabled": True,
"ipv4": {
"address": [{
"ipv6": {
"forwarding": True,
"address": [{
"ip": "2001:db8:3c4d:50::1",
"prefix-length": 64
}]
}
},
{
"name": "lo",
"enabled": True,
"ipv4": {
"address": [{
"ip": "192.168.100.1",
"prefix-length": 32
}]
},
"ipv6": {
"address": [{
"ip": "2001:db8:3c4d:100::1",
"prefix-length": 128
}]
}
}]
},
"ipv6": {
"address": [{
"ip": "2001:db8:3c4d:100::1",
"prefix-length": 128
}]
}
}
]
}
})
]
}
})
target.put_config_dict("ietf-routing", {
"routing": {
"control-plane-protocols": {
@@ -104,10 +102,10 @@ def config_target1(target, data, link):
},
"ipv6": {
"route": [{
"destination-prefix": "2001:db8:3c4d:200::1/128",
"next-hop": {
"next-hop-address": "2001:db8:3c4d:50::2"
}
"destination-prefix": "2001:db8:3c4d:200::1/128",
"next-hop": {
"next-hop-address": "2001:db8:3c4d:50::2"
}
}]
}
}
@@ -115,6 +113,8 @@ def config_target1(target, data, link):
}
}
})
def config_target2(target, link):
target.put_config_dict("ietf-interfaces", {
"interfaces": {
@@ -143,8 +143,8 @@ def config_target2(target, link):
"forwarding": True,
"ipv4": {
"address": [{
"ip": "192.168.200.1",
"prefix-length": 32
"ip": "192.168.200.1",
"prefix-length": 32
}]
},
"ipv6": {
@@ -168,18 +168,18 @@ def config_target2(target, link):
"static-routes": {
"ipv4": {
"route": [{
"destination-prefix": "0.0.0.0/0",
"next-hop": {
"next-hop-address": "192.168.50.1"
}
}]
"destination-prefix": "0.0.0.0/0",
"next-hop": {
"next-hop-address": "192.168.50.1"
}
}]
},
"ipv6": {
"route": [{
"destination-prefix": "::/0",
"next-hop": {
"next-hop-address": "2001:db8:3c4d:50::1"
}
"destination-prefix": "::/0",
"next-hop": {
"next-hop-address": "2001:db8:3c4d:50::1"
}
}]
}
}
@@ -188,6 +188,7 @@ def config_target2(target, link):
}
})
with infamy.Test() as test:
with test.step("Configure targets"):
env = infamy.Env()
@@ -210,21 +211,21 @@ with infamy.Test() as test:
_, hport0 = env.ltop.xlate("PC", "data")
with infamy.IsolatedMacVlan(hport0) as ns0:
with test.step("Configure host addresses and routes"):
ns0.addip("2001:db8:3c4d:10::2", prefix_length=64, proto="ipv6")
ns0.addroute("2001:db8:3c4d:200::1/128", "2001:db8:3c4d:10::1", proto="ipv6")
ns0.addip("192.168.10.2")
ns0.addroute("192.168.200.1/32", "192.168.10.1")
ns0.addip("2001:db8:3c4d:10::2", prefix_length=64, proto="ipv6")
ns0.addroute("2001:db8:3c4d:200::1/128", "2001:db8:3c4d:10::1", proto="ipv6")
ns0.addip("192.168.10.2")
ns0.addroute("192.168.200.1/32", "192.168.10.1")
with test.step("Verify that dut2 is reachable on 192.168.200.1 from PC:data"):
ns0.must_reach("192.168.200.1")
ns0.must_reach("192.168.200.1")
with test.step("Verify that R2 is reachable on 2001:db8:3c4d:200::1 from PC:data"):
ns0.must_reach("2001:db8:3c4d:200::1")
ns0.must_reach("2001:db8:3c4d:200::1")
with test.step("Remove all static routes from R1"):
R1.delete_xpath("/ietf-routing:routing/control-plane-protocols")
parallel(until(lambda: route.ipv4_route_exist(R1, "192.168.200.1/32") == False),
until(lambda: route.ipv6_route_exist(R1, "2001:db8:3c4d:200::1/128") == False))
parallel(until(lambda: route.ipv4_route_exist(R1, "192.168.200.1/32") is False),
until(lambda: route.ipv6_route_exist(R1, "2001:db8:3c4d:200::1/128") is False))
with test.step("Verify R2 is no longer reachable on either IPv4 or IPv6 from PC:data"):
infamy.parallel(ns0.must_not_reach("192.168.200.1"),
+34 -18
View File
@@ -1,10 +1,15 @@
"""
IETF Routing helper methods
"""
def _get_routes(target, protocol):
xpath="/ietf-routing:routing/ribs"
xpath = "/ietf-routing:routing/ribs"
rib = target.get_data(xpath)["routing"]["ribs"]["rib"]
for r in rib:
if r["name"] != protocol:
continue
return r.get("routes", {}).get("route",{})
return r.get("routes", {}).get("route", {})
return {}
@@ -47,57 +52,68 @@ def _exist_route(target, prefix, nexthop, ip, source_protocol):
return True
return False
def ipv4_route_exist(target, prefix, nexthop=None,source_protocol=None):
return _exist_route(target, prefix, nexthop, "ipv4",source_protocol)
def ipv6_route_exist(target, prefix, nexthop=None,source_protocol=None):
return _exist_route(target, prefix, nexthop, "ipv6",source_protocol)
def ipv4_route_exist(target, prefix, nexthop=None, source_protocol=None):
return _exist_route(target, prefix, nexthop, "ipv4", source_protocol)
def ipv6_route_exist(target, prefix, nexthop=None, source_protocol=None):
return _exist_route(target, prefix, nexthop, "ipv6", source_protocol)
def _get_ospf_status(target):
xpath="/ietf-routing:routing/control-plane-protocols"
rib = target.get_data(xpath)["routing"]["control-plane-protocols"].get("control-plane-protocol", {})
xpath = "/ietf-routing:routing/control-plane-protocols"
protos = target.get_data(xpath)["routing"]["control-plane-protocols"]
rib = protos.get("control-plane-protocol", {})
for p in rib:
if p["type"] == "infix-routing:ospfv2":
return p.get("ospf") or p.get("ietf-ospf:ospf")
return []
def _get_ospf_status_area(target, area_id):
ospf=_get_ospf_status(target)
ospf = _get_ospf_status(target)
for area in ospf.get("areas", {}).get("area", {}):
if area["area-id"] == area_id:
return area
return {}
def _get_ospf_status_area_interface(target, area_id, ifname):
area=_get_ospf_status_area(target,area_id)
def _get_ospf_status_area_interface(target, area_id, ifname):
area = _get_ospf_status_area(target, area_id)
for interface in area.get("interfaces", {}).get("interface", {}):
if interface.get("name") == ifname:
return interface
return {}
def ospf_get_neighbor(target, area_id, ifname, neighbour_id, full=True):
ospf_interface=_get_ospf_status_area_interface(target,area_id, ifname)
ospf_interface = _get_ospf_status_area_interface(target, area_id, ifname)
for neighbor in ospf_interface.get("neighbors", {}).get("neighbor", {}):
if neighbor.get("neighbor-router-id") == neighbour_id:
if full == False:
if full is False:
return True
if(neighbor.get("state") == "full"):
if neighbor.get("state") == "full":
return True
return False
def ospf_get_interface_type(target, area_id, ifname):
ospf_interface=_get_ospf_status_area_interface(target,area_id,ifname)
ospf_interface = _get_ospf_status_area_interface(target, area_id, ifname)
return ospf_interface.get("interface-type", None)
def ospf_get_interface_passive(target, area_id, ifname):
ospf_interface=_get_ospf_status_area_interface(target,area_id,ifname)
ospf_interface = _get_ospf_status_area_interface(target, area_id, ifname)
return ospf_interface.get("passive", False)
def ospf_is_area_nssa(target, area_id):
area=_get_ospf_status_area(target, area_id)
def ospf_is_area_nssa(target, area_id):
area = _get_ospf_status_area(target, area_id)
if area.get("area-type", "") == "ietf-ospf:nssa-area":
return True