[pytest] Add support to populate DUT FDB entries

ansible/roles/test/files/ptftests/populate_fdb.py

@@ -0,0 +1,173 @@
+import ipaddress
+import json
+import logging
+import ptf
+
+# Packet Test Framework imports
+import ptf
+import ptf.packet as scapy
+import ptf.testutils as testutils
+from ptf import config
+from ptf.base_tests import BaseTest
+
+logger = logging.getLogger(__name__)
+
+class PopulateFdb(BaseTest):
+    '''
+        Populate DUT FDB entries
+    '''
+    TCP_DST_PORT = 5000
+    TCP_SRC_PORT = 6000
+
+    def __init__(self):
+        '''
+            class constructor
+
+            Args:
+                None
+
+            Returns:
+                None
+        '''
+        BaseTest.__init__(self)
+
+    def setUp(self):
+        '''
+            Sets up Populate FDB instance data
+
+            Args:
+                None
+
+            Returns:
+                None
+        '''
+        self.dataplane = ptf.dataplane_instance
+        self.dataplane.flush()
+
+        self.testParams = testutils.test_params_get()
+        self.packetCount = self.testParams['packet_count']
+        self.startMac = self.testParams['start_mac']
+
+        self.configFile = self.testParams['config_data']
+        with open(self.configFile) as fp:
+            self.configData = json.load(fp)
+
+        self.dutMac = self.configData['dut_mac']
+        self.macToIpRatio = [int(i) for i in self.testParams['mac_to_ip_ratio'].split(':')]
+        self.assertTrue(
+            len(self.macToIpRatio) == 2 and self.macToIpRatio[0] > 0 and self.macToIpRatio[1] > 0,
+            'Invalid MAC to IP ratio: {0}'.format(self.testParams['mac_to_ip_ratio'])
+        )
+
+        if config["log_dir"] is not None:
+            filename = os.path.join(config["log_dir"], str(self)) + ".pcap"
+            self.dataplane.start_pcap(filename)
+
+    def tearDown(self):
+        '''
+            Tears down FDB instance data
+
+            Args:
+                None
+
+            Returns:
+                None
+        '''
+        if config["log_dir"] is not None:
+            self.dataplane.stop_pcap()
+
+    def __convertMacToInt(self, mac):
+        '''
+            Converts MAC address to integer
+
+            Args:
+                mac (str): MAC Address
+
+            Returns:
+                mac (int): integer representation of MAC address
+        '''
+        return int(mac.translate(None, ":.- "), 16)
+
+    def __convertMacToStr(self, mac):
+        '''
+            Converts MAC address to string
+
+            Args:
+                mac (int): MAC Address
+
+            Returns:
+                mac (str): string representation of MAC address
+        '''
+        mac = "{:012x}".format(mac)
+        return ":".join(mac[i : i + 2] for i in range(0, len(mac), 2))
+
+    def __prepareVmIp(self):
+        '''
+            Prepares VM IP addresses
+
+            Args:
+                None
+
+            Returns:
+                vmIp (dict): Map containing vlan to VM IP address
+        '''
+        vmIp = {}
+        for vlan, config in self.configData['vlan_interfaces'].items():
+            prefixLen = self.configData['vlan_interfaces'][vlan]['prefixlen']
+            ipCount = 2**(32 - prefixLen) - 3
+            numDistinctIp = self.packetCount * self.macToIpRatio[1] / self.macToIpRatio[0]
+            self.assertTrue(
+                ipCount >= numDistinctIp,
+                'Vlan network \'{0}\' does not support the requested number of IPs \'{1}\''.format(
+                    ipCount,
+                    numDistinctIp
+                )
+            )
+            vmIp[vlan] = ipaddress.ip_address(unicode(config['addr'])) + 1
+
+        return vmIp
+
+    def __populateDutFdb(self):
+        '''
+            Populates DUT FDB entries
+
+            Args:
+                None
+
+            Returns:
+                None
+        '''
+        packet = testutils.simple_tcp_packet(
+            eth_dst=self.dutMac,
+            tcp_sport=self.TCP_SRC_PORT,
+            tcp_dport=self.TCP_DST_PORT
+        )
+        vmIp = self.__prepareVmIp()
+        macInt = self.__convertMacToInt(self.startMac)
+        numMac = numIp = 0
+        for i in range(self.packetCount):
+            port = i % len(self.configData['vlan_ports'])
+            vlan = self.configData['vlan_ports'][port]['vlan']
+
+            if i % self.macToIpRatio[1] == 0:
+                mac = self.__convertMacToStr(macInt + i)
+                numMac += 1
+            if i % self.macToIpRatio[0] == 0:
+                vmIp[vlan] = ipaddress.ip_address(unicode(vmIp[vlan])) + 1
+                numIp += 1
+
+            packet[scapy.Ether].src = mac
+            packet[scapy.IP].src = str(vmIp[vlan])
+            packet[scapy.IP].dst = self.configData['vlan_interfaces'][vlan]['addr']
+            testutils.send(self, self.configData['vlan_ports'][port]['index'], packet)
+
+        logger.info(
+            'Generated {0} packets with distinct {1} MAC addresses and {2} IP addresses'.format(
+                self.packetCount,
+                numMac,
+                numIp
+            )
+        )
+
+    def runTest(self):
+        self.__populateDutFdb()

tests/fdb/args/populate_fdb_args.py

