diff options
author | 2016-10-03 19:44:57 +0200 | |
---|---|---|
committer | 2016-10-03 19:58:19 +0000 | |
commit | f56b77a0764222cc45a9df572df901067a273356 (patch) | |
tree | 7f11bcde6219608ad2a433655ea58cd11747025a /test/test_ip6.py | |
parent | 6c3ebcc2bfd36a5835a99225ad667e4403293ffb (diff) |
test: new test infrastructure
Change-Id: I73ca19c431743f6b39669c583d9222a6559346ef
Signed-off-by: Jan Gelety <jgelety@cisco.com>
Signed-off-by: Juraj Sloboda <jsloboda@cisco.com>
Signed-off-by: Stefan Kobza <skobza@cisco.com>
Signed-off-by: Matej Klotton <mklotton@cisco.com>
Signed-off-by: Maciek Konstantynowicz <mkonstan@cisco.com>
Signed-off-by: Damjan Marion <damarion@cisco.com>
Diffstat (limited to 'test/test_ip6.py')
-rw-r--r-- | test/test_ip6.py | 281 |
1 files changed, 281 insertions, 0 deletions
diff --git a/test/test_ip6.py b/test/test_ip6.py new file mode 100644 index 00000000..38808a9e --- /dev/null +++ b/test/test_ip6.py @@ -0,0 +1,281 @@ +#!/usr/bin/env python + +import logging +logging.getLogger("scapy.runtime").setLevel(logging.ERROR) + +import unittest +from framework import VppTestCase, VppTestRunner +from util import Util + +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) + + +@unittest.skip('Not finished yet.\n') +class TestIPv6(Util, 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) + + # Make vpp_api_test see interfaces created using debug CLI (in function create_interfaces) + cls.api("sw_interface_dump") + + cls.INT_DETAILS = dict() + + cls.INT_DETAILS[0] = cls.HardInt() + + cls.INT_DETAILS[1] = cls.Dot1QSubint(100) + cls.create_vlan_subif(1, cls.INT_DETAILS[1].vlan) + + # 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) + + # 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 + + 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) + + 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] + 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) + 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]) / + 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)] + self.extend_packet(p, size) + pkts.append(p) + return pkts + + def verify_capture(self, o, capture): + last_info = {} + for i in self.interfaces: + last_info[i] = None + for packet in capture: + self.remove_dot1_layers(o, packet) # Check VLAN tags and Ethernet header + 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.assertTrue(next_info is not None) + self.assertEqual(packet_index, next_info.index) + saved_packet = next_info.data + # Check standard fields + self.assertEqual(ip.src, saved_packet[IPv6].src) + self.assertEqual(ip.dst, saved_packet[IPv6].dst) + 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() + 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)) + + def test_fib(self): + """ IPv6 FIB test """ + + for i in self.interfaces: + pkts = self.create_stream(i) + self.pg_add_stream(i, pkts) + + self.pg_enable_capture(self.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) + + +if __name__ == '__main__': + unittest.main(testRunner = VppTestRunner) |