[pytest] Add support to populate DUT FDB entries

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

@@ -0,0 +1,176 @@
+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
+
+            It accepts MAC to IP ratio and packet count. It generates packets withratio of distinct MAC addresses
+            to distinct IP addresses as provided. The IP addresses starts from VLAN address pool.
+
+            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/common/fixtures/populate_fdb.py

@@ -0,0 +1,128 @@
+import json
+import logging
+import pytest
+
+from ptf_runner import ptf_runner
+
+logger = logging.getLogger(__name__)
+
+class PopulateFdb:
+    """
+        PopulateFdb populates DUT FDB entries
+
+        It 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.
+
+        Command line sample:
+            pytest testbed_setup/test_populate_fdb.py --testbed=<testbed> --inventory=<inventory> --testbed_file=<testbed fiel> \
+            --host-pattern={<dut>|all} --module-path=<ansible library path> --mac_to_ip_ratio=100:1 --packet_count=8000
+
+            where:
+                mac_to_ip_ratio: Ratio of distinct MAC addresses to distinct IP addresses assigned to VM
+                packet_count: Number of packets to be created and sent to DUT
+                start_mac: VM start MAC address. Subsequent MAC addresses are increment of 1 on top of start MAC
+    """
+    PTFRUNNER_QLEN = 1000
+    VLAN_CONFIG_FILE = "/tmp/vlan_config.json"
+
+    def __init__(self, request, duthost, ptfhost):
+        """
+            Class constructor
+
+            Args:
+                request: pytest request object
+                duthost (AnsibleHost): Device Under Test (DUT)
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        """
+        self.macToIpRatio = request.config.getoption("--mac_to_ip_ratio")
+        self.startMac = request.config.getoption("--start_mac")
+        self.packetCount = request.config.getoption("--packet_count")
+
+        self.duthost = duthost
+        self.ptfhost = ptfhost
+
+    def __prepareVlanConfigData(self):
+        """
+            Prepares Vlan Configuration data
+
+            Args:
+                duthost (AnsibleHost): Device Under Test (DUT)
+                ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+            Returns:
+                None
+        """
+        mgVlanPorts = []
+        mgFacts = self.duthost.minigraph_facts(host=self.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]
+                })
+
+        vlanConfigData = {
+            "vlan_ports": mgVlanPorts,
+            "vlan_interfaces": {vlan["attachto"]: vlan for vlan in mgFacts["minigraph_vlan_interfaces"]},
+            "dut_mac": self.duthost.setup()["ansible_facts"]["ansible_Ethernet0"]["macaddress"]
+        }
+
+        with open(self.VLAN_CONFIG_FILE, 'w') as file:
+            file.write(json.dumps(vlanConfigData, indent=4))
+
+        logger.info("Copying VLan config file to {0}".format(self.ptfhost.hostname))
+        self.ptfhost.copy(src=self.VLAN_CONFIG_FILE, dest="/tmp/")
+
+        logger.info("Copying ptftests to {0}".format(self.ptfhost.hostname))
+        self.ptfhost.copy(src="ptftests", dest="/root")
+
+    def run(self):
+        """
+            Populates DUT FDB entries
+
+            Args:
+                None
+
+            Returns:
+                None
+        """
+        self.__prepareVlanConfigData()
+
+        logger.info("Populate DUT FDB entries")
+        ptf_runner(
+            self.ptfhost,
+            "ptftests",
+            "populate_fdb.PopulateFdb",
+            qlen=self.PTFRUNNER_QLEN,
+            platform_dir="ptftests",
+            platform="remote",
+            params={
+                "start_mac": self.startMac,
+                "config_data": self.VLAN_CONFIG_FILE,
+                "packet_count": self.packetCount,
+                "mac_to_ip_ratio": self.macToIpRatio,
+            },
+            log_file="/tmp/populate_fdb.PopulateFdb.log"
+        )
+
+@pytest.fixture
+def populate_fdb(request, duthost, ptfhost):
+    """
+        Populates DUT FDB entries
+
+        Args:
+            request: pytest request object
+            duthost (AnsibleHost): Device Under Test (DUT)
+            ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+        Returns:
+            None
+    """
+    populateFdb = PopulateFdb(request, duthost, ptfhost)
+
+    populateFdb.run()

tests/testbed_setup/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='Ratio of distinct MAC addresses to distinct IP addresses assigned to VM',
+    )
+
+    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/testbed_setup/conftest.py

@@ -0,0 +1,15 @@
+from args.populate_fdb_args import add_populate_fdb_args
+from common.fixtures.populate_fdb import populate_fdb
+
+# 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/testbed_setup/test_populate_fdb.py

@@ -0,0 +1,15 @@
+import pytest
+
+def test_populate_fdb(populate_fdb):
+    """
+        Populates DUT FDB entries
+
+        Args:
+            request: pytest request object
+            duthost (AnsibleHost): Device Under Test (DUT)
+            ptfhost (AnsibleHost): Packet Test Framework (PTF)
+
+        Returns:
+            None
+    """
+    pass