X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_pipe.py;h=83f5f96c998001b0c37025e59bdb05136c869a5d;hb=853cc9f2ad3ee52cbdd891fb09d51c25678baed0;hp=0e766654d2a137bf6fc261c8f12cf5be880e4fa6;hpb=aad1ee149403994194cf37cef4530b042ba7df3a;p=vpp.git diff --git a/test/test_pipe.py b/test/test_pipe.py index 0e766654d2a..83f5f96c998 100644 --- a/test/test_pipe.py +++ b/test/test_pipe.py @@ -1,5 +1,4 @@ #!/usr/bin/env python3 -from socket import AF_INET, AF_INET6, inet_pton import unittest from ipaddress import IPv4Network @@ -7,7 +6,8 @@ from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase +from asfframework import VppTestRunner from vpp_interface import VppInterface from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath from vpp_acl import AclRule, VppAcl, VppAclInterface @@ -28,20 +28,19 @@ class VppPipe(VppInterface): def west(self): return self.result.pipe_sw_if_index[0] - def __init__(self, test, instance=0xffffffff): + def __init__(self, test, instance=0xFFFFFFFF): super(VppPipe, self).__init__(test) self._test = test self.instance = instance def add_vpp_config(self): self.result = self._test.vapi.pipe_create( - 0 if self.instance == 0xffffffff else 1, - self.instance) + 0 if self.instance == 0xFFFFFFFF else 1, self.instance + ) self.set_sw_if_index(self.result.sw_if_index) def remove_vpp_config(self): - self._test.vapi.pipe_delete( - self.result.sw_if_index) + self._test.vapi.pipe_delete(self.result.sw_if_index) def object_id(self): return "pipe-%d" % (self._sw_if_index) @@ -53,15 +52,19 @@ class VppPipe(VppInterface): return True return False - def set_unnumbered(self, ip_sw_if_index, is_add=True): - res = self._test.vapi.sw_interface_set_unnumbered(ip_sw_if_index, - self.east, is_add) - res = self._test.vapi.sw_interface_set_unnumbered(ip_sw_if_index, - self.west, is_add) + def set_unnumbered(self, ip_sw_if_index, is_east, is_add=True): + if is_east: + res = self._test.vapi.sw_interface_set_unnumbered( + ip_sw_if_index, self.east, is_add + ) + else: + res = self._test.vapi.sw_interface_set_unnumbered( + ip_sw_if_index, self.west, is_add + ) class TestPipe(VppTestCase): - """ Pipes """ + """Pipes""" @classmethod def setUpClass(cls): @@ -86,7 +89,7 @@ class TestPipe(VppTestCase): super(TestPipe, self).tearDown() def test_pipe(self): - """ Pipes """ + """Pipes""" pipes = [VppPipe(self), VppPipe(self, 10)] @@ -97,26 +100,26 @@ class TestPipe(VppTestCase): # # L2 cross-connect pipe0 east with pg0 and west with pg1 # - self.vapi.sw_interface_set_l2_xconnect(self.pg0.sw_if_index, - pipes[0].east, - enable=1) - self.vapi.sw_interface_set_l2_xconnect(pipes[0].east, - self.pg0.sw_if_index, - enable=1) - self.vapi.sw_interface_set_l2_xconnect(self.pg1.sw_if_index, - pipes[0].west, - enable=1) - self.vapi.sw_interface_set_l2_xconnect(pipes[0].west, - self.pg1.sw_if_index, - enable=1) + self.vapi.sw_interface_set_l2_xconnect( + self.pg0.sw_if_index, pipes[0].east, enable=1 + ) + self.vapi.sw_interface_set_l2_xconnect( + pipes[0].east, self.pg0.sw_if_index, enable=1 + ) + self.vapi.sw_interface_set_l2_xconnect( + self.pg1.sw_if_index, pipes[0].west, enable=1 + ) + self.vapi.sw_interface_set_l2_xconnect( + pipes[0].west, self.pg1.sw_if_index, enable=1 + ) # test bi-directional L2 flow pg0<->pg1 - p = (Ether(src=self.pg0.remote_mac, - dst=self.pg1.remote_mac) / - IP(src="1.1.1.1", - dst="1.1.1.2") / - UDP(sport=1234, dport=1234) / - Raw(b'\xa5' * 100)) + p = ( + Ether(src=self.pg0.remote_mac, dst=self.pg1.remote_mac) + / IP(src="1.1.1.1", dst="1.1.1.2") + / UDP(sport=1234, dport=1234) + / Raw(b"\xa5" * 100) + ) self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1) self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0) @@ -124,15 +127,20 @@ class TestPipe(VppTestCase): # # Attach ACL to ensure features are run on the pipe # - rule_1 = AclRule(is_permit=0, proto=17, - src_prefix=IPv4Network("1.1.1.1/32"), - dst_prefix=IPv4Network("1.1.1.2/32"), ports=1234) + rule_1 = AclRule( + is_permit=0, + proto=17, + src_prefix=IPv4Network("1.1.1.1/32"), + dst_prefix=IPv4Network("1.1.1.2/32"), + ports=1234, + ) acl = VppAcl(self, rules=[rule_1]) acl.add_vpp_config() # Apply the ACL on the pipe on output - acl_if_e = VppAclInterface(self, sw_if_index=pipes[0].east, n_input=0, - acls=[acl]) + acl_if_e = VppAclInterface( + self, sw_if_index=pipes[0].east, n_input=0, acls=[acl] + ) acl_if_e.add_vpp_config() self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS) @@ -140,8 +148,9 @@ class TestPipe(VppTestCase): # remove from output and apply on input acl_if_e.remove_vpp_config() - acl_if_w = VppAclInterface(self, sw_if_index=pipes[0].west, n_input=1, - acls=[acl]) + acl_if_w = VppAclInterface( + self, sw_if_index=pipes[0].west, n_input=1, acls=[acl] + ) acl_if_w.add_vpp_config() self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS) @@ -170,30 +179,52 @@ class TestPipe(VppTestCase): self.pg3.resolve_arp() routes = [] - routes.append(VppIpRoute(self, "1.1.1.1", 32, - [VppRoutePath(self.pg3.remote_ip4, - self.pg3.sw_if_index)], - table_id=2)) - routes.append(VppIpRoute(self, "1.1.1.1", 32, - [VppRoutePath("0.0.0.0", pipes[1].east)], - table_id=1)) - routes.append(VppIpRoute(self, "1.1.1.2", 32, - [VppRoutePath("0.0.0.0", pipes[1].west)], - table_id=2)) - routes.append(VppIpRoute(self, "1.1.1.2", 32, - [VppRoutePath(self.pg2.remote_ip4, - self.pg2.sw_if_index)], - table_id=1)) + routes.append( + VppIpRoute( + self, + "1.1.1.1", + 32, + [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)], + table_id=2, + ) + ) + routes.append( + VppIpRoute( + self, + "1.1.1.1", + 32, + [VppRoutePath("0.0.0.0", pipes[1].east)], + table_id=1, + ) + ) + routes.append( + VppIpRoute( + self, + "1.1.1.2", + 32, + [VppRoutePath("0.0.0.0", pipes[1].west)], + table_id=2, + ) + ) + routes.append( + VppIpRoute( + self, + "1.1.1.2", + 32, + [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index)], + table_id=1, + ) + ) for r in routes: r.add_vpp_config() - p_east = (Ether(src=self.pg2.remote_mac, - dst=self.pg2.local_mac) / - IP(src="1.1.1.2", - dst="1.1.1.1") / - UDP(sport=1234, dport=1234) / - Raw(b'\xa5' * 100)) + p_east = ( + Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac) + / IP(src="1.1.1.2", dst="1.1.1.1") + / UDP(sport=1234, dport=1234) + / Raw(b"\xa5" * 100) + ) # bind the pipe ends to the correct tables self.vapi.sw_interface_set_table(pipes[1].west, 0, 2) @@ -203,33 +234,35 @@ class TestPipe(VppTestCase): self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS) # IP enable the Pipes by making them unnumbered - pipes[0].set_unnumbered(self.pg2.sw_if_index) - pipes[1].set_unnumbered(self.pg3.sw_if_index) + pipes[1].set_unnumbered(self.pg2.sw_if_index, True) + pipes[1].set_unnumbered(self.pg3.sw_if_index, False) self.send_and_expect(self.pg2, p_east * NUM_PKTS, self.pg3) # and the return path - p_west = (Ether(src=self.pg3.remote_mac, - dst=self.pg3.local_mac) / - IP(src="1.1.1.1", - dst="1.1.1.2") / - UDP(sport=1234, dport=1234) / - Raw(b'\xa5' * 100)) + p_west = ( + Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) + / IP(src="1.1.1.1", dst="1.1.1.2") + / UDP(sport=1234, dport=1234) + / Raw(b"\xa5" * 100) + ) self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2) # # Use ACLs to test features run on the Pipes # - acl_if_e1 = VppAclInterface(self, sw_if_index=pipes[1].east, n_input=0, - acls=[acl]) + acl_if_e1 = VppAclInterface( + self, sw_if_index=pipes[1].east, n_input=0, acls=[acl] + ) acl_if_e1.add_vpp_config() self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS) self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2) # remove from output and apply on input acl_if_e1.remove_vpp_config() - acl_if_w1 = VppAclInterface(self, sw_if_index=pipes[1].west, n_input=1, - acls=[acl]) + acl_if_w1 = VppAclInterface( + self, sw_if_index=pipes[1].west, n_input=1, acls=[acl] + ) acl_if_w1.add_vpp_config() self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS) self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2) @@ -247,5 +280,5 @@ class TestPipe(VppTestCase): self.vapi.sw_interface_set_table(pipes[1].east, 0, 0) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)