ptf/saibridgeport.py

# Copyright 2021-present Intel Corporation.
#
# 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.
"""
Thrift SAI interface Bridge Port tests
"""

from sai_thrift.sai_headers import *

from sai_base_test import *


@group("draft")
class BridgePortAttributeTest(SaiHelper):
    '''
    Verify bridge port attributes getting and setting
    '''

    def setUp(self):
        super(BridgePortAttributeTest, self).setUp()

        self.port24_bp = sai_thrift_create_bridge_port(
            self.client,
            bridge_id=self.default_1q_bridge,
            port_id=self.port24,
            type=SAI_BRIDGE_PORT_TYPE_PORT,
            admin_state=True,
            fdb_learning_mode=SAI_BRIDGE_PORT_FDB_LEARNING_MODE_HW)
        self.assertNotEqual(self.port24_bp, 0)

    def runTest(self):
        # bridge_id
        # get
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, bridge_id=True)
        self.assertEqual(attr['bridge_id'], self.default_1q_bridge)

        # type
        # get
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, type=True)
        self.assertEqual(attr['type'], SAI_BRIDGE_PORT_TYPE_PORT)

        # port_id
        # get
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, port_id=True)
        self.assertEqual(attr['port_id'], self.port24)

        # fdb_learning_mode
        # get
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, fdb_learning_mode=True)
        self.assertEqual(attr['fdb_learning_mode'],
                         SAI_BRIDGE_PORT_FDB_LEARNING_MODE_HW)
        # set
        status = sai_thrift_set_bridge_port_attribute(
            self.client,
            self.port24_bp,
            fdb_learning_mode=SAI_BRIDGE_PORT_FDB_LEARNING_MODE_DISABLE)
        self.assertEqual(status, SAI_STATUS_SUCCESS)
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, fdb_learning_mode=True)
        self.assertEqual(attr['fdb_learning_mode'],
                         SAI_BRIDGE_PORT_FDB_LEARNING_MODE_DISABLE)

        # admin_state
        # get
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, admin_state=True)
        self.assertEqual(attr['admin_state'], True)
        # set
        status = sai_thrift_set_bridge_port_attribute(
            self.client, self.port24_bp, admin_state=False)
        self.assertEqual(status, SAI_STATUS_SUCCESS)
        attr = sai_thrift_get_bridge_port_attribute(
            self.client, self.port24_bp, admin_state=True)
        self.assertEqual(attr['admin_state'], False)

    def tearDown(self):
        sai_thrift_remove_bridge_port(self.client, self.port24_bp)

        super(BridgePortAttributeTest, self).tearDown()


@group("draft")
class BridgePortCreationTest(SaiHelper):
    '''
    Verify bridge port creation
    '''

    def runTest(self):
        self.noBpDropTest()
        self.bpTypePortCreationTest()

    def noBpDropTest(self):
        '''
        This verifies if packets are dropped on port when no bridge port is
        created
        '''
        print("\nnoBpDropTest")

        no_bp_port = self.dev_port24

        src_mac = "00:11:11:11:11:11"
        dst_mac = "00:22:22:22:22:22"

        pkt = simple_udp_packet(eth_src=src_mac, eth_dst=dst_mac, pktlen=100)

        print("Sending packet on port with no bridge port created")
        send_packet(self, no_bp_port, pkt)
        verify_no_other_packets(self)
        print("\tPacket dropped. OK")

    def bpTypePortCreationTest(self):
        '''
        This verifies creation of bridge port of type port
        Bridge must be 1Q bridge
        '''
        print("\nbpTypePortCreationTest")

        try:
            print("Verifying bp type Port creation for physical port")
            port24_bp = sai_thrift_create_bridge_port(
                self.client,
                bridge_id=self.default_1q_bridge,
                port_id=self.port24,
                type=SAI_BRIDGE_PORT_TYPE_PORT,
                admin_state=True)
            self.assertNotEqual(port24_bp, 0)
            print("\tOK")

            print("Verifying bp type Port creation for LAG logical port")
            test_lag = sai_thrift_create_lag(self.client)
            self.assertNotEqual(test_lag, 0)
            test_lag_member1 = sai_thrift_create_lag_member(
                self.client, lag_id=test_lag, port_id=self.port25)
            self.assertNotEqual(test_lag_member1, 0)
            test_lag_member2 = sai_thrift_create_lag_member(
                self.client, lag_id=test_lag, port_id=self.port26)
            self.assertNotEqual(test_lag_member2, 0)

            lag_bp = sai_thrift_create_bridge_port(
                port_id=test_lag,
                type=SAI_BRIDGE_PORT_TYPE_PORT,
                admin_state=True)
            self.assertNotEqual(lag_bp, 0)
            print("\tOK")

        finally:
            sai_thrift_remove_bridge_port(self.client, lag_bp)
            sai_thrift_remove_lag_member(self.client, test_lag_member1)
            sai_thrift_remove_lag_member(self.client, test_lag_member2)
            sai_thrift_remove_lag(self.client, test_lag)
            sai_thrift_remove_bridge_port(self.client, port24_bp)


