Updates according to comments for Operational and Functional tests for CORD certification
Change-Id: I3e972fe7f6f85f1b95dd729d6d41f6319a52c8b4
diff --git a/Fabric/Tests/operational.py b/Fabric/Tests/operational.py
index 9292a08..1bedd5a 100644
--- a/Fabric/Tests/operational.py
+++ b/Fabric/Tests/operational.py
@@ -13,3 +13,635 @@
# 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