[dts] [PATCH 1/2] tests/vlan_ehancement: add script

[Lijuan Tu <lijuanx.a.tu@intel.com>]

ehance single vlan test,
1, vlan id in [1, random, MAX_VLAN]
2, packet with no tag, matched tag, not-matched tag
3, several times operation on rx_vlan add/rm

Signed-off-by: Lijuan Tu <lijuanx.a.tu@intel.com>
---
 tests/TestSuite_vlan_ehancement.py | 333 +++++++++++++++++++++++++++++++++++++
 1 file changed, 333 insertions(+)
 create mode 100644 tests/TestSuite_vlan_ehancement.py

diff --git a/tests/TestSuite_vlan_ehancement.py b/tests/TestSuite_vlan_ehancement.py
new file mode 100644
index 0000000..516f34b
--- /dev/null
+++ b/tests/TestSuite_vlan_ehancement.py
@@ -0,0 +1,333 @@
+# BSD LICENSE
+#
+# Copyright(c) 2010-2015 Intel Corporation. All rights reserved.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+#   * Redistributions of source code must retain the above copyright
+#     notice, this list of conditions and the following disclaimer.
+#   * Redistributions in binary form must reproduce the above copyright
+#     notice, this list of conditions and the following disclaimer in
+#     the documentation and/or other materials provided with the
+#     distribution.
+#   * Neither the name of Intel Corporation nor the names of its
+#     contributors may be used to endorse or promote products derived
+#     from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+
+"""
+DPDK Test suite.
+
+Test the support of VLAN Offload Features by Poll Mode Drivers.
+
+"""
+
+import utils
+import time
+
+
+from test_case import TestCase
+from pmd_output import PmdOutput
+from packet import Packet, sniff_packets, load_sniff_packets
+
+import random
+MAX_VLAN = 4095
+
+
+class TestVlan(TestCase):
+
+    def set_up_all(self):
+        """
+        Run at the start of each test suite.
+        Vlan Prerequistites
+        """
+        global dutRxPortId
+        global dutTxPortId
+
+        # Based on h/w type, choose how many ports to use
+        ports = self.dut.get_ports()
+
+        # Verify that enough ports are available
+        self.verify(len(ports) >= 1, "Insufficient ports")
+
+        valports = [_ for _ in ports if self.tester.get_local_port(_) != -1]
+        dutRxPortId = valports[0]
+        dutTxPortId = valports[0]
+        portMask = utils.create_mask(valports[:1])
+
+        self.pmdout = PmdOutput(self.dut)
+        self.pmdout.start_testpmd(
+            "Default", "--portmask=%s --port-topology=loop" % portMask)
+
+        self.dut.send_expect("set verbose 1", "testpmd> ")
+        out = self.dut.send_expect("set fwd mac", "testpmd> ")
+        self.verify(
+            'Set mac packet forwarding mode' in out, "set fwd mac error")
+        if self.nic in ["fortville_eagle", "fortville_spirit", "fortville_spirit_single", "fortpark_TLV"]:
+            self.dut.send_expect(
+                "vlan set filter on %s" % dutRxPortId, "testpmd> ")
+            self.dut.send_expect("set promisc all off",  "testpmd> ")
+
+        self.dut.send_expect(
+            "vlan set strip off %s" % dutRxPortId, "testpmd> ")
+
+        if self.kdriver == "fm10k":
+            netobj = self.dut.ports_info[dutRxPortId]['port']
+            netobj.add_vlan(vlan_id=self.vlan)
+
+    def get_tcpdump_package(self):
+        pkts = load_sniff_packets(self.inst)
+        if len(pkts) == 0:
+            return False
+        vlans = []
+        for packet in pkts:
+            vlan = packet.strip_element_vlan("vlan")
+            vlans.append(vlan)
+        return vlans
+
+    def vlan_send_packet(self, vid, num=1):
+        """
+        Send $num of packet to portid, if vid is -1, it means send pakcage not include vlan id.
+        """
+        # The package stream : testTxPort->dutRxPort->dutTxport->testRxPort
+        port = self.tester.get_local_port(dutRxPortId)
+        self.txItf = self.tester.get_interface(port)
+        self.smac = self.tester.get_mac(port)
+
+        port = self.tester.get_local_port(dutTxPortId)
+        self.rxItf = self.tester.get_interface(port)
+
+        # the package dect mac must is dut tx port id when the port promisc is
+        # off
+        self.dmac = self.dut.get_mac_address(dutRxPortId)
+
+        self.inst = sniff_packets(self.rxItf)
+        # FIXME  send a burst with only num packet
+        if vid == -1:
+            pkt = Packet(pkt_type='UDP')
+            pkt.config_layer('ether', {'dst': self.dmac, 'src': self.smac})
+        else:
+            pkt = Packet(pkt_type='VLAN_UDP')
+            pkt.config_layer('ether', {'dst': self.dmac, 'src': self.smac})
+            pkt.config_layer('vlan', {'vlan': vid})
+
+        pkt.send_pkt(tx_port=self.txItf)
+
+    def verify_vlan_packets(self, vid=-1, result=True, drop=False):
+        out = self.get_tcpdump_package()
+        if drop == True:
+            self.verify(out == False, "Packet drop error!")
+        elif result == True:
+            self.verify(vid in out, "Wrong vlan:" + str(out))
+        else:
+            self.verify(vid not in out, "Wrong vlan:" + str(out))
+
+    def set_up(self):
+        """
+        Run before each test case.
+        """
+        pass
+
+    def test_vlan_tag(self):
+        """
+        enable filter on and set rx_vlan
+        enable filter on and rx_vlan rm
+        """
+
+        if self.kdriver == "fm10k":
+            print utils.RED("fm10k not support this case\n")
+            return
+        random_vlan = random.randint(1, MAX_VLAN - 1)
+        self.dut.send_expect("vlan set filter on %s" %
+                             dutRxPortId, "testpmd> ", 20)
+        rx_vlans = [1, random_vlan, MAX_VLAN]
+        self.dut.send_expect(
+            "vlan set strip off  %s" % dutRxPortId, "testpmd> ")
+        self.dut.send_expect("start", "testpmd> ", 120)
+        for rx_vlan in rx_vlans:
+            wrong_vlan = rx_vlan % MAX_VLAN + 1
+            self.dut.send_expect("rx_vlan add 0x%x %s" %
+                                 (rx_vlan, dutRxPortId), "testpmd> ")
+            # 1, Send packets with matched tag ,Receive Packet with no change
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(rx_vlan)
+            # 2, Send packets with no tag ,Receive Packet with no change
+            self.vlan_send_packet(-1)
+            self.verify_vlan_packets(rx_vlan, result=False)
+            # 3, Send packets with not matched tag, Packet droped
+            self.vlan_send_packet(wrong_vlan)
+            self.verify_vlan_packets(wrong_vlan, drop=True)
+
+            self.dut.send_expect("rx_vlan rm 0x%x %d" %
+                                 (rx_vlan, dutRxPortId), "testpmd> ", 30)
+            # 1, Send packets with matched tag, Packet droped
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(drop=True)
+            # 2, Send packets with no tag ,Receive Packet with no change
+            self.vlan_send_packet(-1)
+            self.verify_vlan_packets(rx_vlan, result=False)
+            # 3, Send packets with not matched tag, Packet droped
+            self.vlan_send_packet(wrong_vlan)
+            self.verify_vlan_packets(drop=True)
+
+    def test_vlan_filter_off(self):
+        """
+        disable vlan filter
+        """
+        random_vlan = random.randint(1, MAX_VLAN - 1)
+        self.dut.send_expect("vlan set filter off %s" %
+                             dutRxPortId, "testpmd> ", 20)
+        rx_vlans = [1, random_vlan, MAX_VLAN]
+        self.dut.send_expect("start", "testpmd> ", 120)
+        for rx_vlan in rx_vlans:
+            # 1, Send packets with matched tag ,Receive Packet with no change
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(rx_vlan)
+            # 2, Send packets with no tag ,Receive Packet with no change
+            self.vlan_send_packet(-1)
+            self.verify_vlan_packets(rx_vlan, result=False)
+            # 3, Send packets with not matched tag, Receive Packet with no
+            # change
+            wrong_vlan = rx_vlan % MAX_VLAN + 1
+            self.vlan_send_packet(wrong_vlan)
+            self.verify_vlan_packets(wrong_vlan)
+
+    def test_rx_vlan_add_rm(self):
+        """
+        rx_vlan add/rm
+        """
+        random_vlan = random.randint(1, MAX_VLAN - 1)
+        self.dut.send_expect(
+            "vlan set filter on %s" % dutRxPortId, "testpmd> ", 20)
+        self.dut.send_expect("start", "testpmd> ", 120)
+        rx_vlans = [1, random_vlan, MAX_VLAN]
+        for i in range(5):
+            vlan = (random_vlan + i) % 4096 + (random_vlan + i) / 4096
+            rx_vlans.append(vlan)
+
+        for rx_vlan in rx_vlans:
+            self.dut.send_expect("rx_vlan add 0x%x %s" %
+                                 (rx_vlan, dutRxPortId), "testpmd> ")
+        # 1, Send packets with vid, vid+1, vid+2, vid+3 ,
+        # check packet can be received no change
+        for rx_vlan in rx_vlans:
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(rx_vlan)
+
+        for rx_vlan in rx_vlans:
+            self.dut.send_expect("rx_vlan rm 0x%x %d" %
+                                 (rx_vlan, dutRxPortId), "testpmd> ", 30)
+        # 2, Send packets with vid, vid+1, vid+2, vid+3 ,
+        # check packet dropped.
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(rx_vlan, drop=True)
+
+    def test_vlan_strip_config_on(self):
+        """
+        Test Case: vlan strip on
+        """
+        self.dut.send_expect("vlan set filter off %s" %
+                             dutRxPortId, "testpmd> ", 20)
+        self.dut.send_expect(
+            "vlan set strip on %s" % dutRxPortId, "testpmd> ", 20)
+        self.dut.send_expect("start", "testpmd> ", 120)
+        random_vlan = random.randint(1, MAX_VLAN - 1)
+        rx_vlans = [1, random_vlan, MAX_VLAN]
+        for rx_vlan in rx_vlans:
+            # 1, Send packets with matched tag ,Receive Packet with VLAN be
+            # stripped
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(rx_vlan, result=False)
+            # 2, Send packets with no tag ,Receive Packet with no change
+            self.vlan_send_packet(-1)
+            self.verify_vlan_packets(rx_vlan, result=False)
+            # 3, Send packets with not matched tag, Receive Packet with VLAN be
+            # stripped
+            wrong_vlan = rx_vlan % MAX_VLAN + 1
+            self.vlan_send_packet(wrong_vlan)
+            self.verify_vlan_packets(wrong_vlan, result=False)
+
+    def test_vlan_strip_config_off(self):
+
+        if self.kdriver == "fm10k":
+            print utils.RED("fm10k not support this case\n")
+            return
+        self.dut.send_expect("vlan set filter off %s" %
+                             dutRxPortId, "testpmd> ", 20)
+        self.dut.send_expect("vlan set strip off %s" %
+                             dutRxPortId, "testpmd> ", 20)
+        self.dut.send_expect("start", "testpmd> ", 120)
+        random_vlan = random.randint(1, MAX_VLAN - 1)
+        rx_vlans = [1, random_vlan, MAX_VLAN]
+        for rx_vlan in rx_vlans:
+            # 1, Send packets with matched tag ,Receive Packet with no change
+            self.vlan_send_packet(rx_vlan)
+            self.verify_vlan_packets(rx_vlan)
+            # 2, Send packets with no tag ,Receive Packet with no change
+            self.vlan_send_packet(-1)
+            self.verify_vlan_packets(rx_vlan, result=False)
+            # 3, Send packets with not matched tag, Receive Packet with no
+            # change
+            wrong_vlan = rx_vlan % MAX_VLAN + 1
+            self.vlan_send_packet(wrong_vlan)
+            self.verify_vlan_packets(wrong_vlan)
+
+    def test_vlan_enable_vlan_insertion(self):
+        """
+        Enable VLAN header insertion in transmitted packets
+        """
+        if self.kdriver == "fm10k":
+            netobj = self.dut.ports_info[dutTxPortId]['port']
+            netobj.add_vlan(vlan_id=self.vlan)
+            netobj.add_txvlan(vlan_id=self.vlan)
+
+        self.dut.send_expect("start", "testpmd> ")
+
+        random_vlan = random.randint(1, MAX_VLAN - 1)
+        tx_vlans = [1, random_vlan, MAX_VLAN]
+        for tx_vlan in tx_vlans:
+            self.dut.send_expect("tx_vlan set %s %d" %
+                                 (dutTxPortId, tx_vlan), "testpmd> ")
+            # Send packets with no tag ,Receive Packet with matched tag.
+            self.vlan_send_packet(-1)
+            self.verify_vlan_packets(tx_vlan)
+
+            self.dut.send_expect(
+                "tx_vlan reset %s" % dutTxPortId, "testpmd> ", 30)
+
+        if self.kdriver == "fm10k":
+            netobj = self.dut.ports_info[dutTxPortId]['port']
+            # not delete vlan for self.vlan will used later
+            netobj.delete_txvlan(vlan_id=self.vlan)
+
+    def tear_down(self):
+        """
+        Run after each test case.
+        """
+        self.dut.send_expect("stop", "testpmd> ")
+        pass
+
+    def tear_down_all(self):
+        """
+        Run after each test suite.
+        """
+        self.dut.kill_all()
+        if self.kdriver == "fm10k":
+            netobj = self.dut.ports_info[dutRxPortId]['port']
+            netobj.delete_txvlan(vlan_id=self.vlan)
+            netobj.delete_vlan(vlan_id=self.vlan)
-- 
1.9.3