@group("draft")
class BridgePortStateTest(SaiHelper):
    '''
    Verify switch behavior in particular bridge port state cases
    '''

    def setUp(self):
        super(BridgePortStateTest, self).setUp()

        self.vlan_id = 10
        self.src_mac = "00:11:11:11:11:11"
        self.dst_mac = "00:22:22:22:22:22"

        self.pkt = simple_udp_packet(eth_dst=self.dst_mac,
                                     eth_src=self.src_mac,
                                     pktlen=100)
        self.tag_pkt = simple_udp_packet(eth_dst=self.dst_mac,
                                         eth_src=self.src_mac,
                                         dl_vlan_enable=True,
                                         vlan_vid=self.vlan_id,
                                         pktlen=104)

        self.chck_pkt = simple_udp_packet(eth_dst=self.src_mac,
                                          eth_src=self.dst_mac,
                                          pktlen=100)
        self.chck_tag_pkt = simple_udp_packet(eth_dst=self.src_mac,
                                              eth_src=self.dst_mac,
                                              dl_vlan_enable=True,
                                              vlan_vid=self.vlan_id,
                                              pktlen=104)

    def runTest(self):
        self.bpStateDownFlushTest()
        self.bpStateDownNoLearnTest()

    def bpStateDownFlushTest(self):
        '''
        This verifies if FDB entries related to MAC addresses learned on port
        corresponding to bridge port are flushed while bridge port admin state
        is setting to false
        '''
        print("\nbpStateDownFlushTest")

        test_dev_port = self.dev_port0
        chck_dev_port = self.dev_port1
        test_bp = self.port0_bp

        # other VLAN 10 ports
        flood_port_list = [[chck_dev_port],
                           [self.dev_port4, self.dev_port5, self.dev_port6]]
        chck_flood_port_list = [[test_dev_port],
                                [self.dev_port4, self.dev_port5,
                                 self.dev_port6]]

        fdb_entry = sai_thrift_fdb_entry_t(switch_id=self.switch_id,
                                           mac_address=self.src_mac,
                                           bv_id=self.vlan10)

        flood_pkt_list = [self.tag_pkt, self.pkt]
        chck_flood_pkt_list = [self.chck_pkt, self.chck_pkt]

        try:
            print("Sending packet on port %d to learn MAC address - "
                  "will flood" % test_dev_port)
            send_packet(self, test_dev_port, self.pkt)
            verify_each_packet_on_multiple_port_lists(
                self, flood_pkt_list, flood_port_list)
            time.sleep(2)

            print("Checking if MAC address was learned")
            status = sai_thrift_set_fdb_entry_attribute(
                self.client,
                fdb_entry,
                packet_action=SAI_PACKET_ACTION_FORWARD)
            self.assertEqual(status, SAI_STATUS_SUCCESS)
            print("\tOK")

            print("Sending packet on port %d, %s -> %s - forwarding" %
                  (chck_dev_port, self.dst_mac, self.src_mac))
            send_packet(self, chck_dev_port, self.chck_tag_pkt)
            verify_packets(self, self.chck_pkt, [test_dev_port])
            print("\tOK")

            print("Setting bridge port state to down")
            status = sai_thrift_set_bridge_port_attribute(
                self.client, test_bp, admin_state=False)
            self.assertEqual(status, SAI_STATUS_SUCCESS)

            print("Checking if FDB entry was removed")
            status = sai_thrift_set_fdb_entry_attribute(
                self.client,
                fdb_entry,
                packet_action=SAI_PACKET_ACTION_FORWARD)
            self.assertEqual(status, SAI_STATUS_ITEM_NOT_FOUND)

            print("Sending packet on port %d, %s -> %s - will flood" %
                  (chck_dev_port, self.dst_mac, self.src_mac))
            send_packet(self, chck_dev_port, self.chck_tag_pkt)
            verify_each_packet_on_multiple_port_lists(
                self, chck_flood_pkt_list, chck_flood_port_list)
            print("\tOK")

            print("\tVerification complete")

        finally:
            sai_thrift_flush_fdb_entries(
                self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
            sai_thrift_set_bridge_port_attribute(
                self.client, test_bp, admin_state=True)

    def bpStateDownNoLearnTest(self):
        '''
        This checks if MAC addresses are not learned on port when bridge port
        state is down
        We set the down state on port0 and send a packet form port0 with given
        SMAC: 00:11:11:11:11:11, and then we verify it was broadcasted on other
        ports in VLAN.
        Next we send a packet from port1 with DMAC 00:11:11:11:11:11
        and we verify it was also broadcasted on other ports in VLAN
        '''
        print("\nbpStateDownNoLearnTest()")

        test_dev_port = self.dev_port0
        chck_dev_port = self.dev_port1
        test_bp = self.port0_bp
        new_bp_port = self.port24
        new_bp_dev_port = self.dev_port24

        try:
            # set port0_bp state to DOWN
            status = sai_thrift_set_bridge_port_attribute(
                self.client, test_bp, admin_state=False)
            self.assertEqual(status, SAI_STATUS_SUCCESS)
            print("Port %d state set to DOWN" % test_dev_port)

            flood_pkt_list = [self.tag_pkt, self.pkt]
            flood_port_list = [[chck_dev_port],
                               [self.dev_port4, self.dev_port5,
                                self.dev_port6]]
            print("Sending packet on port %d, %s -> %s - will flood" %
                  (test_dev_port, self.src_mac, self.dst_mac))
            send_packet(self, test_dev_port, self.pkt)
            verify_each_packet_on_multiple_port_lists(
                self, flood_pkt_list, flood_port_list)
            print("\tOK")

            chck_flood_pkt_list = [self.chck_pkt, self.chck_pkt]
            chck_flood_port_list = [[test_dev_port],
                                    [self.dev_port4, self.dev_port5,
                                     self.dev_port6]]
            print("Sending packet on port %d with DMAC %s - "
                  "MAC not learned, will flood" %
                  (chck_dev_port, self.src_mac))
            send_packet(self, chck_dev_port, self.chck_tag_pkt)
            verify_each_packet_on_multiple_port_lists(
                self, chck_flood_pkt_list, chck_flood_port_list)
            print("\tOK")

            sai_thrift_flush_fdb_entries(
                self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)

            # create new bridge port with admin state down
            new_bp = sai_thrift_create_bridge_port(
                self.client,
                bridge_id=self.default_1q_bridge,
                port_id=new_bp_port,
                type=SAI_BRIDGE_PORT_TYPE_PORT,
                admin_state=False)
            self.assertNotEqual(new_bp, 0)
            new_vlan_member = sai_thrift_create_vlan_member(
                self.client,
                vlan_id=self.vlan10,
                bridge_port_id=new_bp,
                vlan_tagging_mode=SAI_VLAN_TAGGING_MODE_UNTAGGED)
            self.assertNotEqual(new_vlan_member, 0)
            sai_thrift_set_port_attribute(
                self.client, new_bp_port, port_vlan_id=self.vlan_id)

            flood_pkt_list = [self.pkt, self.tag_pkt, self.pkt]
            flood_port_list = [[test_dev_port], [chck_dev_port],
                               [self.dev_port4, self.dev_port5,
                                self.dev_port6]]
            print("Sending packet on newly created bp port %d, %s -> %s - "
                  "will flood" % (new_bp_dev_port, self.src_mac, self.dst_mac))
            send_packet(self, new_bp_dev_port, self.pkt)
            verify_each_packet_on_multiple_port_lists(
                self, flood_pkt_list, flood_port_list)
            print("\tOK")

            chck_flood_pkt_list = [self.chck_pkt, self.chck_pkt, self.chck_pkt]
            chck_flood_port_list = [[new_bp_dev_port], [test_dev_port],
                                    [self.dev_port4, self.dev_port5,
                                     self.dev_port6]]
            print("Sending packet on port %d with DMAC %s - "
                  "MAC not learned, will flood" %
                  (chck_dev_port, self.src_mac))
            send_packet(self, chck_dev_port, self.chck_tag_pkt)
            verify_each_packet_on_multiple_port_lists(
                self, chck_flood_pkt_list, chck_flood_port_list)
            print("\tOK")

            print("Verification complete")

        finally:
            sai_thrift_flush_fdb_entries(
                self.client, entry_type=SAI_FDB_FLUSH_ENTRY_TYPE_ALL)
            sai_thrift_set_port_attribute(
                self.client, new_bp_port, port_vlan_id=0)
            sai_thrift_remove_vlan_member(self.client, new_vlan_member)
            sai_thrift_remove_bridge_port(self.client, new_bp)

            # restore initial port0 admin state
            sai_thrift_set_bridge_port_attribute(
                self.client, test_bp, admin_state=True)


@group("draft")
class BridgePortNoFloodTest(SaiHelperBase):
    '''
    Verify bridge port no flood test case
    '''

    def setUp(self):
        super(BridgePortNoFloodTest, self).setUp()

        self.test_port_list = [self.port0, self.port1, self.port2, self.port3]
        self.dev_test_port_list = [
            self.dev_port0,
            self.dev_port1,
            self.dev_port2,
            self.dev_port3]
        vlan_id = 1
        for port in self.test_port_list:
            sai_thrift_set_port_attribute(
                self.client, port, port_vlan_id=vlan_id)
            attr = sai_thrift_get_port_attribute(
                self.client, port, port_vlan_id=True)
            self.assertEqual(attr['SAI_PORT_ATTR_PORT_VLAN_ID'], vlan_id)

    def runTest(self):
        print("SAIBridgePortNoFloodTest")

        src_mac = "00:11:11:11:11:11"
        dst_mac = "00:22:22:22:22:22"

        pkt = simple_udp_packet(eth_dst=dst_mac,
                                eth_src=src_mac,
                                pktlen=100)
        print("Sending packet on port with no bridge port created")
        send_packet(self, self.dev_port1, pkt)
        verify_no_other_packets(self)
        print("\tPacket dropped. OK")

    def tearDown(self):
        # revert original port's VLAN id
        vlan_id = 10
        for port in self.test_port_list:
            sai_thrift_set_port_attribute(
                self.client, port, port_vlan_id=vlan_id)
            attr = sai_thrift_get_port_attribute(
                self.client, port, port_vlan_id=True)
            self.assertTrue(attr['SAI_PORT_ATTR_PORT_VLAN_ID'], vlan_id)

        super(BridgePortNoFloodTest, self).tearDown()