| |
| # 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 |