test_vlan.py

import pytest
import ptf.packet as scapy
import ptf.testutils as testutils
from ptf.mask import Mask
from collections import defaultdict

import time
import json
import itertools
import logging
import pprint

from common.errors import RunAnsibleModuleFail

logger = logging.getLogger(__name__)

vlan_id_list = [ 100, 200 ]


@pytest.fixture(scope="module")
def cfg_facts(duthost):
    return duthost.config_facts(host=duthost.hostname, source="persistent")['ansible_facts']

@pytest.fixture(scope="module")
def vlan_intfs_list():
    return [ { 'vlan_id': vlan, 'ip': '192.168.{}.1/24'.format(vlan) } for vlan in vlan_id_list  ]

@pytest.fixture(scope="module")
def vlan_ports_list(cfg_facts, ptfhost):
    vlan_ports_list = []
    config_ports = cfg_facts['PORT']
    config_portchannels = cfg_facts.get('PORTCHANNEL', {})
    config_port_indices = cfg_facts['port_index_map']
    ptf_ports_available_in_topo = ptfhost.host.options['variable_manager'].extra_vars.get("ifaces_map")

    pvid_cycle = itertools.cycle(vlan_id_list)

    # when running on t0 we can use the portchannel members
    if config_portchannels:
        for po in config_portchannels.keys()[:2]:
            port = config_portchannels[po]['members'][0]
            vlan_ports_list.append({
                'dev' : po,
                'port_index' : config_port_indices[port],
                'pvid' : pvid_cycle.next(),
                'permit_vlanid' : { vid : {
                    'peer_ip' : '192.168.{}.{}'.format(vid, 2 + config_port_indices.keys().index(port)),
                    'remote_ip' : '{}.1.1.{}'.format(vid, 2 + config_port_indices.keys().index(port))
                    } for vid in vlan_id_list }
            })

    ports = [ port for port in config_ports
        if config_port_indices[port] in ptf_ports_available_in_topo and config_ports[port].get('admin_status', 'down') == 'up' ]

    for port in ports[:4]:
        vlan_ports_list.append({
            'dev' : port,
            'port_index' : config_port_indices[port],
            'pvid' : pvid_cycle.next(),
            'permit_vlanid' : { vid : {
                'peer_ip' : '192.168.{}.{}'.format(vid, 2 + config_port_indices.keys().index(port)),
                'remote_ip' : '{}.1.1.{}'.format(vid, 2 + config_port_indices.keys().index(port))
                } for vid in vlan_id_list }
        })

    return vlan_ports_list


def create_vlan_interfaces(vlan_ports_list, vlan_intfs_list, duthost, ptfhost):
    for vlan_port in vlan_ports_list:
        for permit_vlanid in vlan_port["permit_vlanid"].keys():
            if int(permit_vlanid) != vlan_port["pvid"]:

                ptfhost.command("ip link add link eth{idx} name eth{idx}.{pvid} type vlan id {pvid}".format(
                   idx=vlan_port["port_index"],
                   pvid=permit_vlanid
                ))

                ptfhost.command("ip link set eth{idx}.{pvid} up".format(
                   idx=vlan_port["port_index"],
                   pvid=permit_vlanid
                ))

    for vlan in vlan_intfs_list:
        duthost.command("config interface ip add Vlan{} {}".format(vlan['vlan_id'], vlan['ip']))

