diff options
Diffstat (limited to 'test/test_ip6.py')
| -rw-r--r-- | test/test_ip6.py | 323 |
1 files changed, 104 insertions, 219 deletions
diff --git a/test/test_ip6.py b/test/test_ip6.py index 38808a9e1b7..92bb350d1b9 100644 --- a/test/test_ip6.py +++ b/test/test_ip6.py @@ -1,250 +1,126 @@ #!/usr/bin/env python -import logging -logging.getLogger("scapy.runtime").setLevel(logging.ERROR) - import unittest +import socket +from logging import * + from framework import VppTestCase, VppTestRunner -from util import Util +from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint from scapy.packet import Raw from scapy.layers.l2 import Ether, Dot1Q -from scapy.layers.inet6 import (IPv6, UDP, - ICMPv6ND_NS, ICMPv6NDOptSrcLLAddr, - ICMPv6ND_NA, ICMPv6NDOptDstLLAddr) +from scapy.layers.inet6 import ICMPv6ND_NS, IPv6, UDP -@unittest.skip('Not finished yet.\n') -class TestIPv6(Util, VppTestCase): +class TestIPv6(VppTestCase): """ IPv6 Test Case """ @classmethod def setUpClass(cls): super(TestIPv6, cls).setUpClass() - try: - cls.create_interfaces_and_subinterfaces() - - # configure IPv6 on hardware interfaces - cls.config_ip6(cls.interfaces) - - cls.config_ip6_on_software_interfaces(cls.interfaces) - - # resolve ICMPv6 ND using hardware interfaces - cls.resolve_icmpv6_nd(cls.interfaces) - - # let VPP know MAC addresses of peer (sub)interfaces - # cls.resolve_icmpv6_nd_on_software_interfaces(cls.interfaces) - cls.send_neighbour_advertisement_on_software_interfaces(cls.interfaces) - - # config 2M FIB enries - #cls.config_fib_entries(2000000) - cls.config_fib_entries(1000000) - - except Exception as e: - super(TestIPv6, cls).tearDownClass() - raise - - def tearDown(self): - self.cli(2, "show int") - self.cli(2, "show trace") - self.cli(2, "show hardware") - self.cli(2, "show ip arp") - # self.cli(2, "show ip fib") # 2M entries - self.cli(2, "show error") - self.cli(2, "show run") - - @classmethod - def create_vlan_subif(cls, pg_index, vlan): - cls.api("create_vlan_subif pg%u vlan %u" % (pg_index, vlan)) - - @classmethod - def create_dot1ad_subif(cls, pg_index, sub_id, outer_vlan_id, inner_vlan_id): - cls.api("create_subif pg%u sub_id %u outer_vlan_id %u inner_vlan_id %u dot1ad" - % (pg_index, sub_id, outer_vlan_id, inner_vlan_id)) - - class SoftInt(object): - pass - - class HardInt(SoftInt): - pass - - class Subint(SoftInt): - def __init__(self, sub_id): - self.sub_id = sub_id - - class Dot1QSubint(Subint): - def __init__(self, sub_id, vlan=None): - if vlan is None: - vlan = sub_id - super(TestIPv6.Dot1QSubint, self).__init__(sub_id) - self.vlan = vlan - - class Dot1ADSubint(Subint): - def __init__(self, sub_id, outer_vlan, inner_vlan): - super(TestIPv6.Dot1ADSubint, self).__init__(sub_id) - self.outer_vlan = outer_vlan - self.inner_vlan = inner_vlan - - @classmethod - def create_interfaces_and_subinterfaces(cls): - cls.interfaces = range(3) - - cls.create_interfaces(cls.interfaces) + def setUp(self): + super(TestIPv6, self).setUp() - # Make vpp_api_test see interfaces created using debug CLI (in function create_interfaces) - cls.api("sw_interface_dump") + # create 3 pg interfaces + self.create_pg_interfaces(range(3)) - cls.INT_DETAILS = dict() + # create 2 subinterfaces for p1 and pg2 + self.sub_interfaces = [ + VppDot1QSubint(self, self.pg1, 100), + VppDot1QSubint(self, self.pg2, 200)] + # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400) - cls.INT_DETAILS[0] = cls.HardInt() + # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc. + self.flows = dict() + self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if] + self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if] + self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if] - cls.INT_DETAILS[1] = cls.Dot1QSubint(100) - cls.create_vlan_subif(1, cls.INT_DETAILS[1].vlan) + # packet sizes + self.pg_if_packet_sizes = [64, 512, 1518, 9018] + self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4] - # FIXME: Wrong packet format/wrong layer on output of interface 2 - #self.INT_DETAILS[2] = self.Dot1ADSubint(10, 200, 300) - #self.create_dot1ad_subif(2, self.INT_DETAILS[2].sub_id, self.INT_DETAILS[2].outer_vlan, self.INT_DETAILS[2].inner_vlan) + self.interfaces = list(self.pg_interfaces) + self.interfaces.extend(self.sub_interfaces) - # Use dor1q for now - cls.INT_DETAILS[2] = cls.Dot1QSubint(200) - cls.create_vlan_subif(2, cls.INT_DETAILS[2].vlan) - - for i in cls.interfaces: - det = cls.INT_DETAILS[i] - if isinstance(det, cls.Subint): - cls.api("sw_interface_set_flags pg%u.%u admin-up" % (i, det.sub_id)) - - # IP adresses on subinterfaces - MY_SOFT_IP6S = {} - VPP_SOFT_IP6S = {} - - @classmethod - def config_ip6_on_software_interfaces(cls, args): - for i in args: - cls.MY_SOFT_IP6S[i] = "fd01:%u::2" % i - cls.VPP_SOFT_IP6S[i] = "fd01:%u::1" % i - if isinstance(cls.INT_DETAILS[i], cls.Subint): - interface = "pg%u.%u" % (i, cls.INT_DETAILS[i].sub_id) - else: - interface = "pg%u" % i - cls.api("sw_interface_add_del_address %s %s/32" % (interface, cls.VPP_SOFT_IP6S[i])) - cls.log("My subinterface IPv6 address is %s" % (cls.MY_SOFT_IP6S[i])) - - # let VPP know MAC addresses of peer (sub)interfaces - @classmethod - def resolve_icmpv6_nd_on_software_interfaces(cls, args): - for i in args: - ip = cls.VPP_SOFT_IP6S[i] - cls.log("Sending ICMPv6ND_NS request for %s on port %u" % (ip, i)) - nd_req = (Ether(dst="ff:ff:ff:ff:ff:ff", src=cls.MY_MACS[i]) / - IPv6(src=cls.MY_SOFT_IP6S[i], dst=ip) / - ICMPv6ND_NS(tgt=ip) / - ICMPv6NDOptSrcLLAddr(lladdr=cls.MY_MACS[i])) - cls.pg_add_stream(i, nd_req) - cls.pg_enable_capture([i]) - - cls.cli(2, "trace add pg-input 1") - cls.pg_start() - - # We don't need to read output - - # let VPP know MAC addresses of peer (sub)interfaces - @classmethod - def send_neighbour_advertisement_on_software_interfaces(cls, args): - for i in args: - ip = cls.VPP_SOFT_IP6S[i] - cls.log("Sending ICMPv6ND_NA message for %s on port %u" % (ip, i)) - pkt = (Ether(dst="ff:ff:ff:ff:ff:ff", src=cls.MY_MACS[i]) / - IPv6(src=cls.MY_SOFT_IP6S[i], dst=ip) / - ICMPv6ND_NA(tgt=ip, R=0, S=0) / - ICMPv6NDOptDstLLAddr(lladdr=cls.MY_MACS[i])) - cls.pg_add_stream(i, pkt) - cls.pg_enable_capture([i]) - - cls.cli(2, "trace add pg-input 1") - cls.pg_start() - - @classmethod - def config_fib_entries(cls, count): - n_int = len(cls.interfaces) - for i in cls.interfaces: - cls.api("ip_add_del_route fd02::1/128 via %s count %u" % (cls.VPP_SOFT_IP6S[i], count / n_int)) - - @classmethod - def add_dot1_layers(cls, i, packet): - assert(type(packet) is Ether) - payload = packet.payload - det = cls.INT_DETAILS[i] - if isinstance(det, cls.Dot1QSubint): - packet.remove_payload() - packet.add_payload(Dot1Q(vlan=det.sub_id) / payload) - elif isinstance(det, cls.Dot1ADSubint): - packet.remove_payload() - packet.add_payload(Dot1Q(vlan=det.outer_vlan) / Dot1Q(vlan=det.inner_vlan) / payload) - packet.type = 0x88A8 + # setup all interfaces + for i in self.interfaces: + i.admin_up() + i.config_ip6() + i.resolve_ndp() - def remove_dot1_layers(self, i, packet): - self.assertEqual(type(packet), Ether) - payload = packet.payload - det = self.INT_DETAILS[i] - if isinstance(det, self.Dot1QSubint): - self.assertEqual(type(payload), Dot1Q) - self.assertEqual(payload.vlan, self.INT_DETAILS[i].vlan) - payload = payload.payload - elif isinstance(det, self.Dot1ADSubint): # TODO: change 88A8 type - self.assertEqual(type(payload), Dot1Q) - self.assertEqual(payload.vlan, self.INT_DETAILS[i].outer_vlan) - payload = payload.payload - self.assertEqual(type(payload), Dot1Q) - self.assertEqual(payload.vlan, self.INT_DETAILS[i].inner_vlan) - payload = payload.payload - packet.remove_payload() - packet.add_payload(payload) + # config 2M FIB enries + self.config_fib_entries(200) - def create_stream(self, pg_id): - pg_targets = [None] * 3 - pg_targets[0] = [1, 2] - pg_targets[1] = [0, 2] - pg_targets[2] = [0, 1] + def tearDown(self): + super(TestIPv6, self).tearDown() + if not self.vpp_dead: + info(self.vapi.cli("show ip6 neighbors")) + # info(self.vapi.cli("show ip6 fib")) # many entries + + def config_fib_entries(self, count): + n_int = len(self.interfaces) + percent = 0 + counter = 0.0 + dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1") + dest_addr_len = 128 + for i in self.interfaces: + next_hop_address = i.local_ip6n + for j in range(count / n_int): + self.vapi.ip_add_del_route( + dest_addr, dest_addr_len, next_hop_address, is_ipv6=1) + counter = counter + 1 + if counter / count * 100 > percent: + info("Configure %d FIB entries .. %d%% done" % + (count, percent)) + percent = percent + 1 + + def create_stream(self, src_if, packet_sizes): pkts = [] for i in range(0, 257): - target_pg_id = pg_targets[pg_id][i % 2] - info = self.create_packet_info(pg_id, target_pg_id) + dst_if = self.flows[src_if][i % 2] + info = self.create_packet_info( + src_if.sw_if_index, dst_if.sw_if_index) payload = self.info_to_payload(info) - p = (Ether(dst=self.VPP_MACS[pg_id], src=self.MY_MACS[pg_id]) / - IPv6(src=self.MY_SOFT_IP6S[pg_id], dst=self.MY_SOFT_IP6S[target_pg_id]) / + p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / + IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) / UDP(sport=1234, dport=1234) / Raw(payload)) info.data = p.copy() - self.add_dot1_layers(pg_id, p) - if not isinstance(self.INT_DETAILS[pg_id], self.Subint): - packet_sizes = [76, 512, 1518, 9018] - else: - packet_sizes = [76, 512, 1518+4, 9018+4] - size = packet_sizes[(i / 2) % len(packet_sizes)] + if isinstance(src_if, VppSubInterface): + p = src_if.add_dot1_layer(p) + size = packet_sizes[(i // 2) % len(packet_sizes)] self.extend_packet(p, size) pkts.append(p) return pkts - def verify_capture(self, o, capture): - last_info = {} + def verify_capture(self, dst_if, capture): + info("Verifying capture on interface %s" % dst_if.name) + last_info = dict() for i in self.interfaces: - last_info[i] = None + last_info[i.sw_if_index] = None + is_sub_if = False + dst_sw_if_index = dst_if.sw_if_index + if hasattr(dst_if, 'parent'): + is_sub_if = True for packet in capture: - self.remove_dot1_layers(o, packet) # Check VLAN tags and Ethernet header + if is_sub_if: + # Check VLAN tags and Ethernet header + packet = dst_if.remove_dot1_layer(packet) self.assertTrue(Dot1Q not in packet) try: ip = packet[IPv6] udp = packet[UDP] payload_info = self.payload_to_info(str(packet[Raw])) packet_index = payload_info.index - src_pg = payload_info.src - dst_pg = payload_info.dst - self.assertEqual(dst_pg, o) - self.log("Got packet on port %u: src=%u (id=%u)" % (o, src_pg, packet_index), 2) - next_info = self.get_next_packet_info_for_interface2(src_pg, dst_pg, last_info[src_pg]) - last_info[src_pg] = next_info + self.assertEqual(payload_info.dst, dst_sw_if_index) + debug("Got packet on port %s: src=%u (id=%u)" % + (dst_if.name, payload_info.src, packet_index)) + next_info = self.get_next_packet_info_for_interface2( + payload_info.src, dst_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 @@ -254,28 +130,37 @@ class TestIPv6(Util, VppTestCase): self.assertEqual(udp.sport, saved_packet[UDP].sport) self.assertEqual(udp.dport, saved_packet[UDP].dport) except: - self.log("Unexpected or invalid packet:") - packet.show() + error("Unexpected or invalid packet:") + error(packet.show()) raise for i in self.interfaces: - remaining_packet = self.get_next_packet_info_for_interface2(i, o, last_info[i]) - self.assertTrue(remaining_packet is None, "Port %u: Packet expected from source %u didn't arrive" % (o, i)) + remaining_packet = self.get_next_packet_info_for_interface2( + i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index]) + self.assertTrue( + remaining_packet is None, + "Interface %s: Packet expected from interface %s didn't arrive" % + (dst_if.name, i.name)) def test_fib(self): """ IPv6 FIB test """ - for i in self.interfaces: - pkts = self.create_stream(i) - self.pg_add_stream(i, pkts) + pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes) + self.pg0.add_stream(pkts) + + for i in self.sub_interfaces: + pkts = self.create_stream(i, self.sub_if_packet_sizes) + i.parent.add_stream(pkts) - self.pg_enable_capture(self.interfaces) + self.pg_enable_capture(self.pg_interfaces) self.pg_start() - for i in self.interfaces: - out = self.pg_get_capture(i) - self.log("Verifying capture %u" % i) - self.verify_capture(i, out) + pkts = self.pg0.get_capture() + self.verify_capture(self.pg0, pkts) + + for i in self.sub_interfaces: + pkts = i.parent.get_capture() + self.verify_capture(i, pkts) if __name__ == '__main__': - unittest.main(testRunner = VppTestRunner) + unittest.main(testRunner=VppTestRunner) |