test/test_pipe.py

   1 #!/usr/bin/env python3
   2 import unittest
   3 from ipaddress import IPv4Network
   4
   5 from scapy.packet import Raw
   6 from scapy.layers.l2 import Ether
   7 from scapy.layers.inet import IP, UDP
   8
   9 from framework import VppTestCase
  10 from asfframework import VppTestRunner
  11 from vpp_interface import VppInterface
  12 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
  13 from vpp_acl import AclRule, VppAcl, VppAclInterface
  14
  15 NUM_PKTS = 67
  16
  17
  18 class VppPipe(VppInterface):
  19     """
  20     VPP Pipe
  21     """
  22
  23     @property
  24     def east(self):
  25         return self.result.pipe_sw_if_index[1]
  26
  27     @property
  28     def west(self):
  29         return self.result.pipe_sw_if_index[0]
  30
  31     def __init__(self, test, instance=0xFFFFFFFF):
  32         super(VppPipe, self).__init__(test)
  33         self._test = test
  34         self.instance = instance
  35
  36     def add_vpp_config(self):
  37         self.result = self._test.vapi.pipe_create(
  38             0 if self.instance == 0xFFFFFFFF else 1, self.instance
  39         )
  40         self.set_sw_if_index(self.result.sw_if_index)
  41
  42     def remove_vpp_config(self):
  43         self._test.vapi.pipe_delete(self.result.sw_if_index)
  44
  45     def object_id(self):
  46         return "pipe-%d" % (self._sw_if_index)
  47
  48     def query_vpp_config(self):
  49         pipes = self._test.vapi.pipe_dump()
  50         for p in pipes:
  51             if p.sw_if_index == self.result.sw_if_index:
  52                 return True
  53         return False
  54
  55     def set_unnumbered(self, ip_sw_if_index, is_east, is_add=True):
  56         if is_east:
  57             res = self._test.vapi.sw_interface_set_unnumbered(
  58                 ip_sw_if_index, self.east, is_add
  59             )
  60         else:
  61             res = self._test.vapi.sw_interface_set_unnumbered(
  62                 ip_sw_if_index, self.west, is_add
  63             )
  64
  65
  66 class TestPipe(VppTestCase):
  67     """Pipes"""
  68
  69     @classmethod
  70     def setUpClass(cls):
  71         super(TestPipe, cls).setUpClass()
  72
  73     @classmethod
  74     def tearDownClass(cls):
  75         super(TestPipe, cls).tearDownClass()
  76
  77     def setUp(self):
  78         super(TestPipe, self).setUp()
  79
  80         self.create_pg_interfaces(range(4))
  81
  82         for i in self.pg_interfaces:
  83             i.admin_up()
  84
  85     def tearDown(self):
  86         for i in self.pg_interfaces:
  87             i.admin_down()
  88
  89         super(TestPipe, self).tearDown()
  90
  91     def test_pipe(self):
  92         """Pipes"""
  93
  94         pipes = [VppPipe(self), VppPipe(self, 10)]
  95
  96         for p in pipes:
  97             p.add_vpp_config()
  98             p.admin_up()
  99
 100         #
 101         # L2 cross-connect pipe0 east with pg0 and west with pg1
 102         #
 103         self.vapi.sw_interface_set_l2_xconnect(
 104             self.pg0.sw_if_index, pipes[0].east, enable=1
 105         )
 106         self.vapi.sw_interface_set_l2_xconnect(
 107             pipes[0].east, self.pg0.sw_if_index, enable=1
 108         )
 109         self.vapi.sw_interface_set_l2_xconnect(
 110             self.pg1.sw_if_index, pipes[0].west, enable=1
 111         )
 112         self.vapi.sw_interface_set_l2_xconnect(
 113             pipes[0].west, self.pg1.sw_if_index, enable=1
 114         )
 115
 116         # test bi-directional L2 flow pg0<->pg1
 117         p = (
 118             Ether(src=self.pg0.remote_mac, dst=self.pg1.remote_mac)
 119             / IP(src="1.1.1.1", dst="1.1.1.2")
 120             / UDP(sport=1234, dport=1234)
 121             / Raw(b"\xa5" * 100)
 122         )
 123
 124         self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
 125         self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
 126
 127         #
 128         # Attach ACL to ensure features are run on the pipe
 129         #
 130         rule_1 = AclRule(
 131             is_permit=0,
 132             proto=17,
 133             src_prefix=IPv4Network("1.1.1.1/32"),
 134             dst_prefix=IPv4Network("1.1.1.2/32"),
 135             ports=1234,
 136         )
 137         acl = VppAcl(self, rules=[rule_1])
 138         acl.add_vpp_config()
 139
 140         # Apply the ACL on the pipe on output
 141         acl_if_e = VppAclInterface(
 142             self, sw_if_index=pipes[0].east, n_input=0, acls=[acl]
 143         )
 144         acl_if_e.add_vpp_config()
 145
 146         self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
 147         self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
 148
 149         # remove from output and apply on input
 150         acl_if_e.remove_vpp_config()
 151         acl_if_w = VppAclInterface(
 152             self, sw_if_index=pipes[0].west, n_input=1, acls=[acl]
 153         )
 154         acl_if_w.add_vpp_config()
 155
 156         self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
 157         self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
 158
 159         acl_if_w.remove_vpp_config()
 160         self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
 161         self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
 162
 163         #
 164         # L3 routes in two separate tables so a pipe can be used to L3
 165         # x-connect
 166         #
 167         tables = []
 168         tables.append(VppIpTable(self, 1))
 169         tables.append(VppIpTable(self, 2))
 170
 171         for t in tables:
 172             t.add_vpp_config()
 173
 174         self.pg2.set_table_ip4(1)
 175         self.pg2.config_ip4()
 176         self.pg2.resolve_arp()
 177         self.pg3.set_table_ip4(2)
 178         self.pg3.config_ip4()
 179         self.pg3.resolve_arp()
 180
 181         routes = []
 182         routes.append(
 183             VppIpRoute(
 184                 self,
 185                 "1.1.1.1",
 186                 32,
 187                 [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)],
 188                 table_id=2,
 189             )
 190         )
 191         routes.append(
 192             VppIpRoute(
 193                 self,
 194                 "1.1.1.1",
 195                 32,
 196                 [VppRoutePath("0.0.0.0", pipes[1].east)],
 197                 table_id=1,
 198             )
 199         )
 200         routes.append(
 201             VppIpRoute(
 202                 self,
 203                 "1.1.1.2",
 204                 32,
 205                 [VppRoutePath("0.0.0.0", pipes[1].west)],
 206                 table_id=2,
 207             )
 208         )
 209         routes.append(
 210             VppIpRoute(
 211                 self,
 212                 "1.1.1.2",
 213                 32,
 214                 [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index)],
 215                 table_id=1,
 216             )
 217         )
 218
 219         for r in routes:
 220             r.add_vpp_config()
 221
 222         p_east = (
 223             Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
 224             / IP(src="1.1.1.2", dst="1.1.1.1")
 225             / UDP(sport=1234, dport=1234)
 226             / Raw(b"\xa5" * 100)
 227         )
 228
 229         # bind the pipe ends to the correct tables
 230         self.vapi.sw_interface_set_table(pipes[1].west, 0, 2)
 231         self.vapi.sw_interface_set_table(pipes[1].east, 0, 1)
 232
 233         # IP is not enabled on the pipes at this point
 234         self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
 235
 236         # IP enable the Pipes by making them unnumbered
 237         pipes[1].set_unnumbered(self.pg2.sw_if_index, True)
 238         pipes[1].set_unnumbered(self.pg3.sw_if_index, False)
 239
 240         self.send_and_expect(self.pg2, p_east * NUM_PKTS, self.pg3)
 241
 242         # and the return path
 243         p_west = (
 244             Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac)
 245             / IP(src="1.1.1.1", dst="1.1.1.2")
 246             / UDP(sport=1234, dport=1234)
 247             / Raw(b"\xa5" * 100)
 248         )
 249         self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
 250
 251         #
 252         # Use ACLs to test features run on the Pipes
 253         #
 254         acl_if_e1 = VppAclInterface(
 255             self, sw_if_index=pipes[1].east, n_input=0, acls=[acl]
 256         )
 257         acl_if_e1.add_vpp_config()
 258         self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
 259         self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
 260
 261         # remove from output and apply on input
 262         acl_if_e1.remove_vpp_config()
 263         acl_if_w1 = VppAclInterface(
 264             self, sw_if_index=pipes[1].west, n_input=1, acls=[acl]
 265         )
 266         acl_if_w1.add_vpp_config()
 267         self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
 268         self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
 269         acl_if_w1.remove_vpp_config()
 270
 271         self.send_and_expect(self.pg2, p_east * NUM_PKTS, self.pg3)
 272         self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
 273
 274         # cleanup (so the tables delete)
 275         self.pg2.unconfig_ip4()
 276         self.pg2.set_table_ip4(0)
 277         self.pg3.unconfig_ip4()
 278         self.pg3.set_table_ip4(0)
 279         self.vapi.sw_interface_set_table(pipes[1].west, 0, 0)
 280         self.vapi.sw_interface_set_table(pipes[1].east, 0, 0)
 281
 282
 283 if __name__ == "__main__":
 284     unittest.main(testRunner=VppTestRunner)