@pytest.fixture(scope="module", autouse=True)
def setup_vlan(ptfadapter, duthost, ptfhost, vlan_ports_list, vlan_intfs_list, cfg_facts):

    # --------------------- Setup -----------------------
    try:
        # Generate vlan info
        portchannel_interfaces = cfg_facts.get('PORTCHANNEL_INTERFACE', {})

        logger.info("Shutdown lags, flush IP addresses")
        for portchannel, ips in portchannel_interfaces.items():
            duthost.command('config interface shutdown {}'.format(portchannel))
            for ip in ips:
                duthost.command('config interface ip remove {} {}'.format(portchannel, ip))

        # Wait some time for route, neighbor, next hop groups to be removed,
        # otherwise PortChannel RIFs are still referenced and won't be removed
        time.sleep(90)

        logger.info("Add vlans, assign IPs")
        for vlan in vlan_intfs_list:
            duthost.command('config vlan add {}'.format(vlan['vlan_id']))
            duthost.command("config interface ip add Vlan{} {}".format(vlan['vlan_id'], vlan['ip'].upper()))

        logger.info("Add members to Vlans")
        for vlan_port in vlan_ports_list:
            for permit_vlanid in vlan_port['permit_vlanid'].keys():
                duthost.command('config vlan member add {tagged} {id} {port}'.format(
                    tagged=('--untagged' if vlan_port['pvid'] == permit_vlanid else ''),
                    id=permit_vlanid,
                    port=vlan_port['dev']
                ))

        # Make sure config applied
        time.sleep(30)

        logger.info("Bringup lags")
        for portchannel in portchannel_interfaces:
            duthost.command('config interface startup {}'.format(portchannel))

        # Make sure config applied
        time.sleep(30)

        logger.info("Create VLAN intf")
        create_vlan_interfaces(vlan_ports_list, vlan_intfs_list, duthost, ptfhost)

        logger.info("Configure route for remote IP")
        for item in vlan_ports_list:
            for i in vlan_ports_list[0]['permit_vlanid']:
                duthost.command('ip route add {} via {}'.format(
                    item['permit_vlanid'][i]['remote_ip'],
                    item['permit_vlanid'][i]['peer_ip']
                    ))

        logger.info("Copy arp_responder to ptfhost")
        ptfhost.copy(src='scripts/arp_responder.py', dest='/opt')

        setUpArpResponder(vlan_ports_list, ptfhost)

        extra_vars = {
                'arp_responder_args': ''
        }

        ptfhost.host.options['variable_manager'].extra_vars.update(extra_vars)
        ptfhost.template(src='arp_responder.conf.j2', dest='/tmp')
        ptfhost.command("cp /tmp/arp_responder.conf.j2 /etc/supervisor/conf.d/arp_responder.conf")

        ptfhost.command('supervisorctl reread')
        ptfhost.command('supervisorctl update')

        logger.info("Start arp_responder")
        ptfhost.command('supervisorctl start arp_responder')

        time.sleep(10)

    # --------------------- Testing -----------------------
        yield
    # --------------------- Teardown -----------------------
    finally:
        tearDown(vlan_ports_list, duthost, ptfhost, vlan_intfs_list, portchannel_interfaces)


def tearDown(vlan_ports_list, duthost, ptfhost, vlan_intfs_list, portchannel_interfaces):

    logger.info("VLAN test ending ...")
    logger.info("Stop arp_responder")
    ptfhost.command('supervisorctl stop arp_responder')

    logger.info("Delete VLAN intf")
    try:
        for vlan_port in vlan_ports_list:
            for permit_vlanid in vlan_port["permit_vlanid"].keys():
                if int(permit_vlanid) != vlan_port["pvid"]:
                    ptfhost.command("ip link delete eth{idx}.{pvid}".format(
                    idx=vlan_port["port_index"],
                    pvid=permit_vlanid
                    ))
    except RunAnsibleModuleFail as e:
        logger.error(e)

    duthost.shell("config reload -y &>/dev/null")

    # make sure Portchannels go up for post-test link sanity
    time.sleep(90)


def setUpArpResponder(vlan_ports_list, ptfhost):
    d = defaultdict(list)
    for vlan_port in vlan_ports_list:
        for permit_vlanid in vlan_port["permit_vlanid"].keys():
            if int(permit_vlanid) == vlan_port["pvid"]:
                iface = "eth{}".format(vlan_port["port_index"])
            else:
                iface = "eth{}.{}".format(vlan_port["port_index"], permit_vlanid)
            d[iface].append(vlan_port["permit_vlanid"][permit_vlanid]["peer_ip"])

    with open('/tmp/from_t1.json', 'w') as file:
        json.dump(d, file)
    ptfhost.copy(src='/tmp/from_t1.json', dest='/tmp/from_t1.json')

def build_icmp_packet(vlan_id, src_mac="00:22:00:00:00:02", dst_mac="ff:ff:ff:ff:ff:ff",
                        src_ip="192.168.0.1", dst_ip="192.168.0.2", ttl=64):

    pkt = testutils.simple_icmp_packet(pktlen=100 if vlan_id == 0 else 104,
                                eth_dst=dst_mac,
                                eth_src=src_mac,
                                dl_vlan_enable=False if vlan_id == 0 else True,
                                vlan_vid=vlan_id,
                                vlan_pcp=0,
                                ip_src=src_ip,
                                ip_dst=dst_ip,
                                ip_ttl=ttl)
    return pkt

