X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Ftest_pppoe.py;h=e396250621f0f2c2d78f985392464a2704fc857f;hb=853cc9f2ad3ee52cbdd891fb09d51c25678baed0;hp=5c8495fed987773545ea7954657b4a31ecc764c6;hpb=7f9b7f9f492d1748d8ba025b3a713058fdb1943d;p=vpp.git diff --git a/test/test_pppoe.py b/test/test_pppoe.py index 5c8495fed98..e396250621f 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 @@ -8,14 +8,15 @@ from scapy.layers.l2 import Ether from scapy.layers.ppp import PPPoE, PPPoED, PPP from scapy.layers.inet import IP -from framework import VppTestCase, VppTestRunner +from framework import VppTestCase +from asfframework import VppTestRunner from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_pppoe_interface import VppPppoeInterface -from util import ppp, ppc +from util import ppp class TestPPPoE(VppTestCase): - """ PPPoE Test Case """ + """PPPoE Test Case""" @classmethod def setUpClass(cls): @@ -43,18 +44,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 @@ -62,9 +63,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 @@ -73,8 +76,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 @@ -82,10 +84,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 @@ -94,8 +98,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 @@ -103,11 +108,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 @@ -124,9 +131,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 @@ -155,7 +164,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)): @@ -179,46 +187,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) @@ -242,45 +257,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) @@ -305,37 +325,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, @@ -359,36 +383,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 @@ -413,66 +440,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) @@ -481,10 +516,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) @@ -509,56 +547,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() # @@ -566,8 +607,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) @@ -576,8 +618,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) @@ -601,5 +644,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)