#!/usr/bin/env python
import unittest
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpMRoute, VppMRoutePath, VppMFibSignal, \
MRouteItfFlags, MRouteEntryFlags, VppIpTable
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, getmacbyip
from scapy.layers.inet6 import IPv6, getmacbyip6
#
# The number of packets sent is set to 91 so that when we replicate more than 3
# times, which we do for some entries, we will generate more than 256 packets
# to the next node in the VLIB graph. Thus we are testing the code's
# correctness handling this over-flow.
# It's also an odd number so we hit any single loops.
#
N_PKTS_IN_STREAM = 91
class TestMFIB(VppTestCase):
""" MFIB Test Case """
def setUp(self):
super(TestMFIB, self).setUp()
def test_mfib(self):
""" MFIB Unit Tests """
error = self.vapi.cli("test mfib")
if error:
self.logger.critical(error)
self.assertEqual(error.find("Failed"), -1)
class TestIPMcast(VppTestCase):
""" IP Multicast Test Case """
def setUp(self):
super(TestIPMcast, self).setUp()
# create 8 pg interfaces
self.create_pg_interfaces(range(9))
# setup interfaces
for i in self.pg_interfaces[:8]:
i.admin_up()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
# one more in a vrf
tbl4 = VppIpTable(self, 10)
tbl4.add_vpp_config()
self.pg8.set_table_ip4(10)
self.pg8.config_ip4()
tbl6 = VppIpTable(self, 10, is_ip6=1)
tbl6.add_vpp_config()
self.pg8.set_table_ip6(10)
self.pg8.config_ip6()
def tearDown(self):
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
i.admin_down()
self.pg8.set_table_ip4(0)
self.pg8.set_table_ip6(0)
super(TestIPMcast, self).tearDown()
def create_stream_ip4(self, src_if, src_ip, dst_ip, payload_size=0):
pkts = []
# default to small packet sizes
p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
IP(src=src_ip, dst=dst_ip) /
UDP(sport=1234, dport=1234))
if not payload_size:
payload_size = 64 - len(p)
p = p / Raw('\xa5' * payload_size)
for i in range(0, N_PKTS_IN_STREAM):
pkts.append(p)
return pkts
def create_stream_ip6(self, src_if, src_ip, dst_ip):
pkts = []
for i in range(0, N_PKTS_IN_STREAM):
info = self.create_packet_info(src_if, src_if)
payload = self.info_to_payload(info)
p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
IPv6(src=src_ip, dst=dst_ip) /
UDP(sport=1234, dport=1234) /
Raw(payload))
info.data = p.copy()
pkts.append(p)
return pkts
def verify_filter(self, capture, sent):
if not len(capture) == len(sent):
# filter out any IPv6 RAs from the captur
for p in capture:
if (p.haslayer(IPv6)):
capture.remove(p)
return capture
def verify_capture_ip4(self, rx_if, sent, dst_mac=None):
rxd = rx_if.get_capture(len(sent))
try:
capture = self.verify_filter(rxd, sent)
self.assertEqual(len(capture), len(sent))
for i in range(len(capture)):
tx = sent[i]
rx = capture[i]
eth = rx[Ether]
self.assertEqual(eth.type, 0x800)
tx_ip = tx[IP]
rx_ip = rx[IP]
if dst_mac is None:
dst_mac = getmacbyip(rx_ip.dst)
# check the MAC address on the RX'd packet is correctly formed
self.assertEqual(eth.dst, dst_mac)
self.assertEqual(rx_ip.src, tx_ip.src)
self.assertEqual(rx_ip.dst, tx_ip.dst)
# IP processing post pop has decremented the TTL
self.assertEqual(rx_ip.ttl + 1, tx_ip.ttl)
except:
raise
def verify_capture_ip6(self, rx_if, sent):
capture = rx_if.get_capture(len(sent))
self.assertEqual(len(capture), len(sent))
for i in range(len(capture)):
tx = sent[i]
rx = capture[i]
eth = rx[Ether]
self.assertEqual(eth.type, 0x86DD)
tx_ip