from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_RS, \
ICMPv6ND_RA, ICMPv6NDOptMTU, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptPrefixInfo, \
ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, ICMPv6DestUnreach, icmp6types, \
- ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply
+ ICMPv6TimeExceeded, ICMPv6EchoRequest, ICMPv6EchoReply, IPv6ExtHdrHopByHop
from scapy.layers.l2 import Ether, Dot1Q
from scapy.packet import Raw
from scapy.utils import inet_pton, inet_ntop
except NameError:
text_type = str
+NUM_PKTS = 67
+
class TestIPv6ND(VppTestCase):
def validate_ra(self, intf, rx, dst_ip=None):
def setUpClass(cls):
super(TestIPv6, cls).setUpClass()
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv6, cls).tearDownClass()
+
def setUp(self):
"""
Perform test setup before test case.
class TestICMPv6Echo(VppTestCase):
""" ICMPv6 Echo Test Case """
+ @classmethod
+ def setUpClass(cls):
+ super(TestICMPv6Echo, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestICMPv6Echo, cls).tearDownClass()
+
def setUp(self):
super(TestICMPv6Echo, self).setUp()
def setUpClass(cls):
super(TestIPv6RD, cls).setUpClass()
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv6RD, cls).tearDownClass()
+
def setUp(self):
super(TestIPv6RD, self).setUp()
def setUpClass(cls):
super(TestIPv6RDControlPlane, cls).setUpClass()
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPv6RDControlPlane, cls).tearDownClass()
+
def setUp(self):
super(TestIPv6RDControlPlane, self).setUp()
# check FIB still contains the SLAAC address
addresses = set(self.get_interface_addresses(fib, self.pg0))
new_addresses = addresses.difference(initial_addresses)
+
self.assertEqual(len(new_addresses), 1)
prefix = list(new_addresses)[0][:8] + '\0\0\0\0\0\0\0\0'
self.assertEqual(inet_ntop(AF_INET6, prefix), '1::')
class IPv6NDProxyTest(TestIPv6ND):
""" IPv6 ND ProxyTest Case """
+ @classmethod
+ def setUpClass(cls):
+ super(IPv6NDProxyTest, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(IPv6NDProxyTest, cls).tearDownClass()
+
def setUp(self):
super(IPv6NDProxyTest, self).setUp()
class TestIPNull(VppTestCase):
""" IPv6 routes via NULL """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPNull, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPNull, cls).tearDownClass()
+
def setUp(self):
super(TestIPNull, self).setUp()
class TestIPDisabled(VppTestCase):
""" IPv6 disabled """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPDisabled, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPDisabled, cls).tearDownClass()
+
def setUp(self):
super(TestIPDisabled, self).setUp()
class TestIP6LoadBalance(VppTestCase):
""" IPv6 Load-Balancing """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6LoadBalance, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6LoadBalance, cls).tearDownClass()
+
def setUp(self):
super(TestIP6LoadBalance, self).setUp()
src_ip_pkts = []
src_mpls_pkts = []
- for ii in range(65):
+ for ii in range(NUM_PKTS):
port_ip_hdr = (
IPv6(dst="3000::1", src="3000:1::1") /
inet6.UDP(sport=1234, dport=1234 + ii) /
class TestIP6Punt(VppTestCase):
""" IPv6 Punt Police/Redirect """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Punt, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Punt, cls).tearDownClass()
+
def setUp(self):
super(TestIP6Punt, self).setUp()
class TestIPDeag(VppTestCase):
""" IPv6 Deaggregate Routes """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIPDeag, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIPDeag, cls).tearDownClass()
+
def setUp(self):
super(TestIPDeag, self).setUp()
class TestIP6Input(VppTestCase):
""" IPv6 Input Exception Test Cases """
+ @classmethod
+ def setUpClass(cls):
+ super(TestIP6Input, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestIP6Input, cls).tearDownClass()
+
def setUp(self):
super(TestIP6Input, self).setUp()
inet6.UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- rx = self.send_and_expect(self.pg0, p_version * 65, self.pg0)
+ rx = self.send_and_expect(self.pg0, p_version * NUM_PKTS, self.pg0)
rx = rx[0]
icmp = rx[ICMPv6TimeExceeded]
l4 /
Raw('\xa5' * 100))
- self.send_and_assert_no_replies(self.pg0, p_version * 65,
+ self.send_and_assert_no_replies(self.pg0, p_version * NUM_PKTS,
remark=msg or "",
timeout=timeout)
+ def test_hop_by_hop(self):
+ """ Hop-by-hop header test """
+
+ p = (Ether(src=self.pg0.remote_mac,
+ dst=self.pg0.local_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
+ IPv6ExtHdrHopByHop() /
+ inet6.UDP(sport=1234, dport=1234) /
+ Raw('\xa5' * 100))
+
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)