Fabric/Tests/operational.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


"""
Check README file
"""
import logging
import os
import time
import json

import Queue
import ofp
import oftest.base_tests as base_tests
from accton_util import *
from oftest import config
from oftest.ofdpa_utils import *
from oftest.testutils import *
from oftest.utils import *

#### Global variables  ####
req_onos = "no" # to identify weather a test case require ONOS controller or not; default value is "no" for all the test cases without ONOS server

### Read config parameters from JSON file 'global_vars.json'
ROOT_DIR = os.path.dirname(os.path.realpath(__file__))
with open(ROOT_DIR + '/' + 'global_vars.json') as f:
    data = json.load(f)

switch_ip = data['switch_ip']
switch_user = data['switch_user']
switch_passwd = data['switch_passwd']
controller_ip = data['controller_ip']
controller_port = data['controller_port']
onos_server_ip = data['onos_server_ip']
onos_user = data['onos_user']
onos_passwd = data['onos_passwd']
onos_port = data['onos_port']


class FabricSW_OP_TC_0005(base_tests.SimpleDataPlane):
    """
    To verify packet flooding when ports associated with L2 Interface groups are flapped administratively (disable/enable) - single interface pair.
    """
    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        vlan_id101 = 101
        vlan_id201 = 201
        global req_onos

        try:
            if len( config[ "port_map" ] ) < 2:
                logging.info( "Port count less than 2, can't run this case" )
                assert False, "Port count less than 2, can't run this case"
                return

            if req_onos == "yes":
                req_onos = "no"
                ofagent_reconfig(self, arg="no")
                ofagent_restart(self)
                time.sleep(90)

            of_ports = config["port_map"].keys()

            create_flows(self, groups, vlan_id101, 1, 2)
            logging.info("Admin down interfaces")
            port_admin(self, of_ports[0], admin_state="disable")
            port_admin(self, of_ports[1], admin_state="disable")
            time.sleep(2)
            logging.info("Admin up interfaces")
            port_admin(self, of_ports[0], admin_state="enable")
            port_admin(self, of_ports[1], admin_state="enable")
            time.sleep(7)

            # Verify flood VLAN101 from port1 to port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)
            transmit_and_verify_packets(self, vlan_id201, 1, 2, False)

        finally:
            delete_all_flows(self.controller)
            delete_groups(self.controller, groups)
            delete_all_groups(self.controller)
            logging.info("End of Test")


class FabricSW_OP_TC_0010(base_tests.SimpleDataPlane):
    """
    To verify packet forwarding when ports associated with L2 Interface groups are flapped administratively (disable/enable) - multiple interface pairs.
    """

    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        try:
            if len( config[ "port_map" ] ) < 3:
                logging.info("Port count less than 3, can't run this case")
                assert False, "Port count less than 3, can't run this case"
                return

            if req_onos == "yes":
                req_onos = "no"
                ofagent_reconfig(self, arg="no")
                ofagent_restart(self)
                time.sleep(90)

            ports = (config["port_map"].keys())
            vlan_id101 = 101
            vlan_id151 = 151

            # Create flows between two pairs of ports
            create_flows(self, groups, vlan_id101, 1, 2)
            create_flows(self, groups, vlan_id151, 1, 3)

            logging.info("Admin down interface " + str(ports[2]))
            port_admin(self, ports[2], admin_state="disable")
            time.sleep(5)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

            logging.info("Admin up interface " + str(ports[2]))
            port_admin(self, ports[2], admin_state="enable")
            time.sleep(7)

            # Send VLAN 151 to port1 and verify packet on port3
            transmit_and_verify_packets(self,vlan_id151, 1, 3, True)

            # Send VLAN 151 to port3 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id151, 3, 1, True)

            logging.info("Admin down interface " + str(ports[0]))
            port_admin(self, ports[0], admin_state="disable")
            time.sleep(3)

            logging.info("Admin up interface " + str(ports[0]))
            port_admin(self, ports[0], admin_state="enable")
            time.sleep(5)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

            # Send VLAN 151 to port1 and verify packet on port3
            transmit_and_verify_packets(self, vlan_id151, 1, 3, True)

        finally:
            delete_all_flows(self.controller)
            delete_groups(self.controller, groups)
            delete_all_groups(self.controller)
            logging.info("End of Test")