@@ -0,0 +1,35 @@
+# Populate FDB Args file
+
+def add_populate_fdb_args(parser):
+    '''
+        Adding arguments required for populate fdb test cases
+
+        Args:
+            parser: pytest parser object
+
+        Returns:
+            None
+    '''
+    parser.addoption(
+        '--mac_to_ip_ratio',
+        action='store',
+        type=str,
+        default='100:1',
+        help='VM MAC to IP assigned ratio',
+    )
+
+    parser.addoption(
+        '--start_mac',
+        action='store',
+        type=str,
+        default='00:25:ae:22:11:00',
+        help='VM start MAC address. Subsequent MAC addresses are increment of 1 on top of start MAC',
+    )
+
+    parser.addoption(
+        '--packet_count',
+        action='store',
+        type=int,
+        default=2000,
+        help='Number of packets to be created and sent to DUT',
+    )

tests/fdb/conftest.py

@@ -0,0 +1,14 @@
+from args.populate_fdb_args import add_populate_fdb_args
+
+# FDB pytest arguments
+def pytest_addoption(parser):
+    '''
+        Adds option to FDB pytest
+
+        Args:
+            parser: pytest parser object
+
+        Returns:
+            None
+    '''
+    add_populate_fdb_args(parser)

tests/fdb/test_populate_fdb.py

@@ -0,0 +1,126 @@
+import json
+import logging
+import pytest
+
+from common.platform.ssh_utils import prepare_testbed_ssh_keys as prepareTestbedSshKeys
+from ptf_runner import ptf_runner
+
+logger = logging.getLogger(__name__)
+
+# Globals
+PTFRUNNER_QLEN = 1000
+VXLAN_CONFIG_FILE = '/tmp/vxlan_config.json'
+
+class TestPopulateFdb:
+    '''
+        TestPopulateFdb populates DUT FDB entries
+    '''
+    @pytest.fixture(scope='class', autouse=True)
+    def prepareVxlanConfigData(self, duthost, ptfhost):
+        '''
+            Prepares Vxlan Configuration data
+
+            Args:
+                duthost (AnsibleHost): Device Under Test (DUT)
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        '''
+        mgVlanPorts = []
+        mgFacts = duthost.minigraph_facts(host=duthost.hostname)['ansible_facts']
+        for vlan, config in mgFacts['minigraph_vlans'].items():
+            for port in config['members']:
+                mgVlanPorts.append({
+                    'port': port,
+                    'vlan': vlan,
+                    'index': mgFacts['minigraph_port_indices'][port]
+                })
+
+        vxlanConfigData = {
+            'vlan_ports': mgVlanPorts,
+            'vlan_interfaces': {vlan['attachto']: vlan for vlan in mgFacts['minigraph_vlan_interfaces']},
+            'dut_mac': duthost.setup()['ansible_facts']['ansible_Ethernet0']['macaddress']
+        }
+
+        with open(VXLAN_CONFIG_FILE, 'w') as file:
+            file.write(json.dumps(vxlanConfigData, indent=4))
+
+        logger.info('Copying VxLan config file to {0}'.format(ptfhost.hostname))
+        ptfhost.copy(src=VXLAN_CONFIG_FILE, dest='/tmp/')
+
+    @pytest.fixture(scope='class', autouse=True)
+    def copyPtfDirectory(self, ptfhost):
+        '''
+            Copys PTF directory to PTF host. This class-scope fixture runs once before test start
+
+            Args:
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        '''
+        ptfhost.copy(src="ptftests", dest="/root")
+
+    @pytest.fixture(scope='class', autouse=True)
+    def removePtfhostIp(self, ptfhost):
+        '''
+            Removes IP assigned to eth<n> inerface of PTF host. This class-scope fixture runs once before test start
+
+            Args:
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        '''
+        ptfhost.script(src='scripts/remove_ip.sh')
+
+    @pytest.fixture(scope='class', autouse=True)
+    def changePtfhostMacAddresses(self, ptfhost):
+        '''
+            Change MAC addresses (unique) on PTF host. This class-scope fixture runs once before test start
+
+            Args:
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        '''
+        ptfhost.script(src="scripts/change_mac.sh")
+
+    def testPopulateFdb(self, request, duthost, ptfhost):
+        '''
+            Populates DUT FDB entries
+
+            The accepts MAC to IP ratio (default 100:1) and packet count (default 2000). It generates packets with
+            ratio of distinct MAC addresses to distinct IP addresses as provided. The IP addresses starts from VLAN
+            address pool.
+
+            Args:
+                request: pytest request object
+                duthost (AnsibleHost): Device Under Test (DUT)
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        '''
+        macToIpRatio = request.config.getoption("--mac_to_ip_ratio")
+        startMac = request.config.getoption("--start_mac")
+        packetCount = request.config.getoption("--packet_count")
+
+        logger.info('Populate DUT FDB entries')
+        ptf_runner(
+            ptfhost,
+            'ptftests',
+            'populate_fdb.PopulateFdb',
+            qlen=PTFRUNNER_QLEN,
+            platform_dir='ptftests',
+            platform='remote',
+            params={
+                'start_mac': startMac,
+                'config_data': VXLAN_CONFIG_FILE,
+                'packet_count': packetCount,
+                'mac_to_ip_ratio': macToIpRatio,
+            },
+            log_file='/tmp/populate_fdb.PopulateFdb.log'
+        )