import os
import time
-from scapy.utils import wrpcap, rdpcap
+from scapy.utils import wrpcap, rdpcap, PcapReader
from vpp_interface import VppInterface
from scapy.layers.l2 import Ether, ARP
-from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA, \
+from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA,\
ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr
from util import ppp
pass
# FIXME this should be an API, but no such exists atm
self.test.vapi.cli(self.capture_cli)
+ self._pcap_reader = None
def add_stream(self, pkts):
"""
return []
return output
+ def wait_for_packet(self, timeout):
+ """
+ Wait for next packet captured with a timeout
+
+ :param timeout: How long to wait for the packet
+
+ :returns: Captured packet if no packet arrived within timeout
+ :raises Exception: if no packet arrives within timeout
+ """
+ limit = time.time() + timeout
+ if self._pcap_reader is None:
+ self.test.logger.debug("Waiting for the capture file to appear")
+ while time.time() < limit:
+ if os.path.isfile(self.out_path):
+ break
+ time.sleep(0) # yield
+ if os.path.isfile(self.out_path):
+ self.test.logger.debug("Capture file appeared after %fs" %
+ (time.time() - (limit - timeout)))
+ self._pcap_reader = PcapReader(self.out_path)
+ else:
+ self.test.logger.debug("Timeout - capture file still nowhere")
+ raise Exception("Packet didn't arrive within timeout")
+
+ self.test.logger.debug("Waiting for packet")
+ while time.time() < limit:
+ p = self._pcap_reader.recv()
+ if p is not None:
+ self.test.logger.debug("Packet received after %fs",
+ (time.time() - (limit - timeout)))
+ return p
+ time.sleep(0) # yield
+ self.test.logger.debug("Timeout - no packets received")
+ raise Exception("Packet didn't arrive within timeout")
+
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) /
pg_interface.enable_capture()
self.test.pg_start()
self.test.logger.info(self.test.vapi.cli("show trace"))
- ndp_reply = pg_interface.get_capture()
- if ndp_reply is None or len(ndp_reply) == 0:
+ replies = pg_interface.get_capture()
+ if replies is None or len(replies) == 0:
self.test.logger.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]
- self.test.logger.info("VPP %s MAC address is %s " %
- (self.name, opt.lladdr))
- self._local_mac = opt.lladdr
- except:
- self.test.logger.error(
- ppp("Unexpected response to NDP request:", ndp_reply))
+ # Enabling IPv6 on an interface can generate more than the
+ # ND reply we are looking for (namely MLD). So loop through
+ # the replies to look for want we want.
+ for ndp_reply in replies:
+ # 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]
+ self.test.logger.info("VPP %s MAC address is %s " %
+ (self.name, opt.lladdr))
+ self._local_mac = opt.lladdr
+ except:
+ self.test.logger.info(
+ ppp("Unexpected response to NDP request:", ndp_reply))
+ # if no packets above provided the local MAC, then this failed.
+ if not hasattr(self, '_local_mac'):
raise