class FabricSW_OP_TC_0015(base_tests.SimpleDataPlane):
    """
    To verify control channels are re-established and flows are re-configuration after fabric switch is rebooted.
    """
    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        vlan_id101 = 101
        try:
            datapathid = get_datapathid(self)

            if len( config[ "port_map" ] ) < 2:
                logging.info( "Port count less than 2, can't run this case" )
                assert False, "Port count less than 2, can't run this case"
                return

            if req_onos == "no":
                req_onos = "yes"
                ofagent_reconfig(self, arg="yes")
                ofagent_restart(self)
                time.sleep(90)

            ports = (config["port_map"].keys())

            # Create flows between two pairs of ports
            add_onos_xconnect(self, datapathid, vlan_id101, ports[0], ports[1])
            time.sleep(10)

            #Reboot switch
            switch_restart(self)
            time.sleep(120)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

        finally:
            # Remove vlan-cross connection
            # send_command_to_onos_cli(self.log_path, "clear_xconnection1", "sr-xconnect-remove of:" + datapathid + " " + str(vlan_id101),
            #                          self.onos_server_ip, self.onos_user, self.onos_passwd, regexp='onos>')
            remove_onos_xconnect(self, datapathid, vlan_id101)
            time.sleep(3)
            logging.info("End of Test")


class FabricSW_OP_TC_0020(base_tests.SimpleDataPlane):
    """
    To verify control channels are re-established and flows are re-configuration after of-agent restart.
    """
    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        vlan_id101 = 101
        try:
            datapathid = get_datapathid(self)

            if len( config[ "port_map" ] ) < 2:
                logging.info( "Port count less than 2, can't run this case" )
                assert False, "Port count less than 2, can't run this case"
                return

            if req_onos == "no":
                req_onos = "yes"
                ofagent_reconfig(self, arg="yes")
                ofagent_restart(self)
                time.sleep(60)

            ports = (config["port_map"].keys())
            # Create flows between two pairs of ports
            add_onos_xconnect(self, datapathid, vlan_id101, ports[0], ports[1])
            time.sleep(2)

            # Agent restart
            ofagent_restart(self)
            time.sleep(90)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

        finally:
            # Remove vlan-cross connection
            remove_onos_xconnect(self, datapathid, vlan_id101)
            time.sleep(3)
            logging.info("End of Test")


class FabricSW_OP_TC_0025(base_tests.SimpleDataPlane):
    """
    To verify packet forwarding when ports associated with L2 Interface groups are flapped physically.
    """
    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        try:
            if len( config[ "port_map" ] ) < 2:
                logging.info( "Port count less than 2, can't run this case" )
                assert False, "Port count less than 2, can't run this case"
                return

            if req_onos == "yes":
                req_onos = "no"
                ofagent_reconfig(self, arg="no")
                ofagent_restart(self)
                time.sleep(90)

            vlan_id101 = 101
            vlan_id201 = 201
            ports = (config["port_map"].keys())

            # Create flows between two pairs of ports
            create_flows(self, groups, vlan_id101, 1, 2)

            logging.info("Simulate plug-out and Plug-in the physical cable on port1 (through bcm port state commands)")
            switch_port_enable(self, ports[0], False)
            time.sleep(2)
            switch_port_enable(self, ports[0], True)
            time.sleep(5)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

            # Send VLAN 201 to port1 and verify no packet on port2
            transmit_and_verify_packets(self, vlan_id201, 1, 2, False)

        finally:
            delete_all_flows(self.controller)
            delete_groups(self.controller, groups)
            delete_all_groups(self.controller)
            logging.info("End of Test")


class FabricSW_OP_TC_0030(base_tests.SimpleDataPlane):
    """
    To verify switch owners the flow rules when vlan-cross-connect pair is configured on operationally down interface and then interfaces are enabled.
    """
    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        try:
            if len( config[ "port_map" ] ) < 2:
                logging.info( "Port count less than 2, can't run this case" )
                assert False, "Port count less than 2, can't run this case"
                return

            if req_onos == "yes":
                req_onos = "no"
                ofagent_reconfig(self, arg="no")
                ofagent_restart(self)
                time.sleep(60)

            ports = (config["port_map"].keys())
            vlan_id101 = 101
            vlan_id201 = 201

            logging.info("Plug-out the physical cable on port1 (through bcm port state commands)")
            switch_port_enable(self, ports[0], False)
            time.sleep(2)

            # Create flows between two pairs of ports
            create_flows(self, groups, vlan_id101, 1, 2)

            logging.info("Plug-in the physical cable on port1 (through bcm port state commands)")
            switch_port_enable(self, ports[0], True)
            time.sleep(5)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            # Send VLAN 201 to port1 and verify no packet on port2
            transmit_and_verify_packets(self, vlan_id201, 1, 2, False)

        finally:
            delete_all_flows(self.controller)
            delete_groups(self.controller, groups)
            delete_all_groups(self.controller)
            logging.info("End of Test")


