Skip to content

Commit

Permalink
[SAI-PTF]Add tunnel encap test and more route test (#1644)
Browse files Browse the repository at this point in the history 
* Add tunnel ecap test  and default route test

Signed-off-by: zhoudongxu <Dongxu.Zhou@microsoft.com>

* refine test description

Signed-off-by: zhoudongxu <Dongxu.Zhou@microsoft.com>

* update test step description

Signed-off-by: zhoudongxu <Dongxu.Zhou@microsoft.com>

* add flush()

Signed-off-by: zhoudongxu <Dongxu.Zhou@microsoft.com>

Signed-off-by: zhoudongxu <Dongxu.Zhou@microsoft.com>
  • Loading branch information
@zhoudongxu
zhoudongxu authored Nov 11, 2022
1 parent 3f59a07 commit 63ed97e
Show file tree
Hide file tree
Showing 2 changed files with 269 additions and 3 deletions.
152 changes: 151 additions & 1 deletion test/sai_test/sai_route_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -983,7 +983,7 @@ def runTest(self):
def tearDown(self):
super().tearDown()

class RemoveRouteTest(T0TestBase):
class RemoveRouteV4Test(T0TestBase):
"""
Verify remove route entry
"""
Expand Down Expand Up @@ -1022,3 +1022,153 @@ def runTest(self):

def tearDown(self):
super().tearDown()

class DefaultRouteV6Test(T0TestBase):
"""
Verify Default Route
"""

def setUp(self):
"""
Test the basic setup process.
"""
super().setUp()

def runTest(self):
"""
Make sure default route and route interface are created as config spec.
Send packet with a DIP which is not exist in the config spec for any route or host neighbor, IPX
Verify no packetes received.
"""
dst_ipv6 = "fc80::f68c:38fe:fe16:bc74"
print("VerifyDefaultRouteTest")

try:
pkt_v6 = simple_udpv6_packet(eth_dst=ROUTER_MAC,
ipv6_dst=dst_ipv6,
ipv6_hlim=64)
send_packet(self, self.dut.port_obj_list[5].dev_port_index, pkt_v6)
verify_no_other_packets(self)
finally:
pass

def tearDown(self):
super().tearDown()

class DefaultRouteV4Test(T0TestBase):
"""
Verify Default Route
"""

def setUp(self):
"""
Test the basic setup process.
"""
super().setUp()

def runTest(self):
"""
Make sure default route and route interface are created as config spec.
Send packet with a DIP which is not exist in the config spec for any route or host neighbor, IPX
Verify no packetes received.
"""
dst_ip = "222.222.222.2"
print("VerifyDefaultRouteTest")
try:
pkt = simple_tcp_packet(eth_dst=ROUTER_MAC,
ip_dst=dst_ip,
ip_id=105,
ip_ttl=64)
send_packet(self, self.dut.port_obj_list[5].dev_port_index, pkt)
verify_no_other_packets(self)
finally:
pass

def tearDown(self):
super().tearDown()

class RouteSameSipDipv4Test(T0TestBase):
"""
Verify route packet with same dip and sip also can be forwarded
"""

def setUp(self):
"""
Test the basic setup process.
"""
super().setUp()

def runTest(self):
"""
1. Make sure common config for route dest IP within 192.168.12.0/24 through RIF(Nhop is Rif) to LAG2 created
2. Send packets for DIP:192.168.12.1~2 SIP 192.168.12.1~2 DMAC: SWITCH_MAC on port5
3. Verify packet received with SMAC: SWITCH_MAC SIP: 192.168.12.1~2 DIP:192.168.12.1~2 on one of LAG2 member
"""
print("RouteSameSipDipv4Test")

self.recv_dev_port_idxs = self.get_dev_port_indexes(self.servers[12][1].l3_lag_obj.member_port_indexs)
try:
for i in range(1, 3):
pkt = simple_tcp_packet(eth_dst=ROUTER_MAC,
ip_dst=self.servers[12][i].ipv4,
ip_src=self.servers[12][i].ipv4,
ip_id=105,
ip_ttl=64)
exp_pkt = simple_tcp_packet(eth_src=ROUTER_MAC,
eth_dst=self.servers[12][i].l3_lag_obj.neighbor_mac,
ip_dst=self.servers[12][i].ipv4,
ip_src=self.servers[12][i].ipv4,
ip_id=105,
ip_ttl=63)
self.dataplane.flush()
send_packet(self, self.dut.port_obj_list[5].dev_port_index, pkt)
verify_packet_any_port(self, exp_pkt, self.recv_dev_port_idxs)
print("received packet with dst_ip:{} on one of lag2 member".format(self.servers[12][i].ipv4))
finally:
pass

def tearDown(self):
super().tearDown()


class RouteSameSipDipv6Test(T0TestBase):
"""
Verify route packet with same dip and sip also can be forwarded
"""

def setUp(self):
"""
Test the basic setup process.
"""
super().setUp()

def runTest(self):
"""
1. Make sure common config for route dest IP within fc02::12:0/112 through RIF(Nhop is Rif) to LAG2 created
2. Send packets for DIP:fc02::12:1~2 SIP:fc02::12:1~2 DMAC:SWITCH_MAC on port5
3. Verify packet received with SMAC:SWITCH_MAC SIP:fc02::12:1~2 DIP:fc02::12:1~2 on one of LAG2 member
"""
print("RouteSameSipDipv6Test")

self.recv_dev_port_idxs = self.get_dev_port_indexes(self.servers[12][1].l3_lag_obj.member_port_indexs)
try:
for i in range(1, 3):
pkt_v6 = simple_udpv6_packet(eth_dst=ROUTER_MAC,
ipv6_src=self.servers[12][i].ipv6,
ipv6_dst=self.servers[12][i].ipv6,
ipv6_hlim=64)
exp_pkt_v6 = simple_udpv6_packet(eth_src=ROUTER_MAC,
eth_dst=self.servers[12][i].l3_lag_obj.neighbor_mac,
ipv6_dst=self.servers[12][i].ipv6,
ipv6_src=self.servers[12][i].ipv6,
ipv6_hlim=63)
self.dataplane.flush()
send_packet(self, self.dut.port_obj_list[5].dev_port_index, pkt_v6)
verify_packet_any_port(self, exp_pkt_v6, self.recv_dev_port_idxs)
print("received packet with dst_ip:{} on one of lag2 member".format(self.servers[12][i].ipv6))
finally:
pass

def tearDown(self):
super().tearDown()

120 changes: 118 additions & 2 deletions test/sai_test/sai_tunnel_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
from sai_test_base import T0TestBase
from sai_utils import *


from ptf.mask import Mask
class IpInIpTnnnelBase(T0TestBase):
'''
This class contains base setup for IP in IP tunnel tests
Expand All @@ -19,6 +18,7 @@ def setUp(self):
self.lpb_ip = LOOPBACK_IPV4
self.tun_lpb_mask = "/32"

self.recv_dev_port_idxs = self.get_dev_port_indexes(self.servers[11][1].l3_lag_obj.member_port_indexs)
self.vm_ip = "100.100.1.1"
self.vm_ipv6 = "2001:0db8::100:1"
self.customer_ip = self.servers[1][1].ipv4
Expand Down Expand Up @@ -180,3 +180,119 @@ def runTest(self):
finally:
pass

class IPInIPTunnelEncapv4Inv4Test(IpInIpTnnnelBase):
"""
We will send normal packet from port1 and verify that packet goes into tunnel, getting a ininip packet, recievinfg a encap packet on lag1.
"""

def setUp(self):
"""
Test the basic setup process.
"""
IpInIpTnnnelBase.setUp(self)

def ipv4inipv4encap(self):
"""
Generate ingress normal packet as decribed by Testing Data Packet
Send normal packet from port1.
expected encap packet as decribed by Testing Data Packet.
Recieve encap packet from lag1, compare it with expected encap packet.
"""

pkt = simple_udp_packet(eth_dst=ROUTER_MAC,
eth_src=self.customer_mac,
ip_dst=self.vm_ip,
ip_src=self.customer_ip,
ip_id=108,
ip_ttl=64,
)
inner_pkt = simple_udp_packet(eth_dst=self.inner_dmac,
eth_src=ROUTER_MAC,
ip_dst=self.vm_ip,
ip_src=self.customer_ip,
ip_id=108,
ip_ttl=63)


ipip_pkt = simple_ipv4ip_packet(eth_dst=self.unbor_mac,
eth_src=ROUTER_MAC,
ip_id=0,
ip_src=self.lpb_ip,
ip_dst=self.tun_ip,
ip_ttl= 63,
inner_frame=inner_pkt['IP'])

print("Verifying IP4inIP4 encapsulation")
m = Mask(ipip_pkt)
m.set_do_not_care_scapy(ptf.packet.IP, "len")
m.set_do_not_care_scapy(ptf.packet.IP, "chksum")
m.set_do_not_care_scapy(ptf.packet.IP, "flags")

self.dataplane.flush()
send_packet(self, self.oport_dev, pkt)
verify_packet_any_port(self, m, self.recv_dev_port_idxs)

print("\tOK")


def runTest(self):
try:
self.ipv4inipv4encap()
finally:
pass

class IPInIPTunnelEncapv6Inv4Test(IpInIpTnnnelBase):
"""
We will send normal packet from port1 and verify that packet goes into tunnel, getting a ininip packet, recievinfg a encap packet on lag1.
"""

def setUp(self):
"""
Test the basic setup process.
"""
IpInIpTnnnelBase.setUp(self)

def ipv6inipv4encap(self):
"""
Generate ingress normal packet as decribed by Testing Data Packet
Send normal packet from port1.
expected encap packet as decribed by Testing Data Packet
Recieve encap packet from lag1, compare it with expected encap packet.
"""

pkt = simple_udpv6_packet(eth_dst=ROUTER_MAC,
eth_src=self.customer_mac,
ipv6_dst=self.vm_ipv6,
ipv6_src=self.customer_ipv6,
ipv6_hlim=64)

inner_pkt = simple_udpv6_packet(eth_dst=self.inner_dmac,
eth_src=ROUTER_MAC,
ipv6_dst=self.vm_ipv6,
ipv6_src=self.customer_ipv6,
ipv6_hlim=63)

ipip_pkt = simple_ipv4ip_packet(eth_dst=self.unbor_mac,
eth_src=ROUTER_MAC,
ip_id=0,
ip_src=self.lpb_ip,
ip_dst=self.tun_ip,
inner_frame=inner_pkt['IPv6'])

print("Verifying IP6inIP4 encapsulation")
m = Mask(ipip_pkt)
m.set_do_not_care_scapy(ptf.packet.IP, "len")
m.set_do_not_care_scapy(ptf.packet.IP, "chksum")
m.set_do_not_care_scapy(ptf.packet.IP, "plen")
m.set_do_not_care_scapy(ptf.packet.IP, "flags")
self.dataplane.flush()
send_packet(self, self.oport_dev, pkt)
verify_packet_any_port(self, m, self.recv_dev_port_idxs)
print("\tOK")


def runTest(self):
try:
self.ipv6inipv4encap()
finally:
pass

0 comments on commit 63ed97e

Please sign in to comment.