[dts] [PATCH 5/6] tests vf_vlan: add test suite for VF vlan validation

[Xu, Qian Q]

When doing vf testing, we many need add crc config on or else the packet RX/TX may have problems. 

>port stop all
>port config all crc-strip on
>port start all




Thanks
Qian


-----Original Message-----
From: dts [mailto:dts-bounces at dpdk.org] On Behalf Of Yong Liu
Sent: Monday, January 11, 2016 3:06 PM
To: dts at dpdk.org
Subject: [dts] [PATCH 5/6] tests vf_vlan: add test suite for VF vlan validation

Signed-off-by: Marvin Liu <yong.liu at intel.com>

diff --git a/tests/TestSuite_vf_vlan.py b/tests/TestSuite_vf_vlan.py new file mode 100644 index 0000000..62fda22
--- /dev/null
+++ b/tests/TestSuite_vf_vlan.py
@@ -0,0 +1,362 @@
+# <COPYRIGHT_TAG>
+
+import re
+import time
+
+import dts
+from qemu_kvm import QEMUKvm
+from test_case import TestCase
+from pmd_output import PmdOutput
+from packet import Packet, sniff_packets, load_sniff_packets from 
+settings import get_nic_name import random
+
+VM_CORES_MASK = 'all'
+
+
+class TestVfVlan(TestCase):
+
+    def set_up_all(self):
+
+        self.dut_ports = self.dut.get_ports(self.nic)
+        self.verify(len(self.dut_ports) > 1, "Insufficient ports")
+        self.vm0 = None
+        self.env_done = False
+
+    def set_up(self):
+        self.setup_vm_env()
+
+    def bind_nic_driver(self, ports, driver=""):
+        # modprobe vfio driver
+        if driver == "vfio-pci":
+            for port in ports:
+                netdev = self.dut.ports_info[port]['port']
+                driver = netdev.get_nic_driver()
+                if driver != 'vfio-pci':
+                    netdev.bind_driver(driver='vfio-pci')
+
+        elif driver == "igb_uio":
+            # igb_uio should insmod as default, no need to check
+            for port in ports:
+                netdev = self.dut.ports_info[port]['port']
+                driver = netdev.get_nic_driver()
+                if driver != 'igb_uio':
+                    netdev.bind_driver(driver='igb_uio')
+        else:
+            for port in ports:
+                netdev = self.dut.ports_info[port]['port']
+                driver_now = netdev.get_nic_driver()
+                if driver == "":
+                    driver = netdev.default_driver
+                if driver != driver_now:
+                    netdev.bind_driver(driver=driver)
+
+    def setup_vm_env(self, driver='default'):
+        """
+        Create testing environment with 2VFs generated from 2PFs
+        """
+        if self.env_done:
+            return
+
+        # bind to default driver
+        self.bind_nic_driver(self.dut_ports[:2], driver="")
+
+        self.used_dut_port_0 = self.dut_ports[0]
+        self.host_intf0 = self.dut.ports_info[self.used_dut_port_0]['intf']
+        tester_port = self.tester.get_local_port(self.used_dut_port_0)
+        self.tester_intf0 = self.tester.get_interface(tester_port)
+
+        self.dut.generate_sriov_vfs_by_port(
+            self.used_dut_port_0, 1, driver=driver)
+        self.sriov_vfs_port_0 = self.dut.ports_info[
+            self.used_dut_port_0]['vfs_port']
+        self.vf0_mac = "00:10:00:00:00:00"
+        self.dut.send_expect("ip link set %s vf 0 mac %s" %
+                             (self.host_intf0, self.vf0_mac), "# ")
+
+        self.used_dut_port_1 = self.dut_ports[1]
+        self.host_intf1 = self.dut.ports_info[self.used_dut_port_1]['intf']
+        self.dut.generate_sriov_vfs_by_port(
+            self.used_dut_port_1, 1, driver=driver)
+        self.sriov_vfs_port_1 = self.dut.ports_info[
+            self.used_dut_port_1]['vfs_port']
+        tester_port = self.tester.get_local_port(self.used_dut_port_1)
+        self.tester_intf1 = self.tester.get_interface(tester_port)
+
+        self.vf1_mac = "00:20:00:00:00:00"
+        self.dut.send_expect("ip link set %s vf 0 mac %s" %
+                             (self.host_intf1, self.vf1_mac), "# ")
+
+        try:
+
+            for port in self.sriov_vfs_port_0:
+                port.bind_driver('pci-stub')
+
+            for port in self.sriov_vfs_port_1:
+                port.bind_driver('pci-stub')
+
+            time.sleep(1)
+            vf0_prop = {'opt_host': self.sriov_vfs_port_0[0].pci}
+            vf1_prop = {'opt_host': self.sriov_vfs_port_1[0].pci}
+
+            # set up VM0 ENV
+            self.vm0 = QEMUKvm(self.dut, 'vm0', 'vf_vlan')
+            self.vm0.set_vm_device(driver='pci-assign', **vf0_prop)
+            self.vm0.set_vm_device(driver='pci-assign', **vf1_prop)
+            self.vm_dut_0 = self.vm0.start()
+            if self.vm_dut_0 is None:
+                raise Exception("Set up VM0 ENV failed!")
+
+        except Exception as e:
+            self.destroy_vm_env()
+            raise Exception(e)
+
+        self.env_done = True
+
+    def destroy_vm_env(self):
+        if getattr(self, 'vm0', None):
+            self.vm_dut_0.kill_all()
+            self.vm0_testpmd = None
+            self.vm0_dut_ports = None
+            # destroy vm0
+            self.vm0.stop()
+            self.vm0 = None
+
+        if getattr(self, 'used_dut_port_0', None):
+            self.dut.destroy_sriov_vfs_by_port(self.used_dut_port_0)
+            port = self.dut.ports_info[self.used_dut_port_0]['port']
+            self.used_dut_port_0 = None
+
+        if getattr(self, 'used_dut_port_1', None):
+            self.dut.destroy_sriov_vfs_by_port(self.used_dut_port_1)
+            port = self.dut.ports_info[self.used_dut_port_1]['port']
+            self.used_dut_port_1 = None
+
+        self.bind_nic_driver(self.dut_ports[:2], driver="igb_uio")
+
+        self.env_done = False
+
+    def test_pvid_vf_tx(self):
+        """
+        Add port based vlan on vf device and check vlan tx work
+        """
+        random_vlan = random.randint(1, 4095)
+
+        self.dut.send_expect(
+            "ip link set %s vf 0 vlan %d" % (self.host_intf0, random_vlan), "# ")
+        out = self.dut.send_expect("ip link show %s" % self.host_intf0, "# ")
+        self.verify("vlan %d" %
+                    random_vlan in out, "Failed to add pvid on VF0")
+
+        self.vm0_dut_ports = self.vm_dut_0.get_ports('any')
+
+        self.vm0_testpmd = PmdOutput(self.vm_dut_0)
+        self.vm0_testpmd.start_testpmd(VM_CORES_MASK)
+        self.vm0_testpmd.execute_cmd('set fwd mac')
+        self.vm0_testpmd.execute_cmd('start')
+
+        pkt = Packet(pkt_type='UDP')
+        pkt.config_layer('ether', {'dst': self.vf1_mac})
+        inst = sniff_packets(self.tester_intf0, timeout=5)
+        pkt.send_pkt(tx_port=self.tester_intf1)
+        pkts = load_sniff_packets(inst)
+
+        self.verify(len(pkts), "Not receive expected packet")
+        self.vm0_testpmd.quit()
+
+        # disable pvid
+        self.dut.send_expect(
+            "ip link set %s vf 0 vlan 0" % (self.host_intf0), "# ")
+
+    def send_and_getout(self, vlan=0, pkt_type="UDP"):
+
+        if pkt_type == "UDP":
+            pkt = Packet(pkt_type='UDP')
+            pkt.config_layer('ether', {'dst': self.vf0_mac})
+        elif pkt_type == "VLAN_UDP":
+            pkt = Packet(pkt_type='VLAN_UDP')
+            pkt.config_layer('dot1q', {'vlan': vlan})
+            pkt.config_layer('ether', {'dst': self.vf0_mac})
+
+        pkt.send_pkt(tx_port=self.tester_intf0)
+        out = self.vm_dut_0.get_session_output(timeout=2)
+
+        return out
+
+    def test_add_pvid_vf(self):
+        random_vlan = random.randint(1, 4095)
+
+        self.dut.send_expect(
+            "ip link set %s vf 0 vlan %d" % (self.host_intf0, random_vlan), "# ")
+        out = self.dut.send_expect("ip link show %s" % self.host_intf0, "# ")
+        self.verify("vlan %d" %
+                    random_vlan in out, "Failed to add pvid on VF0")
+
+        # start testpmd in VM
+        self.vm0_dut_ports = self.vm_dut_0.get_ports('any')
+
+        self.vm0_testpmd = PmdOutput(self.vm_dut_0)
+        self.vm0_testpmd.start_testpmd(VM_CORES_MASK)
+        self.vm0_testpmd.execute_cmd('set fwd rxonly')
+        self.vm0_testpmd.execute_cmd('set verbose 1')
+        self.vm0_testpmd.execute_cmd('start')
+
+        out = self.send_and_getout(vlan=random_vlan, pkt_type="VLAN_UDP")
+        self.verify("received" in out, "Failed to received vlan 
+ packet!!!")
+
+        # send packet without vlan
+        out = self.send_and_getout(pkt_type="UDP")
+        self.verify("received" not in out, "Received packet without 
+ vlan!!!")
+
+        # send packet with vlan not matched
+        wrong_vlan = (random_vlan + 1) % 4096
+        out = self.send_and_getout(vlan=wrong_vlan, pkt_type="VLAN_UDP")
+        self.verify(
+            "received" not in out, "Received pacekt with wrong 
+ vlan!!!")
+
+        # remove vlan
+        self.dut.send_expect(
+            "ip link set %s vf 0 vlan 0" % self.host_intf0, "# ")
+
+        # send packet with vlan
+        out = self.send_and_getout(vlan=random_vlan, pkt_type="VLAN_UDP")
+        self.verify(
+            "received" not in out, "Received vlan packet without 
+ pvid!!!")
+
+        # send packe with vlan 0
+        out = self.send_and_getout(vlan=0, pkt_type="VLAN_UDP")
+        self.verify(
+            "received" in out, "Not recevied packet with vlan 0!!!")
+
+        # send packet without vlan
+        out = self.send_and_getout(vlan=0, pkt_type="UDP")
+        self.verify("received" in out, "Not received packet without 
+ vlan!!!")
+
+        self.vm0_testpmd.quit()
+
+        # disable pvid
+        self.dut.send_expect(
+            "ip link set %s vf 0 vlan 0" % (self.host_intf0), "# ")
+
+    def test_vf_vlan_tx(self):
+        random_vlan = random.randint(1, 4095)
+        # start testpmd in VM
+        self.vm0_dut_ports = self.vm_dut_0.get_ports('any')
+
+        self.vm0_testpmd = PmdOutput(self.vm_dut_0)
+        self.vm0_testpmd.start_testpmd(VM_CORES_MASK)
+        self.vm0_testpmd.execute_cmd('set verbose 1')
+
+        self.vm0_testpmd.execute_cmd('tx_vlan set 0 %d' % random_vlan)
+
+        inst = sniff_packets(self.tester_intf0, timeout=5)
+        self.vm0_testpmd.execute_cmd('set burst 1')
+        self.vm0_testpmd.execute_cmd('start tx_first')
+
+        # strip sniffered vlans
+        pkts = load_sniff_packets(inst)
+        vlans = []
+        for pkt in pkts:
+            vlan = pkt.strip_element_dot1q("vlan")
+            vlans.append(vlan)
+
+        self.verify(
+            random_vlan in vlans, "Tx packet with vlan not 
+ received!!!")
+
+        self.vm0_testpmd.quit()
+
+    def test_vf_vlan_rx(self):
+        random_vlan = random.randint(1, 4095)
+        # start testpmd in VM
+        self.vm0_dut_ports = self.vm_dut_0.get_ports('any')
+
+        self.vm0_testpmd = PmdOutput(self.vm_dut_0)
+        self.vm0_testpmd.start_testpmd(VM_CORES_MASK)
+        self.vm0_testpmd.execute_cmd('set fwd rxonly')
+        self.vm0_testpmd.execute_cmd('set verbose 1')
+        self.vm0_testpmd.execute_cmd('vlan set strip off 0')
+        self.vm0_testpmd.execute_cmd('start')
+
+        # send packet without vlan
+        out = self.send_and_getout(vlan=0, pkt_type="UDP")
+        self.verify(
+            "received 1 packets" in out, "Not received normal packet as 
+ default!!!")
+
+        # send packet with vlan 0
+        out = self.send_and_getout(vlan=0, pkt_type="VLAN_UDP")
+        self.verify("VLAN tci=0x0"
+                    in out, "Not received vlan 0 packet as default!!!")
+
+        self.vm0_testpmd.execute_cmd('rx_vlan add %d 0' % random_vlan)
+        # send packet with same vlan
+        out = self.send_and_getout(vlan=random_vlan, pkt_type="VLAN_UDP")
+        vlan_hex = hex(random_vlan)
+        self.verify("VLAN tci=%s" %
+                    vlan_hex in out, "Not received expected vlan 
+ packet!!!")
+
+        pkt = Packet(pkt_type='VLAN_UDP')
+        wrong_vlan = (random_vlan + 1) % 4096
+        # send packet with wrong vlan
+        out = self.send_and_getout(vlan=wrong_vlan, pkt_type="VLAN_UDP")
+        self.verify(
+            "received 1 packets" not in out, "Received filtered vlan 
+ packet!!!")
+
+        self.vm0_testpmd.execute_cmd('rx_vlan rm 0 %d' % random_vlan)
+        # send packet with vlan 0
+        out = self.send_and_getout(vlan=0, pkt_type="VLAN_UDP")
+        self.verify("VLAN tci=0x0"
+                    in out, "Not received vlan 0 packet as default!!!")
+
+        # send packet without vlan
+        out = self.send_and_getout(pkt_type="UDP")
+        self.verify("received 1 packets" in out,
+                    "Not received normal packet after remove vlan 
+ filter!!!")
+
+        # send packet without vlan
+        out = self.send_and_getout(vlan=random_vlan, pkt_type="VLAN_UDP")
+        self.verify(
+            "received 1 packets" in out, "Not received vlan packet 
+ without vlan filter!!!")
+
+        self.vm0_testpmd.quit()
+
+    def test_vf_vlan_strip(self):
+        random_vlan = random.randint(1, 4095)
+        # start testpmd in VM
+        self.vm0_dut_ports = self.vm_dut_0.get_ports('any')
+
+        self.vm0_testpmd = PmdOutput(self.vm_dut_0)
+        self.vm0_testpmd.start_testpmd(VM_CORES_MASK)
+        self.vm0_testpmd.execute_cmd('set fwd rxonly')
+        self.vm0_testpmd.execute_cmd('set verbose 1')
+        self.vm0_testpmd.execute_cmd('vlan set strip on 0')
+        self.vm0_testpmd.execute_cmd('rx_vlan add %d 0' % random_vlan)
+        self.vm0_testpmd.execute_cmd('start')
+
+        out = self.send_and_getout(vlan=random_vlan, pkt_type="VLAN_UDP")
+        # enable strip, vlan will be in mbuf
+        vlan_hex = hex(random_vlan)
+        self.verify("VLAN tci=%s" %
+                    vlan_hex in out, "Failed to strip vlan packet!!!")
+
+        self.vm0_testpmd.execute_cmd('vlan set strip off 0')
+
+        out = self.send_and_getout(vlan=random_vlan, pkt_type="VLAN_UDP")
+        self.verify(
+            "received 1 packets" in out, "Not received vlan packet as expected!!!")
+        nic_type = self.vm_dut_0.ports_info[0]['type']
+        nic_name = get_nic_name(nic_type)
+        if nic_name in ['fvl10g_vf']:
+            self.verify("VLAN tci=%s" %
+                        vlan_hex in out, "Failed to disable strip vlan!!!")
+        else:
+            self.verify("VLAN tci=0x0" in out, "Failed to disable strip 
+ vlan!!!")
+
+        self.vm0_testpmd.quit()
+
+    def tear_down(self):
+        self.vm_dut_0.kill_all()
+        pass
+
+    def tear_down_all(self):
+        self.destroy_vm_env()
+        pass
--
1.9.3