class FabricSW_OP_TC_0035(base_tests.SimpleDataPlane):
    """
    To verify flows are successfully modified when ports associated with vlan-cross-connect pair is changed.
    """
    controller_port = controller_port
    onos_port = onos_port
    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    onos_server_ip = onos_server_ip
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        vlan_id101 = 101
        try:
            datapathid = get_datapathid(self)
            if len( config[ "port_map" ] ) < 3:
                logging.info( "Port count less than 3, can't run this case" )
                assert False, "Port count less than 3, can't run this case"
                return

            if req_onos == "no":
                req_onos = "yes"
                ofagent_reconfig(self, arg="yes")

            ofagent_restart(self)
            time.sleep(90)

            ports = (config["port_map"].keys())

            # Create flows between two pairs of ports
            add_onos_xconnect(self, datapathid, vlan_id101, ports[0], ports[1])
            time.sleep(10)

            #Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            #Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

            # Modify vlan-cross connect with port3 instead of port2
            add_onos_xconnect(self, datapathid, vlan_id101, ports[0], ports[2])
            time.sleep(10)
            #
            # #Send VLAN 101 to port1 and verify packet on port3
            transmit_and_verify_packets(self, vlan_id101, 1, 3, True)
            #
            # #Send VLAN 101 to port3 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 3, 1, True)
            #
            # #Send VLAN 101 to port2 and verify no packets on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, False)

        finally:
            #Remove vlan-cross connection
            remove_onos_xconnect(self, datapathid, vlan_id101)
            time.sleep(5)
            logging.info("End of Test")


class FabricSW_OP_TC_0040(base_tests.SimpleDataPlane):
    """
    To verify flows are removed when vlan-cross-connect pair is removed.
    """

    switch_user = switch_user
    switch_passwd = switch_passwd
    switch_ip = switch_ip
    controller_ip = controller_ip
    controller_port = controller_port
    onos_server_ip = onos_server_ip
    onos_port = onos_port
    onos_user = onos_user
    onos_passwd = onos_passwd

    def runTest(self):
        groups = Queue.LifoQueue()
        global req_onos
        vlan_id101 = 101
        try:
            datapathid = get_datapathid(self)

            if len( config[ "port_map" ] ) < 2:
                logging.info( "Port count less than 2, can't run this case" )
                assert False, "Port count less than 2, can't run this case"
                return

            if req_onos == "no":
                req_onos = "yes"
                ofagent_reconfig(self, arg="yes")
                ofagent_restart(self)
                time.sleep(90)

            ports = (config["port_map"].keys())

            # Create flows between two pairs of ports
            add_onos_xconnect(self, datapathid, vlan_id101, ports[0], ports[1])
            time.sleep(10)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, True)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, True)

            # Remove vlan-cross connection
            remove_onos_xconnect(self, datapathid, vlan_id101)
            time.sleep(10)

            # Send VLAN 101 to port1 and verify packet on port2
            transmit_and_verify_packets(self, vlan_id101, 1, 2, False)

            # Send VLAN 101 to port2 and verify packet on port1
            transmit_and_verify_packets(self, vlan_id101, 2, 1, False)

        finally:
            # Remove vlan-cross connection
            remove_onos_xconnect(self, datapathid, vlan_id101)
            logging.info("End of Test")


