#!/usr/bin/env python
-from abc import *
-import os
import subprocess
import unittest
import tempfile
from inspect import getdoc
from hook import StepHook, PollHook
from vpp_pg_interface import VppPGInterface
+from vpp_lo_interface import VppLoInterface
from vpp_papi_provider import VppPapiProvider
from scapy.packet import Raw
from log import *
cls.pg_interfaces = result
return result
+ @classmethod
+ def create_loopback_interfaces(cls, interfaces):
+ """
+ Create loopback interfaces
+
+ :param interfaces: iterable indexes of the interfaces
+
+ """
+ result = []
+ for i in interfaces:
+ intf = VppLoInterface(cls, i)
+ setattr(cls, intf.name, intf)
+ result.append(intf)
+ cls.lo_interfaces = result
+ return result
+
@staticmethod
def extend_packet(packet, size):
"""
--- /dev/null
+#!/usr/bin/env python
+import unittest
+from random import choice, randint
+
+from scapy.packet import Raw
+from scapy.layers.l2 import Ether
+from scapy.layers.inet import IP, UDP
+from logging import *
+
+from framework import VppTestCase, VppTestRunner
+
+""" IRB Test Case
+
+config
+ L2 MAC learning enabled in l2bd
+ 2 routed interfaces untagged, bvi
+ 2 bridged interfaces in l2bd with bvi
+test
+ sending ip4 eth pkts between routed interfaces
+ 2 routed interfaces
+ 2 bridged interfaces
+ 64B, 512B, 1518B, 9200B (ether_size)
+ burst of pkts per interface
+ 257pkts per burst
+ routed pkts hitting different FIB entries
+ bridged pkts hitting different MAC entries
+verify
+ all packets received correctly
+"""
+
+
+class TestIpIrb(VppTestCase):
+ """ IRB Test Case """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestIpIrb, cls).setUpClass()
+
+ cls.pg_if_packet_sizes = [64, 512, 1518, 9018] # packet sizes
+ cls.bd_id = 10
+ cls.remote_hosts_count = 250
+
+ # create 3 pg interfaces, 1 loopback interface
+ cls.create_pg_interfaces(range(3))
+ cls.create_loopback_interfaces(range(1))
+
+ cls.interfaces = list(cls.pg_interfaces)
+ cls.interfaces.extend(cls.lo_interfaces)
+
+ for i in cls.interfaces:
+ i.admin_up()
+
+ # Create BD with MAC learning enabled and put interfaces to this BD
+ cls.vapi.sw_interface_set_l2_bridge(
+ cls.loop0.sw_if_index, bd_id=cls.bd_id, bvi=1)
+ cls.vapi.sw_interface_set_l2_bridge(
+ cls.pg0.sw_if_index, bd_id=cls.bd_id)
+ cls.vapi.sw_interface_set_l2_bridge(
+ cls.pg1.sw_if_index, bd_id=cls.bd_id)
+
+ cls.loop0.config_ip4()
+ cls.pg2.config_ip4()
+
+ # configure MAC address binding to IPv4 neighbors on loop0
+ cls.loop0.generate_remote_hosts(cls.remote_hosts_count)
+ cls.loop0.configure_extend_ipv4_mac_binding()
+ # configure MAC address on pg2
+ cls.pg2.resolve_arp()
+
+ # one half of hosts are behind pg0 second behind pg1
+ half = cls.remote_hosts_count // 2
+ cls.pg0.remote_hosts = cls.loop0.remote_hosts[:half]
+ cls.pg1.remote_hosts = cls.loop0.remote_hosts[half:]
+
+ def tearDown(self):
+ super(TestIpIrb, self).tearDown()
+ if not self.vpp_dead:
+ info(self.vapi.cli("show l2patch"))
+ info(self.vapi.cli("show l2fib verbose"))
+ info(self.vapi.cli("show bridge-domain %s detail" % self.bd_id))
+ info(self.vapi.cli("show ip arp"))
+
+ def create_stream(self, src_ip_if, dst_ip_if, packet_sizes):
+ pkts = []
+ for i in range(0, 257):
+ remote_dst_host = choice(dst_ip_if.remote_hosts)
+ info = self.create_packet_info(
+ src_ip_if.sw_if_index, dst_ip_if.sw_if_index)
+ payload = self.info_to_payload(info)
+ p = (Ether(dst=src_ip_if.local_mac, src=src_ip_if.remote_mac) /
+ IP(src=src_ip_if.remote_ip4,
+ dst=remote_dst_host.ip4) /
+ UDP(sport=1234, dport=1234) /
+ Raw(payload))
+ info.data = p.copy()
+ size = packet_sizes[(i // 2) % len(packet_sizes)]
+ self.extend_packet(p, size)
+ pkts.append(p)
+ return pkts
+
+ def create_stream_l2_to_ip(self, src_l2_if, src_ip_if, dst_ip_if,
+ packet_sizes):
+ pkts = []
+ for i in range(0, 257):
+ info = self.create_packet_info(
+ src_ip_if.sw_if_index, dst_ip_if.sw_if_index)
+ payload = self.info_to_payload(info)
+
+ host = choice(src_l2_if.remote_hosts)
+
+ p = (Ether(src=host.mac,
+ dst = src_ip_if.local_mac) /
+ IP(src=host.ip4,
+ dst=dst_ip_if.remote_ip4) /
+ UDP(sport=1234, dport=1234) /
+ Raw(payload))
+
+ info.data = p.copy()
+ size = packet_sizes[(i // 2) % len(packet_sizes)]
+ self.extend_packet(p, size)
+
+ pkts.append(p)
+ return pkts
+
+ def verify_capture_l2_to_ip(self, dst_ip_if, src_ip_if, capture):
+ last_info = dict()
+ for i in self.interfaces:
+ last_info[i.sw_if_index] = None
+
+ dst_ip_sw_if_index = dst_ip_if.sw_if_index
+
+ for packet in capture:
+ ip = packet[IP]
+ udp = packet[IP][UDP]
+ payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
+ packet_index = payload_info.index
+
+ self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
+
+ next_info = self.get_next_packet_info_for_interface2(
+ payload_info.src, dst_ip_sw_if_index,
+ last_info[payload_info.src])
+ last_info[payload_info.src] = next_info
+ self.assertTrue(next_info is not None)
+ saved_packet = next_info.data
+ self.assertTrue(next_info is not None)
+
+ # MAC: src, dst
+ self.assertEqual(packet.src, dst_ip_if.local_mac)
+ self.assertEqual(packet.dst, dst_ip_if.remote_mac)
+
+ # IP: src, dst
+ host = src_ip_if.host_by_ip4(ip.src)
+ self.assertIsNotNone(host)
+ self.assertEqual(ip.dst, saved_packet[IP].dst)
+ self.assertEqual(ip.dst, dst_ip_if.remote_ip4)
+
+ # UDP:
+ self.assertEqual(udp.sport, saved_packet[UDP].sport)
+ self.assertEqual(udp.dport, saved_packet[UDP].dport)
+
+ def verify_capture(self, dst_ip_if, src_ip_if, capture):
+ last_info = dict()
+ for i in self.interfaces:
+ last_info[i.sw_if_index] = None
+
+ dst_ip_sw_if_index = dst_ip_if.sw_if_index
+
+ for packet in capture:
+ ip = packet[IP]
+ udp = packet[IP][UDP]
+ payload_info = self.payload_to_info(str(packet[IP][UDP][Raw]))
+ packet_index = payload_info.index
+
+ self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
+
+ next_info = self.get_next_packet_info_for_interface2(
+ payload_info.src, dst_ip_sw_if_index,
+ last_info[payload_info.src])
+ last_info[payload_info.src] = next_info
+ self.assertTrue(next_info is not None)
+ self.assertEqual(packet_index, next_info.index)
+ saved_packet = next_info.data
+ self.assertTrue(next_info is not None)
+
+ # MAC: src, dst
+ self.assertEqual(packet.src, dst_ip_if.local_mac)
+ host = dst_ip_if.host_by_mac(packet.dst)
+
+ # IP: src, dst
+ self.assertEqual(ip.src, src_ip_if.remote_ip4)
+ self.assertEqual(ip.dst, saved_packet[IP].dst)
+ self.assertEqual(ip.dst, host.ip4)
+
+ # UDP:
+ self.assertEqual(udp.sport, saved_packet[UDP].sport)
+ self.assertEqual(udp.dport, saved_packet[UDP].dport)
+
+ def test_ip4_irb_1(self):
+ """ IPv4 IRB test 1
+
+ Test scenario:
+ ip traffic from pg2 interface must ends in both pg0 and pg1
+ - arp entry present in loop0 interface for dst IP
+ - no l2 entree configured, pg0 and pg1 are same
+ """
+
+ stream = self.create_stream(
+ self.pg2, self.loop0, self.pg_if_packet_sizes)
+ self.pg2.add_stream(stream)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rcvd1 = self.pg0.get_capture()
+ rcvd2 = self.pg1.get_capture()
+
+ self.verify_capture(self.loop0, self.pg2, rcvd1)
+ self.verify_capture(self.loop0, self.pg2, rcvd2)
+
+ self.assertListEqual(rcvd1.res, rcvd2.res)
+
+ def test_ip4_irb_2(self):
+ """ IPv4 IRB test 2
+
+ Test scenario:
+ ip traffic from pg0 and pg1 ends on pg2
+ """
+
+ stream1 = self.create_stream_l2_to_ip(
+ self.pg0, self.loop0, self.pg2, self.pg_if_packet_sizes)
+ stream2 = self.create_stream_l2_to_ip(
+ self.pg1, self.loop0, self.pg2, self.pg_if_packet_sizes)
+ self.pg0.add_stream(stream1)
+ self.pg1.add_stream(stream2)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rcvd = self.pg2.get_capture()
+ self.verify_capture_l2_to_ip(self.pg2, self.loop0, rcvd)
+
+ self.assertEqual(len(stream1) + len(stream2), len(rcvd.res))
+
+
+if __name__ == '__main__':
+ unittest.main(testRunner=VppTestRunner)
from framework import VppTestCase, VppTestRunner
from vpp_sub_interface import VppDot1QSubint
-from util import TestHost
+from util import Host
class TestL2bd(VppTestCase):
hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
packets = []
for j in range(start_nr, end_nr):
- host = TestHost(
+ host = Host(
"00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
"172.17.1%02x.%u" % (pg_if.sw_if_index, j))
packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=host.mac))
from logging import *
from framework import VppTestCase, VppTestRunner
-from util import TestHost
+from util import Host
class TestL2xc(VppTestCase):
self.hosts_by_pg_idx[pg_if.sw_if_index] = []
hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
for j in range(0, count):
- host = TestHost(
+ host = Host(
"00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
"172.17.1%02x.%u" % (pg_if.sw_if_index, j))
hosts.append(host)
-from logging import *
+import socket
-
-class TestHost(object):
- """ Generic test host "connected" to VPP. """
+class Host(object):
+ """ Generic test host "connected" to VPPs interface. """
@property
def mac(self):
""" IPv4 address """
return self._ip4
+ @property
+ def ip4n(self):
+ """ IPv4 address """
+ return socket.inet_pton(socket.AF_INET, self._ip4)
+
@property
def ip6(self):
""" IPv6 address """
from abc import abstractmethod, ABCMeta
import socket
-from logging import info, error
-from scapy.layers.l2 import Ether, ARP
+from logging import info
-from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr
+from util import Host
class VppInterface(object):
@property
def remote_mac(self):
"""MAC-address of the remote interface "connected" to this interface"""
- return self._remote_mac
+ return self._remote_hosts[0].mac
@property
def local_mac(self):
@property
def local_ip4n(self):
"""Local IPv4 address - raw, suitable as API parameter"""
- return self._local_ip4n
+ return socket.inet_pton(socket.AF_INET, self._local_ip4)
@property
def remote_ip4(self):
"""IPv4 address of remote peer "connected" to this interface"""
- return self._remote_ip4
+ return self._remote_hosts[0].ip4
@property
def remote_ip4n(self):
"""IPv4 address of remote peer - raw, suitable as API parameter"""
- return self._remote_ip4n
+ return socket.inet_pton(socket.AF_INET, self.remote_ip4)
@property
def local_ip6(self):
@property
def local_ip6n(self):
"""Local IPv6 address - raw, suitable as API parameter"""
- return self._local_ip6n
+ return socket.inet_pton(socket.AF_INET6, self.local_ip6)
@property
def remote_ip6(self):
"""IPv6 address of remote peer "connected" to this interface"""
- return self._remote_ip6
+ return self._remote_hosts[0].ip6
@property
def remote_ip6n(self):
"""IPv6 address of remote peer - raw, suitable as API parameter"""
- return self._remote_ip6n
+ return socket.inet_pton(socket.AF_INET6, self.remote_ip6)
@property
def name(self):
"""Test case creating this interface"""
return self._test
+ @property
+ def remote_hosts(self):
+ """Remote hosts list"""
+ return self._remote_hosts
+
+ @remote_hosts.setter
+ def remote_hosts(self, value):
+ self._remote_hosts = value
+ #TODO: set hosts_by dicts
+
+ def host_by_mac(self, mac):
+ return self._hosts_by_mac[mac]
+
+ def host_by_ip4(self, ip):
+ return self._hosts_by_ip4[ip]
+
+ def host_by_ip6(self, ip):
+ return self._hosts_by_ip6[ip]
+
+ def generate_remote_hosts(self, count=1):
+ """Generate and add remote hosts for the interface."""
+ self._remote_hosts = []
+ self._hosts_by_mac = {}
+ self._hosts_by_ip4 = {}
+ self._hosts_by_ip6 = {}
+ for i in range(2, count+2): # 0: network address, 1: local vpp address
+ mac = "02:%02x:00:00:ff:%02x" % (self.sw_if_index, i)
+ ip4 = "172.16.%u.%u" % (self.sw_if_index, i)
+ ip6 = "fd01:%04x::%04x" % (self.sw_if_index, i)
+ host = Host(mac, ip4, ip6)
+ self._remote_hosts.append(host)
+ self._hosts_by_mac[mac] = host
+ self._hosts_by_ip4[ip4] = host
+ self._hosts_by_ip6[ip6] = host
+
def post_init_setup(self):
"""Additional setup run after creating an interface object"""
- self._remote_mac = "02:00:00:00:ff:%02x" % self.sw_if_index
+
+ self.generate_remote_hosts()
self._local_ip4 = "172.16.%u.1" % self.sw_if_index
self._local_ip4n = socket.inet_pton(socket.AF_INET, self.local_ip4)
- self._remote_ip4 = "172.16.%u.2" % self.sw_if_index
- self._remote_ip4n = socket.inet_pton(socket.AF_INET, self.remote_ip4)
- self._local_ip6 = "fd01:%u::1" % self.sw_if_index
+ self._local_ip6 = "fd01:%04x::1" % self.sw_if_index
self._local_ip6n = socket.inet_pton(socket.AF_INET6, self.local_ip6)
- self._remote_ip6 = "fd01:%u::2" % self.sw_if_index
- self._remote_ip6n = socket.inet_pton(socket.AF_INET6, self.remote_ip6)
r = self.test.vapi.sw_interface_dump()
for intf in r:
self.test.vapi.sw_interface_add_del_address(
self.sw_if_index, addr, addr_len)
+ def configure_extend_ipv4_mac_binding(self):
+ """Configure neighbor MAC to IPv4 addresses."""
+ for host in self._remote_hosts:
+ macn = host.mac.replace(":", "").decode('hex')
+ ipn = host.ip4n
+ self.test.vapi.ip_neighbor_add_del(self.sw_if_index, macn, ipn)
+
def config_ip6(self):
"""Configure IPv6 address on the VPP interface"""
addr = self._local_ip6n
"""Configure IPv6 RA suppress on the VPP interface"""
self.test.vapi.sw_interface_ra_suppress(self.sw_if_index)
- def create_arp_req(self):
- """Create ARP request applicable for this interface"""
- return (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.remote_mac) /
- ARP(op=ARP.who_has, pdst=self.local_ip4,
- psrc=self.remote_ip4, hwsrc=self.remote_mac))
-
- def create_ndp_req(self):
- return (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.remote_mac) /
- IPv6(src=self.remote_ip6, dst=self.local_ip6) /
- ICMPv6ND_NS(tgt=self.local_ip6) /
- ICMPv6NDOptSrcLLAddr(lladdr=self.remote_mac))
-
- def resolve_arp(self, pg_interface=None):
- """Resolve ARP using provided packet-generator interface
-
- :param pg_interface: interface used to resolve, if None then this
- interface is used
-
- """
- if pg_interface is None:
- pg_interface = self
- info("Sending ARP request for %s on port %s" %
- (self.local_ip4, pg_interface.name))
- arp_req = self.create_arp_req()
- pg_interface.add_stream(arp_req)
- pg_interface.enable_capture()
- self.test.pg_start()
- info(self.test.vapi.cli("show trace"))
- arp_reply = pg_interface.get_capture()
- if arp_reply is None or len(arp_reply) == 0:
- info("No ARP received on port %s" % pg_interface.name)
- return
- arp_reply = arp_reply[0]
- # Make Dot1AD packet content recognizable to scapy
- if arp_reply.type == 0x88a8:
- arp_reply.type = 0x8100
- arp_reply = Ether(str(arp_reply))
- try:
- if arp_reply[ARP].op == ARP.is_at:
- info("VPP %s MAC address is %s " %
- (self.name, arp_reply[ARP].hwsrc))
- self._local_mac = arp_reply[ARP].hwsrc
- else:
- info("No ARP received on port %s" % pg_interface.name)
- except:
- error("Unexpected response to ARP request:")
- error(arp_reply.show())
- raise
-
- def resolve_ndp(self, pg_interface=None):
- """Resolve NDP using provided packet-generator interface
-
- :param pg_interface: interface used to resolve, if None then this
- interface is used
-
- """
- if pg_interface is None:
- pg_interface = self
- info("Sending NDP request for %s on port %s" %
- (self.local_ip6, pg_interface.name))
- ndp_req = self.create_ndp_req()
- pg_interface.add_stream(ndp_req)
- pg_interface.enable_capture()
- self.test.pg_start()
- info(self.test.vapi.cli("show trace"))
- ndp_reply = pg_interface.get_capture()
- if ndp_reply is None or len(ndp_reply) == 0:
- info("No NDP received on port %s" % pg_interface.name)
- return
- ndp_reply = ndp_reply[0]
- # Make Dot1AD packet content recognizable to scapy
- if ndp_reply.type == 0x88a8:
- ndp_reply.type = 0x8100
- ndp_reply = Ether(str(ndp_reply))
- try:
- ndp_na = ndp_reply[ICMPv6ND_NA]
- opt = ndp_na[ICMPv6NDOptDstLLAddr]
- info("VPP %s MAC address is %s " %
- (self.name, opt.lladdr))
- self._local_mac = opt.lladdr
- except:
- error("Unexpected response to NDP request:")
- error(ndp_reply.show())
- raise
-
def admin_up(self):
""" Put interface ADMIN-UP """
self.test.vapi.sw_interface_set_flags(self.sw_if_index, admin_up_down=1)
--- /dev/null
+
+from vpp_interface import VppInterface
+
+
+class VppLoInterface(VppInterface):
+ """
+ VPP loopback interface
+ """
+
+ def __init__(self, test, lo_index):
+ """ Create VPP loopback interface """
+ self._lo_index = lo_index
+ self._test = test
+ r = self.test.vapi.create_loopback()
+ self._sw_if_index = r.sw_if_index
+ self.post_init_setup()
"""
return self.api(vpp_papi.create_vlan_subif, (sw_if_index, vlan))
+ def create_loopback(self, mac=''):
+ """
+
+ :param mac: (Optional)
+ """
+ return self.api(vpp_papi.create_loopback, (mac,))
+
def ip_add_del_route(
self,
dst_address,
dst_address_length,
dst_address,
next_hop_address))
+
+ def ip_neighbor_add_del(self,
+ sw_if_index,
+ mac_address,
+ dst_address,
+ vrf_id=0,
+ is_add=1,
+ is_ipv6=0,
+ is_static=0,
+ ):
+ """ Add neighbor MAC to IPv4 or IPv6 address.
+
+ :param sw_if_index:
+ :param mac_address:
+ :param dst_address:
+ :param vrf_id: (Default value = 0)
+ :param is_add: (Default value = 1)
+ :param is_ipv6: (Default value = 0)
+ :param is_static: (Default value = 0)
+ """
+
+ return self.api(
+ vpp_papi.ip_neighbor_add_del,
+ (vrf_id,
+ sw_if_index,
+ is_add,
+ is_ipv6,
+ is_static,
+ mac_address,
+ dst_address
+ )
+ )
import os
import time
-from logging import error
+from logging import error, info
from scapy.utils import wrpcap, rdpcap
from vpp_interface import VppInterface
+from scapy.layers.l2 import Ether, ARP
+from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr
+
class VppPGInterface(VppInterface):
"""
" packets arrived" % self.out_path)
return []
return output
+
+ def create_arp_req(self):
+ """Create ARP request applicable for this interface"""
+ return (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.remote_mac) /
+ ARP(op=ARP.who_has, pdst=self.local_ip4,
+ psrc=self.remote_ip4, hwsrc=self.remote_mac))
+
+ def create_ndp_req(self):
+ """Create NDP - NS applicable for this interface"""
+ return (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.remote_mac) /
+ IPv6(src=self.remote_ip6, dst=self.local_ip6) /
+ ICMPv6ND_NS(tgt=self.local_ip6) /
+ ICMPv6NDOptSrcLLAddr(lladdr=self.remote_mac))
+
+ def resolve_arp(self, pg_interface=None):
+ """Resolve ARP using provided packet-generator interface
+
+ :param pg_interface: interface used to resolve, if None then this
+ interface is used
+
+ """
+ if pg_interface is None:
+ pg_interface = self
+ info("Sending ARP request for %s on port %s" %
+ (self.local_ip4, pg_interface.name))
+ arp_req = self.create_arp_req()
+ pg_interface.add_stream(arp_req)
+ pg_interface.enable_capture()
+ self.test.pg_start()
+ info(self.test.vapi.cli("show trace"))
+ arp_reply = pg_interface.get_capture()
+ if arp_reply is None or len(arp_reply) == 0:
+ info("No ARP received on port %s" % pg_interface.name)
+ return
+ arp_reply = arp_reply[0]
+ # Make Dot1AD packet content recognizable to scapy
+ if arp_reply.type == 0x88a8:
+ arp_reply.type = 0x8100
+ arp_reply = Ether(str(arp_reply))
+ try:
+ if arp_reply[ARP].op == ARP.is_at:
+ info("VPP %s MAC address is %s " %
+ (self.name, arp_reply[ARP].hwsrc))
+ self._local_mac = arp_reply[ARP].hwsrc
+ else:
+ info("No ARP received on port %s" % pg_interface.name)
+ except:
+ error("Unexpected response to ARP request:")
+ error(arp_reply.show())
+ raise
+
+ def resolve_ndp(self, pg_interface=None):
+ """Resolve NDP using provided packet-generator interface
+
+ :param pg_interface: interface used to resolve, if None then this
+ interface is used
+
+ """
+ if pg_interface is None:
+ pg_interface = self
+ info("Sending NDP request for %s on port %s" %
+ (self.local_ip6, pg_interface.name))
+ ndp_req = self.create_ndp_req()
+ pg_interface.add_stream(ndp_req)
+ pg_interface.enable_capture()
+ self.test.pg_start()
+ info(self.test.vapi.cli("show trace"))
+ ndp_reply = pg_interface.get_capture()
+ if ndp_reply is None or len(ndp_reply) == 0:
+ info("No NDP received on port %s" % pg_interface.name)
+ return
+ ndp_reply = ndp_reply[0]
+ # Make Dot1AD packet content recognizable to scapy
+ if ndp_reply.type == 0x88a8:
+ ndp_reply.type = 0x8100
+ ndp_reply = Ether(str(ndp_reply))
+ try:
+ ndp_na = ndp_reply[ICMPv6ND_NA]
+ opt = ndp_na[ICMPv6NDOptDstLLAddr]
+ info("VPP %s MAC address is %s " %
+ (self.name, opt.lladdr))
+ self._local_mac = opt.lladdr
+ except:
+ error("Unexpected response to NDP request:")
+ error(ndp_reply.show())
+ raise
from scapy.layers.l2 import Ether, Dot1Q
from abc import abstractmethod, ABCMeta
from vpp_interface import VppInterface
+from vpp_pg_interface import VppPGInterface
-class VppSubInterface(VppInterface):
+class VppSubInterface(VppPGInterface):
__metaclass__ = ABCMeta
@property
self._vlan = vlan
r = self.test.vapi.create_vlan_subif(parent.sw_if_index, self.vlan)
self._sw_if_index = r.sw_if_index
- self.post_init_setup()
+ VppInterface.post_init_setup(self)
def create_arp_req(self):
- packet = VppInterface.create_arp_req(self)
+ packet = VppPGInterface.create_arp_req(self)
return self.add_dot1_layer(packet)
def create_ndp_req(self):
- packet = VppInterface.create_ndp_req(self)
+ packet = VppPGInterface.create_ndp_req(self)
return self.add_dot1_layer(packet)
def add_dot1_layer(self, packet):
two_tags=1,
exact_match=1)
self._sw_if_index = r.sw_if_index
- self.post_init_setup()
+ VppInterface.post_init_setup(self)
def create_arp_req(self):
- packet = VppInterface.create_arp_req(self)
+ packet = VppPGInterface.create_arp_req(self)
return self.add_dot1_layer(packet)
def create_ndp_req(self):
- packet = VppInterface.create_ndp_req(self)
+ packet = VppPGInterface.create_ndp_req(self)
return self.add_dot1_layer(packet)
def add_dot1_layer(self, packet):
Code Review / vpp.git / commitdiff