from scapy.layers.ppp import PPPoE, PPPoED, PPP
from scapy.layers.inet import IP
from scapy.layers.ppp import PPPoE, PPPoED, PPP
from scapy.layers.inet import IP
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_pppoe_interface import VppPppoeInterface
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_pppoe_interface import VppPppoeInterface
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):
# 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
- 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):
# 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)
+ )
- 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
+ ):
# 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)
+ )
# 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)
+ )
raise
def verify_encaped_pppoe(self, src_if, capture, sent, session_id):
raise
def verify_encaped_pppoe(self, src_if, capture, sent, session_id):
self.assertEqual(len(capture), len(sent))
for i in range(len(capture)):
self.assertEqual(len(capture), len(sent))
for i in range(len(capture)):
- 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)],
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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)],
+ )
- 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
+ )
- 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
+ )
- 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
+ )
#
# 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
+ )
- 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)],
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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)],
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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)],
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
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)
- 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)],
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
- 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
+ )
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
+ )