def create_flows(self, groups, vid, port1, port2):
    ports = (config["port_map"].keys())
    pair = [ports[(port1-1)], ports[(port2-1)]]

    for port in pair:
        L2gid, l2msg = add_one_l2_interface_group(self.controller, port, vid, True, False)
        add_one_vlan_table_flow(self.controller, port, vlan_id=vid, flag=VLAN_TABLE_FLAG_ONLY_TAG)
        groups.put(L2gid)

    msg = add_l2_flood_group(self.controller, pair, vid, vid)
    groups.put(msg.group_id)
    add_bridge_flow(self.controller, None, vid, msg.group_id, True)

    logging.info(
        "Add a flow to TABLE-60 to match vlan {} and action Send to Controller".format(vid))
    add_acl_rule(self.controller, vlan_id=vid)

    match = ofp.match()
    match.oxm_list.append(ofp.oxm.vlan_vid(vid))


def transmit_and_verify_packets(self, vid, port1, port2, recv):
    # @parameters:
    # vid - VLAN ID to create a flow
    # port1 - a port where to send packets
    # port2 - a port where to verify packets
    # recv = True or False: True - we expect receive packets on port2; False - we don't expect recive packets on port2

    ports = (config["port_map"].keys())
    pair = [ports[(port1-1)], ports[(port2-1)]]

    # verify flood
    logging.info("Creating a double tagged vlan packet with outer vlan id {}".format(vid))
    # change dest based on port number
    mac_src = '00:12:34:56:78:%02X' % pair[0]
    parsed_pkt = simple_tcp_packet_two_vlan(pktlen=108, out_dl_vlan_enable=True,
                                                out_vlan_vid=vid, in_dl_vlan_enable=True, in_vlan_vid=10,
                                                eth_dst='00:12:34:56:78:9a', eth_src=mac_src)

    pkt = str(parsed_pkt)
    self.dataplane.send(pair[0], pkt)
    # self won't rx packet
    verify_no_packet(self, pkt, pair[0])
    if recv == True:
        # verify rx packet
        tmp_ports = list(pair)
        tmp_ports.remove(pair[0])
        verify_packets(self, pkt, tmp_ports)
    else:
        #Verify the packet is not flooded
        verify_no_packet(self, pkt, ports[(port2-1)])



def port_admin(self, port, admin_state):
    # @parameters:
    # port - number of port to set admin state disable or enable
    # admin_state for port = enable | disable

    # Retrieve Port Configuration
    logging.info("Sends Features Request and retrieve Port Configuration from reply")
    (hw_addr, port_config, advert) = \
        port_config_get(self.controller, port)
    self.assertTrue(port_config is not None, "Did not get port config")

    logging.debug("Admin state bit of port " + str(port) + " is now " +
                  str(port_config & ofp.OFPPC_PORT_DOWN))

    # Modify Port Configuration
    logging.info("Modify Port Configuration using Port Modification Message:OFPT_PORT_MOD")

    if admin_state == "disable":

        rv = port_config_set(self.controller, port,
                             port_config ^ ofp.OFPPC_PORT_DOWN, ofp.OFPPC_PORT_DOWN)
        self.assertTrue(rv != -1, "Error sending port mod")
        do_barrier(self.controller)

        # Verify change took place with features request
        logging.info("Verify the change")
        (hw_addr, port_config2, advert) = port_config_get(self.controller, port)

        logging.debug("Admin state bit " + str(port) + " is now " +
                      str(port_config2 & ofp.OFPPC_PORT_DOWN))
        self.assertTrue(port_config2 is not None, "Did not get port config2")
        self.assertTrue(port_config2 & ofp.OFPPC_PORT_DOWN !=
                        port_config & ofp.OFPPC_PORT_DOWN,
                        "Bit change did not take")

    if admin_state == "enable":
        rv = port_config_set(self.controller, port, port_config ^ ofp.OFPPC_PORT_DOWN, 0)
        self.assertTrue(rv != -1, "Error sending port mod")
        do_barrier(self.controller)

        # Verify change took place with features request
        logging.info("Verify the change")
        (hw_addr, port_config2, advert) = port_config_get(self.controller, port)

        logging.debug("Admin state bit " + str(port) + " is now " + str(port_config2 & ofp.OFPPC_PORT_DOWN))
        self.assertTrue(port_config2 is not None, "Did not get port config2")
        self.assertTrue(port_config2 & ofp.OFPPC_PORT_DOWN != port_config & ofp.OFPPC_PORT_DOWN, "Bit change did not take")


def get_datapathid(self):
    feature_reply = get_featureReplay(self)
    str_datapath_id_f = "{:016x}".format(feature_reply.datapath_id)
    return str_datapath_id_f