test/test_l3xc.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from framework import VppTestCase
   6 from asfframework import VppTestRunner
   7 from vpp_ip_route import VppRoutePath
   8
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether
  11 from scapy.layers.inet import IP, UDP
  12
  13 from vpp_object import VppObject
  14
  15 NUM_PKTS = 67
  16
  17
  18 def find_l3xc(test, sw_if_index, dump_sw_if_index=None):
  19     if not dump_sw_if_index:
  20         dump_sw_if_index = sw_if_index
  21     xcs = test.vapi.l3xc_dump(dump_sw_if_index)
  22     for xc in xcs:
  23         if sw_if_index == xc.l3xc.sw_if_index:
  24             return True
  25     return False
  26
  27
  28 class VppL3xc(VppObject):
  29     def __init__(self, test, intf, paths, is_ip6=False):
  30         self._test = test
  31         self.intf = intf
  32         self.is_ip6 = is_ip6
  33         self.paths = paths
  34         self.encoded_paths = []
  35         for path in self.paths:
  36             self.encoded_paths.append(path.encode())
  37
  38     def add_vpp_config(self):
  39         self._test.vapi.l3xc_update(
  40             l3xc={
  41                 "is_ip6": self.is_ip6,
  42                 "sw_if_index": self.intf.sw_if_index,
  43                 "n_paths": len(self.paths),
  44                 "paths": self.encoded_paths,
  45             }
  46         )
  47         self._test.registry.register(self, self._test.logger)
  48
  49     def remove_vpp_config(self):
  50         self._test.vapi.l3xc_del(is_ip6=self.is_ip6, sw_if_index=self.intf.sw_if_index)
  51
  52     def query_vpp_config(self):
  53         return find_l3xc(self._test, self.intf.sw_if_index)
  54
  55     def object_id(self):
  56         return "l3xc-%d" % self.intf.sw_if_index
  57
  58
  59 class TestL3xc(VppTestCase):
  60     """L3XC Test Case"""
  61
  62     @classmethod
  63     def setUpClass(cls):
  64         super(TestL3xc, cls).setUpClass()
  65
  66     @classmethod
  67     def tearDownClass(cls):
  68         super(TestL3xc, cls).tearDownClass()
  69
  70     def setUp(self):
  71         super(TestL3xc, self).setUp()
  72
  73         self.create_pg_interfaces(range(6))
  74
  75         for i in self.pg_interfaces:
  76             i.admin_up()
  77             i.config_ip4()
  78             i.resolve_arp()
  79             i.config_ip6()
  80             i.resolve_ndp()
  81
  82     def tearDown(self):
  83         for i in self.pg_interfaces:
  84             i.unconfig_ip4()
  85             i.unconfig_ip6()
  86             i.admin_down()
  87         super(TestL3xc, self).tearDown()
  88
  89     def test_l3xc4(self):
  90         """IPv4 X-Connect"""
  91
  92         #
  93         # x-connect pg0 to pg1 and pg2 to pg3->5
  94         #
  95         l3xc_1 = VppL3xc(
  96             self, self.pg0, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)]
  97         )
  98         l3xc_1.add_vpp_config()
  99         l3xc_2 = VppL3xc(
 100             self,
 101             self.pg2,
 102             [
 103                 VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index),
 104                 VppRoutePath(self.pg4.remote_ip4, self.pg4.sw_if_index),
 105                 VppRoutePath(self.pg5.remote_ip4, self.pg5.sw_if_index),
 106             ],
 107         )
 108         l3xc_2.add_vpp_config()
 109
 110         self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xFFFFFFFF))
 111
 112         self.logger.info(self.vapi.cli("sh l3xc"))
 113
 114         #
 115         # fire in packets. If it's forwarded then the L3XC was successful,
 116         # since default routing will drop it
 117         #
 118         p_1 = (
 119             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 120             / IP(src="1.1.1.1", dst="1.1.1.2")
 121             / UDP(sport=1234, dport=1234)
 122             / Raw(b"\xa5" * 100)
 123         )
 124         # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)
 125
 126         p_2 = []
 127         for ii in range(NUM_PKTS):
 128             p_2.append(
 129                 Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 130                 / IP(src="1.1.1.1", dst="1.1.1.2")
 131                 / UDP(sport=1000 + ii, dport=1234)
 132                 / Raw(b"\xa5" * 100)
 133             )
 134         self.send_and_expect_load_balancing(
 135             self.pg2, p_2, [self.pg3, self.pg4, self.pg5]
 136         )
 137
 138         l3xc_2.remove_vpp_config()
 139         self.send_and_assert_no_replies(self.pg2, p_2)
 140
 141
 142 if __name__ == "__main__":
 143     unittest.main(testRunner=VppTestRunner)