#!/usr/bin/env python3 import time import unittest from scapy.contrib.lacp import LACP, SlowProtocol, MarkerProtocol from scapy.layers.l2 import Ether from framework import VppTestCase, VppTestRunner from vpp_memif import remove_all_memif_vpp_config, VppSocketFilename, VppMemif from vpp_bond_interface import VppBondInterface from vpp_papi import VppEnum, MACAddress bond_mac = "02:02:02:02:02:02" lacp_dst_mac = '01:80:c2:00:00:02' LACP_COLLECTION_AND_DISTRIBUTION_STATE = 63 class TestMarker(VppTestCase): """LACP Marker Protocol Test Case """ @classmethod def setUpClass(cls): super().setUpClass() # Test variables cls.pkts_per_burst = 257 # Number of packets per burst # create 3 pg interfaces cls.create_pg_interfaces(range(1)) # packet sizes cls.pg_if_packet_sizes = [64, 512, 1518] # , 9018] # setup all interfaces for i in cls.pg_interfaces: i.admin_up() @classmethod def tearDownClass(cls): super().tearDownClass() def setUp(self): super().setUp() def tearDown(self): super().tearDown() def show_commands_at_teardown(self): self.logger.info(self.vapi.ppcli("show interface")) def test_marker_request(self): """ Marker Request test """ # topology # # +-+ +-+ # memif1 -----|B| |B|---- memif11 # |o| |o| # |n|------|n| # |d| |d| # pg0 -----|0| |1| # +-+ +-+ socket1 = VppSocketFilename( self, socket_id=1, socket_filename="%s/memif.sock1" % self.tempdir) socket1.add_vpp_config() socket11 = VppSocketFilename( self, socket_id=2, socket_filename="%s/memif.sock1" % self.tempdir) socket11.add_vpp_config() memif1 = VppMemif( self, role=VppEnum.vl_api_memif_role_t.MEMIF_ROLE_API_MASTER, mode=VppEnum.vl_api_memif_mode_t.MEMIF_MODE_API_ETHERNET, socket_id=1) memif1.add_vpp_config() memif1.admin_up() memif11 = VppMemif( self, role=VppEnum.vl_api_memif_role_t.MEMIF_ROLE_API_SLAVE, mode=VppEnum.vl_api_memif_mode_t.MEMIF_MODE_API_ETHERNET, socket_id=2) memif11.add_vpp_config() memif11.admin_up() bond0 = VppBondInterface( self, mode=VppEnum.vl_api_bond_mode_t.BOND_API_MODE_LACP, use_custom_mac=1, mac_address=bond_mac) bond0.add_vpp_config() bond0.admin_up() bond1 = VppBondInterface( self, mode=VppEnum.vl_api_bond_mode_t.BOND_API_MODE_LACP) bond1.add_vpp_config() bond1.admin_up() bond0.add_member_vpp_bond_interface(sw_if_index=memif1.sw_if_index) bond1.add_member_vpp_bond_interface(sw_if_index=memif11.sw_if_index) # wait for memif protocol exchange and hardware carrier to come up self.assertEqual(memif1.wait_for_link_up(10), True) self.assertEqual(memif11.wait_for_link_up(10), True) # verify memif1 in bond0 intfs = self.vapi.sw_member_interface_dump( sw_if_index=bond0.sw_if_index) for intf in intfs: self.assertEqual(intf.sw_if_index, memif1.sw_if_index) # verify memif11 in bond1 intfs = self.vapi.sw_member_interface_dump( sw_if_index=bond1.sw_if_index) for intf in intfs: self.assertEqual(intf.sw_if_index, memif11.sw_if_index) self.vapi.ppcli("trace add memif-input 100") # create marker request marker = (Ether(src=bond_mac, dst=lacp_dst_mac) / SlowProtocol() / MarkerProtocol(marker_type=1, requester_port=1, requester_system=bond_mac, requester_transaction_id=1)) bond1.add_member_vpp_bond_interface(sw_if_index=self.pg0.sw_if_index) self.pg0.add_stream(marker) self.pg_enable_capture(self.pg_interfaces) self.pg_start() show_trace = self.vapi.ppcli("show trace max 100") self.assertIn("Marker Information TLV:", show_trace) bond0.remove_vpp_config() bond1.remove_vpp_config() class TestLACP(VppTestCase): """LACP Test Case """ @classmethod def setUpClass(cls): super().setUpClass() @classmethod def tearDownClass(cls): super().tearDownClass() def setUp(self): su