X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_pppoe.py;h=bd66c313e42b50f9d42ea6ba5bb7581fb8d41dd3;hb=109f3ce4eba0a6ba869752f24cfaae73ab70bb2d;hp=b6a1eef4b80d1679c01784033634a871f514b9ad;hpb=a279d9cf615bd60439085ba103125e6f2fa3b27c;p=vpp.git diff --git a/test/test_pppoe.py b/test/test_pppoe.py index b6a1eef4b80..bd66c313e42 100644 --- a/test/test_pppoe.py +++ b/test/test_pppoe.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 import socket import unittest @@ -15,7 +15,7 @@ from util import ppp, ppc class TestPPPoE(VppTestCase): - """ PPPoE Test Case """ + """PPPoE Test Case""" @classmethod def setUpClass(cls): @@ -25,6 +25,10 @@ class TestPPPoE(VppTestCase): cls.dst_ip = "100.1.1.100" cls.dst_ipn = socket.inet_pton(socket.AF_INET, cls.dst_ip) + @classmethod + def tearDownClass(cls): + super(TestPPPoE, cls).tearDownClass() + def setUp(self): super(TestPPPoE, self).setUp() @@ -39,18 +43,18 @@ class TestPPPoE(VppTestCase): def tearDown(self): super(TestPPPoE, self).tearDown() + for i in self.pg_interfaces: + i.unconfig_ip4() + i.admin_down() + + def show_commands_at_teardown(self): self.logger.info(self.vapi.cli("show int")) self.logger.info(self.vapi.cli("show pppoe fib")) self.logger.info(self.vapi.cli("show pppoe session")) self.logger.info(self.vapi.cli("show ip fib")) self.logger.info(self.vapi.cli("show trace")) - for i in self.pg_interfaces: - i.unconfig_ip4() - i.admin_down() - - def create_stream_pppoe_discovery(self, src_if, dst_if, - client_mac, count=1): + def create_stream_pppoe_discovery(self, src_if, dst_if, client_mac, count=1): packets = [] for i in range(count): # create packet info stored in the test case instance @@ -58,9 +62,11 @@ class TestPPPoE(VppTestCase): # convert the info into packet payload payload = self.info_to_payload(info) # create the packet itself - p = (Ether(dst=src_if.local_mac, src=client_mac) / - PPPoED(sessionid=0) / - Raw(payload)) + p = ( + Ether(dst=src_if.local_mac, src=client_mac) + / PPPoED(sessionid=0) + / Raw(payload) + ) # store a copy of the packet in the packet info info.data = p.copy() # append the packet to the list @@ -69,8 +75,7 @@ class TestPPPoE(VppTestCase): # return the created packet list return packets - def create_stream_pppoe_lcp(self, src_if, dst_if, - client_mac, session_id, count=1): + def create_stream_pppoe_lcp(self, src_if, dst_if, client_mac, session_id, count=1): packets = [] for i in range(count): # create packet info stored in the test case instance @@ -78,10 +83,12 @@ class TestPPPoE(VppTestCase): # convert the info into packet payload payload = self.info_to_payload(info) # create the packet itself - p = (Ether(dst=src_if.local_mac, src=client_mac) / - PPPoE(sessionid=session_id) / - PPP(proto=0xc021) / - Raw(payload)) + p = ( + Ether(dst=src_if.local_mac, src=client_mac) + / PPPoE(sessionid=session_id) + / PPP(proto=0xC021) + / Raw(payload) + ) # store a copy of the packet in the packet info info.data = p.copy() # append the packet to the list @@ -90,8 +97,9 @@ class TestPPPoE(VppTestCase): # return the created packet list return packets - def create_stream_pppoe_ip4(self, src_if, dst_if, - client_mac, session_id, client_ip, count=1): + def create_stream_pppoe_ip4( + self, src_if, dst_if, client_mac, session_id, client_ip, count=1 + ): packets = [] for i in range(count): # create packet info stored in the test case instance @@ -99,11 +107,13 @@ class TestPPPoE(VppTestCase): # convert the info into packet payload payload = self.info_to_payload(info) # create the packet itself - p = (Ether(dst=src_if.local_mac, src=client_mac) / - PPPoE(sessionid=session_id) / - PPP(proto=0x0021) / - IP(src=client_ip, dst=self.dst_ip) / - Raw(payload)) + p = ( + Ether(dst=src_if.local_mac, src=client_mac) + / PPPoE(sessionid=session_id) + / PPP(proto=0x0021) + / IP(src=client_ip, dst=self.dst_ip) + / Raw(payload) + ) # store a copy of the packet in the packet info info.data = p.copy() # append the packet to the list @@ -120,9 +130,11 @@ class TestPPPoE(VppTestCase): # convert the info into packet payload payload = self.info_to_payload(info) # create the packet itself - p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / - IP(src=dst_ip, dst=client_ip) / - Raw(payload)) + p = ( + Ether(dst=src_if.local_mac, src=src_if.remote_mac) + / IP(src=dst_ip, dst=client_ip) + / Raw(payload) + ) # store a copy of the packet in the packet info info.data = p.copy() # append the packet to the list @@ -151,7 +163,6 @@ class TestPPPoE(VppTestCase): raise def verify_encaped_pppoe(self, src_if, capture, sent, session_id): - self.assertEqual(len(capture), len(sent)) for i in range(len(capture)): @@ -175,46 +186,53 @@ class TestPPPoE(VppTestCase): raise def test_PPPoE_Decap(self): - """ PPPoE Decap Test """ + """PPPoE Decap Test""" self.vapi.cli("clear trace") # # Add a route that resolves the server's destination # - route_sever_dst = VppIpRoute(self, "100.1.1.100", 32, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + route_sever_dst = VppIpRoute( + self, + "100.1.1.100", + 32, + [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], + ) route_sever_dst.add_vpp_config() # Send PPPoE Discovery - tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1, - self.pg0.remote_mac) + tx0 = self.create_stream_pppoe_discovery( + self.pg0, self.pg1, self.pg0.remote_mac + ) self.pg0.add_stream(tx0) self.pg_start() # Send PPPoE PPP LCP - tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id) + tx1 = self.create_stream_pppoe_lcp( + self.pg0, self.pg1, self.pg0.remote_mac, self.session_id + ) self.pg0.add_stream(tx1) self.pg_start() # Create PPPoE session - pppoe_if = VppPppoeInterface(self, - self.pg0.remote_ip4, - self.pg0.remote_mac, - self.session_id) + pppoe_if = VppPppoeInterface( + self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id + ) pppoe_if.add_vpp_config() + pppoe_if.set_unnumbered(self.pg0.sw_if_index) # # Send tunneled packets that match the created tunnel and # are decapped and forwarded # - tx2 = self.create_stream_pppoe_ip4(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id, - self.pg0.remote_ip4) + tx2 = self.create_stream_pppoe_ip4( + self.pg0, + self.pg1, + self.pg0.remote_mac, + self.session_id, + self.pg0.remote_ip4, + ) self.pg0.add_stream(tx2) self.pg_enable_capture(self.pg_interfaces) @@ -238,45 +256,50 @@ class TestPPPoE(VppTestCase): route_sever_dst.remove_vpp_config() def test_PPPoE_Encap(self): - """ PPPoE Encap Test """ + """PPPoE Encap Test""" self.vapi.cli("clear trace") # # Add a route that resolves the server's destination # - route_sever_dst = VppIpRoute(self, "100.1.1.100", 32, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + route_sever_dst = VppIpRoute( + self, + "100.1.1.100", + 32, + [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], + ) route_sever_dst.add_vpp_config() # Send PPPoE Discovery - tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1, - self.pg0.remote_mac) + tx0 = self.create_stream_pppoe_discovery( + self.pg0, self.pg1, self.pg0.remote_mac + ) self.pg0.add_stream(tx0) self.pg_start() # Send PPPoE PPP LCP - tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id) + tx1 = self.create_stream_pppoe_lcp( + self.pg0, self.pg1, self.pg0.remote_mac, self.session_id + ) self.pg0.add_stream(tx1) self.pg_start() # Create PPPoE session - pppoe_if = VppPppoeInterface(self, - self.pg0.remote_ip4, - self.pg0.remote_mac, - self.session_id) + pppoe_if = VppPppoeInterface( + self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id + ) pppoe_if.add_vpp_config() + pppoe_if.set_unnumbered(self.pg0.sw_if_index) # # Send a packet stream that is routed into the session # - packets are PPPoE encapped # self.vapi.cli("clear trace") - tx2 = self.create_stream_ip4(self.pg1, self.pg0, - self.pg0.remote_ip4, self.dst_ip, 65) + tx2 = self.create_stream_ip4( + self.pg1, self.pg0, self.pg0.remote_ip4, self.dst_ip, 65 + ) self.pg1.add_stream(tx2) self.pg_enable_capture(self.pg_interfaces) @@ -301,37 +324,41 @@ class TestPPPoE(VppTestCase): route_sever_dst.remove_vpp_config() def test_PPPoE_Add_Twice(self): - """ PPPoE Add Same Session Twice Test """ + """PPPoE Add Same Session Twice Test""" self.vapi.cli("clear trace") # # Add a route that resolves the server's destination # - route_sever_dst = VppIpRoute(self, "100.1.1.100", 32, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + route_sever_dst = VppIpRoute( + self, + "100.1.1.100", + 32, + [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], + ) route_sever_dst.add_vpp_config() # Send PPPoE Discovery - tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1, - self.pg0.remote_mac) + tx0 = self.create_stream_pppoe_discovery( + self.pg0, self.pg1, self.pg0.remote_mac + ) self.pg0.add_stream(tx0) self.pg_start() # Send PPPoE PPP LCP - tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id) + tx1 = self.create_stream_pppoe_lcp( + self.pg0, self.pg1, self.pg0.remote_mac, self.session_id + ) self.pg0.add_stream(tx1) self.pg_start() # Create PPPoE session - pppoe_if = VppPppoeInterface(self, - self.pg0.remote_ip4, - self.pg0.remote_mac, - self.session_id) + pppoe_if = VppPppoeInterface( + self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id + ) pppoe_if.add_vpp_config() + pppoe_if.set_unnumbered(self.pg0.sw_if_index) # # The double create (create the same session twice) should fail, @@ -355,36 +382,39 @@ class TestPPPoE(VppTestCase): route_sever_dst.remove_vpp_config() def test_PPPoE_Del_Twice(self): - """ PPPoE Delete Same Session Twice Test """ + """PPPoE Delete Same Session Twice Test""" self.vapi.cli("clear trace") # # Add a route that resolves the server's destination # - route_sever_dst = VppIpRoute(self, "100.1.1.100", 32, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + route_sever_dst = VppIpRoute( + self, + "100.1.1.100", + 32, + [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], + ) route_sever_dst.add_vpp_config() # Send PPPoE Discovery - tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1, - self.pg0.remote_mac) + tx0 = self.create_stream_pppoe_discovery( + self.pg0, self.pg1, self.pg0.remote_mac + ) self.pg0.add_stream(tx0) self.pg_start() # Send PPPoE PPP LCP - tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id) + tx1 = self.create_stream_pppoe_lcp( + self.pg0, self.pg1, self.pg0.remote_mac, self.session_id + ) self.pg0.add_stream(tx1) self.pg_start() # Create PPPoE session - pppoe_if = VppPppoeInterface(self, - self.pg0.remote_ip4, - self.pg0.remote_mac, - self.session_id) + pppoe_if = VppPppoeInterface( + self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id + ) pppoe_if.add_vpp_config() # Delete PPPoE session @@ -409,66 +439,74 @@ class TestPPPoE(VppTestCase): route_sever_dst.remove_vpp_config() def test_PPPoE_Decap_Multiple(self): - """ PPPoE Decap Multiple Sessions Test """ + """PPPoE Decap Multiple Sessions Test""" self.vapi.cli("clear trace") # # Add a route that resolves the server's destination # - route_sever_dst = VppIpRoute(self, "100.1.1.100", 32, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + route_sever_dst = VppIpRoute( + self, + "100.1.1.100", + 32, + [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], + ) route_sever_dst.add_vpp_config() # Send PPPoE Discovery 1 - tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1, - self.pg0.remote_mac) + tx0 = self.create_stream_pppoe_discovery( + self.pg0, self.pg1, self.pg0.remote_mac + ) self.pg0.add_stream(tx0) self.pg_start() # Send PPPoE PPP LCP 1 - tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id) + tx1 = self.create_stream_pppoe_lcp( + self.pg0, self.pg1, self.pg0.remote_mac, self.session_id + ) self.pg0.add_stream(tx1) self.pg_start() # Create PPPoE session 1 - pppoe_if1 = VppPppoeInterface(self, - self.pg0.remote_ip4, - self.pg0.remote_mac, - self.session_id) + pppoe_if1 = VppPppoeInterface( + self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id + ) pppoe_if1.add_vpp_config() + pppoe_if1.set_unnumbered(self.pg0.sw_if_index) # Send PPPoE Discovery 2 - tx3 = self.create_stream_pppoe_discovery(self.pg2, self.pg1, - self.pg2.remote_mac) + tx3 = self.create_stream_pppoe_discovery( + self.pg2, self.pg1, self.pg2.remote_mac + ) self.pg2.add_stream(tx3) self.pg_start() # Send PPPoE PPP LCP 2 - tx4 = self.create_stream_pppoe_lcp(self.pg2, self.pg1, - self.pg2.remote_mac, - self.session_id + 1) + tx4 = self.create_stream_pppoe_lcp( + self.pg2, self.pg1, self.pg2.remote_mac, self.session_id + 1 + ) self.pg2.add_stream(tx4) self.pg_start() # Create PPPoE session 2 - pppoe_if2 = VppPppoeInterface(self, - self.pg2.remote_ip4, - self.pg2.remote_mac, - self.session_id + 1) + pppoe_if2 = VppPppoeInterface( + self, self.pg2.remote_ip4, self.pg2.remote_mac, self.session_id + 1 + ) pppoe_if2.add_vpp_config() + pppoe_if2.set_unnumbered(self.pg0.sw_if_index) # # Send tunneled packets that match the created tunnel and # are decapped and forwarded # - tx2 = self.create_stream_pppoe_ip4(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id, - self.pg0.remote_ip4) + tx2 = self.create_stream_pppoe_ip4( + self.pg0, + self.pg1, + self.pg0.remote_mac, + self.session_id, + self.pg0.remote_ip4, + ) self.pg0.add_stream(tx2) self.pg_enable_capture(self.pg_interfaces) @@ -477,10 +515,13 @@ class TestPPPoE(VppTestCase): rx2 = self.pg1.get_capture(len(tx2)) self.verify_decapped_pppoe(self.pg0, rx2, tx2) - tx5 = self.create_stream_pppoe_ip4(self.pg2, self.pg1, - self.pg2.remote_mac, - self.session_id + 1, - self.pg2.remote_ip4) + tx5 = self.create_stream_pppoe_ip4( + self.pg2, + self.pg1, + self.pg2.remote_mac, + self.session_id + 1, + self.pg2.remote_ip4, + ) self.pg2.add_stream(tx5) self.pg_enable_capture(self.pg_interfaces) @@ -505,56 +546,59 @@ class TestPPPoE(VppTestCase): route_sever_dst.remove_vpp_config() def test_PPPoE_Encap_Multiple(self): - """ PPPoE Encap Multiple Sessions Test """ + """PPPoE Encap Multiple Sessions Test""" self.vapi.cli("clear trace") # # Add a route that resolves the server's destination # - route_sever_dst = VppIpRoute(self, "100.1.1.100", 32, - [VppRoutePath(self.pg1.remote_ip4, - self.pg1.sw_if_index)]) + route_sever_dst = VppIpRoute( + self, + "100.1.1.100", + 32, + [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)], + ) route_sever_dst.add_vpp_config() # Send PPPoE Discovery 1 - tx0 = self.create_stream_pppoe_discovery(self.pg0, self.pg1, - self.pg0.remote_mac) + tx0 = self.create_stream_pppoe_discovery( + self.pg0, self.pg1, self.pg0.remote_mac + ) self.pg0.add_stream(tx0) self.pg_start() # Send PPPoE PPP LCP 1 - tx1 = self.create_stream_pppoe_lcp(self.pg0, self.pg1, - self.pg0.remote_mac, - self.session_id) + tx1 = self.create_stream_pppoe_lcp( + self.pg0, self.pg1, self.pg0.remote_mac, self.session_id + ) self.pg0.add_stream(tx1) self.pg_start() # Create PPPoE session 1 - pppoe_if1 = VppPppoeInterface(self, - self.pg0.remote_ip4, - self.pg0.remote_mac, - self.session_id) + pppoe_if1 = VppPppoeInterface( + self, self.pg0.remote_ip4, self.pg0.remote_mac, self.session_id + ) pppoe_if1.add_vpp_config() # Send PPPoE Discovery 2 - tx3 = self.create_stream_pppoe_discovery(self.pg2, self.pg1, - self.pg2.remote_mac) + tx3 = self.create_stream_pppoe_discovery( + self.pg2, self.pg1, self.pg2.remote_mac + ) self.pg2.add_stream(tx3) self.pg_start() # Send PPPoE PPP LCP 2 - tx4 = self.create_stream_pppoe_lcp(self.pg2, self.pg1, - self.pg2.remote_mac, - self.session_id + 1) + tx4 = self.create_stream_pppoe_lcp( + self.pg2, self.pg1, self.pg2.remote_mac, self.session_id + 1 + ) self.pg2.add_stream(tx4) self.pg_start() # Create PPPoE session 2 - pppoe_if2 = VppPppoeInterface(self, - self.pg2.remote_ip4, - self.pg2.remote_mac, - self.session_id + 1) + pppoe_if2 = VppPppoeInterface( + self, self.pg2.remote_ip4, self.pg2.remote_mac, self.session_id + 1 + ) pppoe_if2.add_vpp_config() # @@ -562,8 +606,9 @@ class TestPPPoE(VppTestCase): # - packets are PPPoE encapped # self.vapi.cli("clear trace") - tx2 = self.create_stream_ip4(self.pg1, self.pg0, - self.pg0.remote_ip4, self.dst_ip) + tx2 = self.create_stream_ip4( + self.pg1, self.pg0, self.pg0.remote_ip4, self.dst_ip + ) self.pg1.add_stream(tx2) self.pg_enable_capture(self.pg_interfaces) @@ -572,8 +617,9 @@ class TestPPPoE(VppTestCase): rx2 = self.pg0.get_capture(len(tx2)) self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id) - tx5 = self.create_stream_ip4(self.pg1, self.pg2, - self.pg2.remote_ip4, self.dst_ip) + tx5 = self.create_stream_ip4( + self.pg1, self.pg2, self.pg2.remote_ip4, self.dst_ip + ) self.pg1.add_stream(tx5) self.pg_enable_capture(self.pg_interfaces) @@ -597,5 +643,6 @@ class TestPPPoE(VppTestCase): # Delete a route that resolves the server's destination route_sever_dst.remove_vpp_config() -if __name__ == '__main__': + +if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)