Code Review
/
vpp.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
ip: change icmp4 throttle
[vpp.git]
/
test
/
test_pppoe.py
diff --git
a/test/test_pppoe.py
b/test/test_pppoe.py
index
1d0aeff
..
c83d7d5
100644
(file)
--- a/
test/test_pppoe.py
+++ b/
test/test_pppoe.py
@@
-1,26
+1,21
@@
-#!/usr/bin/env python
+#!/usr/bin/env python
3
+import socket
import unittest
import unittest
-from logging import *
-
-from framework import VppTestCase, VppTestRunner
-from vpp_ip_route import VppIpRoute, VppRoutePath
-from vpp_pppoe_interface import VppPppoeInterface, VppPppoe6Interface
-from vpp_papi_provider import L2_VTR_OP
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.ppp import PPPoE, PPPoED, PPP
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.ppp import PPPoE, PPPoED, PPP
-from scapy.layers.inet import IP, UDP
-from scapy.layers.inet6 import IPv6
-from scapy.volatile import RandMAC, RandIP
+from scapy.layers.inet import IP
-from util import ppp, ppc, mactobinary
-import socket
+from framework import VppTestCase, VppTestRunner
+from vpp_ip_route import VppIpRoute, VppRoutePath
+from vpp_pppoe_interface import VppPppoeInterface
+from util import ppp, ppc
class TestPPPoE(VppTestCase):
class TestPPPoE(VppTestCase):
- """
PPPoE Test Case
"""
+ """
PPPoE Test Case
"""
@classmethod
def setUpClass(cls):
@classmethod
def setUpClass(cls):
@@
-30,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)
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()
def setUp(self):
super(TestPPPoE, self).setUp()
@@
-44,18
+43,18
@@
class TestPPPoE(VppTestCase):
def tearDown(self):
super(TestPPPoE, self).tearDown()
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"))
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
packets = []
for i in range(count):
# create packet info stored in the test case instance
@@
-63,9
+62,11
@@
class TestPPPoE(VppTestCase):
# convert the info into packet payload
payload = self.info_to_payload(info)
# create the packet itself
# 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
# store a copy of the packet in the packet info
info.data = p.copy()
# append the packet to the list
@@
-74,8
+75,7
@@
class TestPPPoE(VppTestCase):
# return the created packet list
return packets
# 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
packets = []
for i in range(count):
# create packet info stored in the test case instance
@@
-83,10
+83,12
@@
class TestPPPoE(VppTestCase):
# convert the info into packet payload
payload = self.info_to_payload(info)
# create the packet itself
# 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
# store a copy of the packet in the packet info
info.data = p.copy()
# append the packet to the list
@@
-95,8
+97,9
@@
class TestPPPoE(VppTestCase):
# return the created packet list
return packets
# 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
packets = []
for i in range(count):
# create packet info stored in the test case instance
@@
-104,11
+107,13
@@
class TestPPPoE(VppTestCase):
# convert the info into packet payload
payload = self.info_to_payload(info)
# create the packet itself
# 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
# store a copy of the packet in the packet info
info.data = p.copy()
# append the packet to the list
@@
-125,9
+130,11
@@
class TestPPPoE(VppTestCase):
# convert the info into packet payload
payload = self.info_to_payload(info)
# create the packet itself
# 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
# store a copy of the packet in the packet info
info.data = p.copy()
# append the packet to the list
@@
-180,46
+187,53
@@
class TestPPPoE(VppTestCase):
raise
def test_PPPoE_Decap(self):
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
#
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
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
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
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.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
#
#
# 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)
self.pg0.add_stream(tx2)
self.pg_enable_capture(self.pg_interfaces)
@@
-243,45
+257,50
@@
class TestPPPoE(VppTestCase):
route_sever_dst.remove_vpp_config()
def test_PPPoE_Encap(self):
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
#
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
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
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
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.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")
#
# 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)
self.pg1.add_stream(tx2)
self.pg_enable_capture(self.pg_interfaces)
@@
-306,44
+325,48
@@
class TestPPPoE(VppTestCase):
route_sever_dst.remove_vpp_config()
def test_PPPoE_Add_Twice(self):
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
#
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
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
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
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.add_vpp_config()
+ pppoe_if.set_unnumbered(self.pg0.sw_if_index)
#
# The double create (create the same session twice) should fail,
# and we should still be able to use the original
#
try:
#
# The double create (create the same session twice) should fail,
# and we should still be able to use the original
#
try:
-
gr
e_if.add_vpp_config()
+
pppo
e_if.add_vpp_config()
except Exception:
pass
else:
except Exception:
pass
else:
@@
-360,36
+383,39
@@
class TestPPPoE(VppTestCase):
route_sever_dst.remove_vpp_config()
def test_PPPoE_Del_Twice(self):
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
#
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
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
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
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
pppoe_if.add_vpp_config()
# Delete PPPoE session
@@
-400,7
+426,7
@@
class TestPPPoE(VppTestCase):
# and we should still be able to use the original
#
try:
# and we should still be able to use the original
#
try:
-
gr
e_if.remove_vpp_config()
+
pppo
e_if.remove_vpp_config()
except Exception:
pass
else:
except Exception:
pass
else:
@@
-414,66
+440,74
@@
class TestPPPoE(VppTestCase):
route_sever_dst.remove_vpp_config()
def test_PPPoE_Decap_Multiple(self):
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
#
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
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
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
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.add_vpp_config()
+ pppoe_if1.set_unnumbered(self.pg0.sw_if_index)
# Send PPPoE Discovery 2
# 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
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
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.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
#
#
# 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)
self.pg0.add_stream(tx2)
self.pg_enable_capture(self.pg_interfaces)
@@
-482,10
+516,13
@@
class TestPPPoE(VppTestCase):
rx2 = self.pg1.get_capture(len(tx2))
self.verify_decapped_pppoe(self.pg0, rx2, tx2)
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)
self.pg2.add_stream(tx5)
self.pg_enable_capture(self.pg_interfaces)
@@
-510,56
+547,59
@@
class TestPPPoE(VppTestCase):
route_sever_dst.remove_vpp_config()
def test_PPPoE_Encap_Multiple(self):
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
#
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
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
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
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
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
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
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.add_vpp_config()
#
@@
-567,8
+607,9
@@
class TestPPPoE(VppTestCase):
# - packets are PPPoE encapped
#
self.vapi.cli("clear trace")
# - 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)
self.pg1.add_stream(tx2)
self.pg_enable_capture(self.pg_interfaces)
@@
-577,8
+618,9
@@
class TestPPPoE(VppTestCase):
rx2 = self.pg0.get_capture(len(tx2))
self.verify_encaped_pppoe(self.pg1, rx2, tx2, self.session_id)
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)
self.pg1.add_stream(tx5)
self.pg_enable_capture(self.pg_interfaces)
@@
-602,5
+644,6
@@
class TestPPPoE(VppTestCase):
# Delete a route that resolves the server's destination
route_sever_dst.remove_vpp_config()
# 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)
unittest.main(testRunner=VppTestRunner)