def verify_icmp_packets(ptfadapter, vlan_ports_list, vlan_port, vlan_id):
    untagged_dst_ports = []
    tagged_dst_ports = []
    untagged_pkts = []
    tagged_pkts = []
    untagged_pkt = build_icmp_packet(0)
    tagged_pkt = build_icmp_packet(vlan_id)

    for port in vlan_ports_list:
        if vlan_port["port_index"] == port["port_index"]:
            # Skip src port
            continue
        if port["pvid"] == vlan_id:
            untagged_dst_ports.append(port["port_index"])
            untagged_pkts.append(untagged_pkt)
        elif vlan_id in map(int, port["permit_vlanid"].keys()):
            tagged_dst_ports.append(port["port_index"])
            tagged_pkts.append(tagged_pkt)
    logger.info("Verify untagged packets from ports " + str(untagged_dst_ports) + " tagged packets from ports " + str(tagged_dst_ports))
    testutils.verify_each_packet_on_each_port(ptfadapter, untagged_pkts+tagged_pkts, untagged_dst_ports+tagged_dst_ports)

@pytest.mark.bsl
def test_vlan_tc1_send_untagged(ptfadapter, vlan_ports_list):
    """
    Test case #1
    Verify packets egress without tag from ports whose PVID same with ingress port
    Verify packets egress with tag from ports who include VLAN ID but PVID different from ingress port.
    """

    logger.info("Test case #1 starting ...")

    for vlan_port in vlan_ports_list:
        pkt = build_icmp_packet(0)
        logger.info("Send untagged packet from {} ...".format(vlan_port["port_index"]))
        logger.info(pkt.sprintf("%Ether.src% %IP.src% -> %Ether.dst% %IP.dst%"))
        testutils.send(ptfadapter, vlan_port["port_index"], pkt)
        verify_icmp_packets(ptfadapter, vlan_ports_list, vlan_port, vlan_port["pvid"])


@pytest.mark.bsl
def test_vlan_tc2_send_tagged(ptfadapter, vlan_ports_list):
    """
    Test case #2
    Send tagged packets from each port.
    Verify packets egress without tag from ports whose PVID same with ingress port
    Verify packets egress with tag from ports who include VLAN ID but PVID different from ingress port.
    """

    logger.info("Test case #2 starting ...")

    for vlan_port in vlan_ports_list:
        for permit_vlanid in map(int, vlan_port["permit_vlanid"].keys()):
            pkt = build_icmp_packet(permit_vlanid)
            logger.info("Send tagged({}) packet from {} ...".format(permit_vlanid, vlan_port["port_index"]))
            logger.info(pkt.sprintf("%Ether.src% %IP.src% -> %Ether.dst% %IP.dst%"))
            testutils.send(ptfadapter, vlan_port["port_index"], pkt)
            verify_icmp_packets(ptfadapter, vlan_ports_list, vlan_port, permit_vlanid)

@pytest.mark.bsl
def test_vlan_tc3_send_invalid_vid(ptfadapter, vlan_ports_list):
    """
    Test case #3
    Send packets with invalid VLAN ID
    Verify no port can receive these pacekts
    """

    logger.info("Test case #3 starting ...")

    invalid_tagged_pkt = build_icmp_packet(4095)
    masked_invalid_tagged_pkt = Mask(invalid_tagged_pkt)
    masked_invalid_tagged_pkt.set_do_not_care_scapy(scapy.Dot1Q, "vlan")

    for vlan_port in vlan_ports_list:
        src_port = vlan_port["port_index"]
        dst_ports = [port["port_index"] for port in vlan_ports_list
                                if port != vlan_port ]
        logger.info("Send invalid tagged packet " + " from " + str(src_port) + "...")
        logger.info(invalid_tagged_pkt.sprintf("%Ether.src% %IP.src% -> %Ether.dst% %IP.dst%"))
        testutils.send(ptfadapter, src_port, invalid_tagged_pkt)
        logger.info("Check on " + str(dst_ports) + "...")
        testutils.verify_no_packet_any(ptfadapter, masked_invalid_tagged_pkt, dst_ports)