test/test_l2xc.py

#!/usr/bin/env python
## @file test_l2xc.py
#  Module to provide L2 cross-connect test case.
#
#  The module provides a set of tools for L2 cross-connect tests.

import logging
logging.getLogger("scapy.runtime").setLevel(logging.ERROR)

import unittest
import random
from framework import VppTestCase, VppTestRunner
from scapy.layers.l2 import Ether, Raw
from scapy.layers.inet import IP, UDP


## Subclass of the VppTestCase class.
#
#  This subclass is a class for L2 cross-connect test cases. It provides methods
#  to create interfaces, configuring L2 cross-connects, creating and verifying
#  packet streams.
class TestL2xc(VppTestCase):
    """ L2XC Test Case """

    # Test variables
    interf_nr = 4           # Number of interfaces
    hosts_nr = 10           # Number of hosts
    pkts_per_burst = 257    # Number of packets per burst

    ## Class method to start the test case.
    #  Overrides setUpClass method in VppTestCase class.
    #  There is used try..except statement to ensure that the tear down of
    #  the class will be executed even if any exception is raised.
    #  @param cls The class pointer.
    @classmethod
    def setUpClass(cls):
        super(TestL2xc, cls).setUpClass()

        try:
            ## Create interfaces
            cls.interfaces = range(TestL2xc.interf_nr)
            cls.create_interfaces(cls.interfaces)

            ## Create bi-directional cross-connects between pg0 and pg1
            cls.api("sw_interface_set_l2_xconnect rx pg0 tx pg1 enable")
            cls.api("sw_interface_set_l2_xconnect rx pg1 tx pg0 enable")

            ## Create bi-directional cross-connects between pg2 and pg3
            cls.api("sw_interface_set_l2_xconnect rx pg2 tx pg3 enable")
            cls.api("sw_interface_set_l2_xconnect rx pg3 tx pg2 enable")

            cls.cli(0, "show l2patch")

            ## Create host MAC and IPv4 lists
            cls.create_host_lists(TestL2xc.hosts_nr)

        except Exception as e:
            cls.tearDownClass()
            raise e

    ## Method to define tear down VPP actions of the test case.
    #  Overrides tearDown method in VppTestCase class.
    #  @param self The object pointer.
    def tearDown(self):
        self.cli(2, "show int")
        self.cli(2, "show trace")
        self.cli(2, "show hardware")
        self.cli(2, "show l2patch")
        self.cli(2, "show error")
        self.cli(2, "show run")

    ## Class method to create required number of MAC and IPv4 addresses.
    #  Create required number of host MAC addresses and distribute them among
    #  interfaces. Create host IPv4 address for every host MAC address too.
    #  @param cls The class pointer.
    #  @param count Integer variable to store the number of MAC addresses to be
    #  created.
    @classmethod
    def create_host_lists(cls, count):
        for i in cls.interfaces:
            cls.MY_MACS[i] = []
            cls.MY_IP4S[i] = []
            for j in range(0, count):
                cls.MY_MACS[i].append("00:00:00:ff:%02x:%02x" % (i, j))
                cls.MY_IP4S[i].append("172.17.1%02x.%u" % (i, j))
        ## @var MY_MACS
        #  Dictionary variable to store list of MAC addresses per interface.
        ## @var MY_IP4S
        #  Dictionary variable to store list of IPv4 addresses per interface.

    ## Method to create packet stream for the packet generator interface.
    #  Create input packet stream for the given packet generator interface with
    #  packets of different length targeted for all other created packet
    #  generator interfaces.
    #  @param self The object pointer.
    #  @param pg_id Integer variable to store the index of the interface to
    #  create the input packet stream.
    #  @return pkts List variable to store created input stream of packets.
    def create_stream(self, pg_id):
        # TODO: use variables to create lists based on interface number
        pg_targets = [None] * 4
        pg_targets[0] = [1]
        pg_targets[1] = [0]
        pg_targets[2] = [3]
        pg_targets[3] = [2]
        pkts = []
        for i in range(0, TestL2xc.pkts_per_burst):
            target_pg_id = pg_targets[pg_id][0]
            target_host_id = random.randrange(len(self.MY_MACS[target_pg_id]))
            source_host_id = random.randrange(len(self.MY_MACS[pg_id]))
            pkt_info = self.create_packet_info(pg_id, target_pg_id)
            payload = self.info_to_payload(pkt_info)
            p = (Ether(dst=self.MY_MACS[target_pg_id][target_host_id],
                       src=self.MY_MACS[pg_id][source_host_id]) /
                 IP(src=self.MY_IP4S[pg_id][source_host_id],
                    dst=self.MY_IP4S[target_pg_id][target_host_id]) /
                 UDP(sport=1234, dport=1234) /
                 Raw(payload))
            pkt_info.data = p.copy()
            packet_sizes = [64, 512, 1518, 9018]
            size = packet_sizes[(i / 2) % len(packet_sizes)]
            self.extend_packet(p, size)
            pkts.append(p)
        return pkts
        ## @var pg_targets
        #  List variable to store list of indexes of target packet generator
        #  interfaces for every source packet generator interface.
        ## @var target_pg_id
        #  Integer variable to store the index of the random target packet
        #  generator interfaces.
        ## @var target_host_id
        #  Integer variable to store the index of the randomly chosen
        #  destination host MAC/IPv4 address.
        ## @var source_host_id
        #  Integer variable to store the index of the randomly chosen source
        #  host MAC/IPv4 address.
        ## @var pkt_info
        #  Object variable to store the information about the generated packet.
        ## @var payload
        #  String variable to store the payload of the packet to be generated.
        ## @var p
        #  Object variable to store the generated packet.
        ## @var packet_sizes
        #  List variable to store required packet sizes.
        ## @var size
        #  List variable to store required packet sizes.

    ## Method to verify packet stream received on the packet generator interface.
    #  Verify packet-by-packet the output stream captured on a given packet
    #  generator (pg) interface using following packet payload data - order of
    #  packet in the stream, index of the source and destination pg interface,
    #  src and dst host IPv4 addresses and src port and dst port values of UDP
    #  layer.
    #  @param self The object pointer.
    #  @param o Integer variable to store the index of the interface to
    #  verify the output packet stream.
    #  @param capture List variable to store the captured output packet stream.
    def verify_capture(self, o, capture):
        last_info = {}
        for i in self.interfaces:
            last_info[i] = None
        for packet in capture:
            try:
                ip = packet[IP]
                udp = packet[UDP]
                payload_info = self.payload_to_info(str(packet[Raw]))
                self.assertEqual(payload_info.dst, o)
                self.log("Got packet on port %u: src=%u (id=%u)"
                         % (o, payload_info.src, payload_info.index), 2)
                next_info = self.get_next_packet_info_for_interface2(
                    payload_info.src, payload_info.dst,
                    last_info[payload_info.src])
                last_info[payload_info.src] = next_info
                self.assertTrue(next_info is not None)
                self.assertEqual(payload_info.index, next_info.index)
                # Check standard fields
                self.assertEqual(ip.src, next_info.data[IP].src)
                self.assertEqual(ip.dst, next_info.data[IP].dst)
                self.assertEqual(udp.sport, next_info.data[UDP].sport)
                self.assertEqual(udp.dport, next_info.data[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))
        ## @var last_info
        #  Dictionary variable to store verified packets per packet generator
        #  interface.
        ## @var ip
        #  Object variable to store the IP layer of the packet.
        ## @var udp
        #  Object variable to store the UDP layer of the packet.
        ## @var payload_info
        #  Object variable to store required information about the packet.
        ## @var next_info
        #  Object variable to store information about next packet.
        ## @var remaining_packet
        #  Object variable to store information about remaining packet.

    ## Method defining L2 cross-connect test case.
    #  Contains steps of the test case.
    #  @param self The object pointer.
    def test_l2xc(self):
        """ L2XC test

        Test scenario:
        1.config
            2 pairs of 2 interfaces, l2xconnected

        2.sending l2 eth packets between 4 interfaces
            64B, 512B, 1518B, 9018B (ether_size)
            burst of packets per interface
        """

        ## Create incoming packet streams for packet-generator interfaces
        for i in self.interfaces:
            pkts = self.create_stream(i)
            self.pg_add_stream(i, pkts)

        ## Enable packet capturing and start packet sending
        self.pg_enable_capture(self.interfaces)
        self.pg_start()

        ## Verify outgoing packet streams per packet-generator interface
        for i in self.interfaces:
            out = self.pg_get_capture(i)
            self.log("Verifying capture %u" % i)
            self.verify_capture(i, out)
        ## @var pkts
        #  List variable to store created input stream of packets for the packet
        #  generator interface.
        ## @var out
        #  List variable to store captured output stream of packets for
        #  the packet generator interface.


if __name__ == '__main__':
    unittest.main(testRunner = VppTestRunner)