1import ipaddress 2import socket 3 4import pytest 5from atf_python.sys.net.rtsock import RtConst 6from atf_python.sys.net.rtsock import Rtsock 7from atf_python.sys.net.rtsock import RtsockRtMessage 8from atf_python.sys.net.rtsock import SaHelper 9from atf_python.sys.net.tools import ToolsHelper 10from atf_python.sys.net.vnet import SingleVnetTestTemplate 11from atf_python.sys.net.vnet import VnetTestTemplate 12 13 14class TestIfOps(VnetTestTemplate): 15 TOPOLOGY = { 16 "vnet1": {"ifaces": ["if1", "if2"]}, 17 "if1": {"prefixes4": [], "prefixes6": []}, 18 "if2": {"prefixes4": [], "prefixes6": []}, 19 } 20 21 @pytest.mark.parametrize("family", ["inet", "inet6"]) 22 @pytest.mark.require_user("root") 23 def test_change_prefix_route(self, family): 24 """Tests that prefix route changes to the new one upon addr deletion""" 25 vnet = self.vnet_map["vnet1"] 26 first_iface = vnet.iface_alias_map["if1"] 27 second_iface = vnet.iface_alias_map["if2"] 28 if family == "inet": 29 first_addr = ipaddress.ip_interface("192.0.2.1/24") 30 second_addr = ipaddress.ip_interface("192.0.2.2/24") 31 else: 32 first_addr = ipaddress.ip_interface("2001:db8::1/64") 33 second_addr = ipaddress.ip_interface("2001:db8::2/64") 34 35 first_iface.setup_addr(str(first_addr)) 36 second_iface.setup_addr(str(second_addr)) 37 38 # At this time prefix should be pointing to the first interface 39 routes = ToolsHelper.get_routes(family) 40 px = [r for r in routes if r["destination"] == str(first_addr.network)][0] 41 assert px["interface-name"] == first_iface.name 42 43 # Now delete address from the first interface and verify switchover 44 first_iface.delete_addr(first_addr.ip) 45 46 routes = ToolsHelper.get_routes(family) 47 px = [r for r in routes if r["destination"] == str(first_addr.network)][0] 48 assert px["interface-name"] == second_iface.name 49 50 @pytest.mark.parametrize( 51 "family", 52 [ 53 "inet", 54 pytest.param("inet6", marks=pytest.mark.xfail(reason="currently fails")), 55 ], 56 ) 57 @pytest.mark.require_user("root") 58 def test_change_prefix_route_same_iface(self, family): 59 """Tests that prefix route changes to the new ifa upon addr deletion""" 60 vnet = self.vnet_map["vnet1"] 61 first_iface = vnet.iface_alias_map["if1"] 62 63 if family == "inet": 64 first_addr = ipaddress.ip_interface("192.0.2.1/24") 65 second_addr = ipaddress.ip_interface("192.0.2.2/24") 66 else: 67 first_addr = ipaddress.ip_interface("2001:db8::1/64") 68 second_addr = ipaddress.ip_interface("2001:db8::2/64") 69 70 first_iface.setup_addr(str(first_addr)) 71 first_iface.setup_addr(str(second_addr)) 72 73 # At this time prefix should be pointing to the first interface 74 routes = ToolsHelper.get_routes(family) 75 px = [r for r in routes if r["destination"] == str(first_addr.network)][0] 76 assert px["interface-name"] == first_iface.name 77 78 # Now delete address from the first interface and verify switchover 79 first_iface.delete_addr(str(first_addr.ip)) 80 81 routes = ToolsHelper.get_routes(family) 82 px = [r for r in routes if r["destination"] == str(first_addr.network)][0] 83 nhop_kidx = px["nhop"] 84 assert px["interface-name"] == first_iface.name 85 nhops = ToolsHelper.get_nhops(family) 86 nh = [nh for nh in nhops if nh["index"] == nhop_kidx][0] 87 assert nh["ifa"] == str(second_addr.ip) 88 89 90class TestRouteCornerCase1(SingleVnetTestTemplate): 91 @pytest.mark.parametrize("family", ["inet", "inet6"]) 92 @pytest.mark.require_user("root") 93 def test_add_direct_route_p2p_wo_ifa(self, family): 94 95 tun_ifname = ToolsHelper.get_output("/sbin/ifconfig tun create").rstrip() 96 tun_ifindex = socket.if_nametoindex(tun_ifname) 97 assert tun_ifindex > 0 98 rtsock = Rtsock() 99 100 if family == "inet": 101 prefix = "172.16.0.0/12" 102 else: 103 prefix = "2a02:6b8::/64" 104 IFT_ETHER = 0x06 105 gw_link = SaHelper.link_sa(ifindex=tun_ifindex, iftype=IFT_ETHER) 106 107 msg = rtsock.new_rtm_add(prefix, gw_link) 108 msg.add_link_attr(RtConst.RTA_IFP, tun_ifindex) 109 rtsock.write_message(msg) 110 111 data = rtsock.read_data(msg.rtm_seq) 112 msg_in = RtsockRtMessage.from_bytes(data) 113 msg_in.print_in_message() 114 115 desired_sa = { 116 RtConst.RTA_DST: msg.get_sa(RtConst.RTA_DST), 117 RtConst.RTA_NETMASK: msg.get_sa(RtConst.RTA_NETMASK), 118 } 119 120 msg_in.verify(msg.rtm_type, desired_sa) 121