new file mode 100644
@@ -0,0 +1,184 @@
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright(c) 2023 Intel Corporation
+#
+
+import re
+import time
+
+from framework.pmd_output import PmdOutput
+from framework.test_case import TestCase
+from framework.utils import GREEN, RED
+
+
+class TimestampTest(TestCase):
+ def set_up_all(self):
+ """
+ Run at the start of each test suite.
+ Generic filter Prerequistites
+ """
+ self.verify(
+ self.nic in ["IGC-I225_LM", "IGC-I226_LM"],
+ "%s nic not support tx timestamp" % self.nic,
+ )
+ self.dut_ports = self.dut.get_ports(self.nic)
+ self.ports_socket = self.dut.get_numa_id(self.dut_ports[0])
+ # Verify that enough ports are available
+ self.verify(len(self.dut_ports) >= 1, "Insufficient ports")
+ self.tester_port0 = self.tester.get_local_port(self.dut_ports[0])
+ self.tester_iface0 = self.tester.get_interface(self.tester_port0)
+ self.pmdout = PmdOutput(self.dut)
+ self.pf_pci = self.dut.ports_info[self.dut_ports[0]]["pci"]
+
+ apps_name = "test-timestamp"
+ self.build_test_app(apps_name)
+
+ def set_up(self):
+ """
+ Run before each test case.
+ """
+ pass
+
+ def build_test_app(self, apps_name):
+ # Compile test app
+ print(self.tester.send_expect("ls", "#", 5))
+ test_rx_timestamp = "/root/pkt_rx_tx.c"
+ out = self.tester.send_expect(
+ "gcc %s -o %s -lpthread" % (test_rx_timestamp, apps_name), "#", 15
+ )
+ self.verify("Error" not in out, "Compilation %s error" % apps_name)
+ output = self.tester.send_expect("find -name %s" % apps_name, "#")
+ self.verify("%s" % apps_name in output, "Compilation %s error" % apps_name)
+
+ def launch_testpmd(self, param, intra):
+ """
+ start testpmd
+ """
+ # Prepare testpmd EAL and parameters
+ self.pmdout.start_testpmd(
+ param=param,
+ eal_param=f"-a {self.pf_pci}",
+ socket=self.ports_socket,
+ )
+ # Test link status
+ res = self.pmdout.wait_link_status_up("all", 5)
+ self.verify(res is True, "there have port link is down")
+ self.pmdout.execute_cmd("set fwd txonly")
+ self.pmdout.execute_cmd("set txtimes 30000000,%d" % intra)
+ self.pmdout.execute_cmd("start", "testpmd>", 5)
+ self.pmdout.execute_cmd("stop")
+
+ def launch_test_app(self, apps_name):
+ print(self.tester.send_expect("ls", "#", 5))
+ self.tester.send_command("./%s -i %s -t" % (apps_name, self.tester_iface0), "#")
+ out = self.tester.get_session_output(timeout=0.1)
+ self.verify("Press Enter to continue" in out, "Launch test app failed.")
+
+ def check_time_interval_error(self, interval):
+ out = self.tester.get_session_output(timeout=0.1)
+ rx_ts_pat = ".*ts\s+(\d+)"
+ rx_ts = re.findall(rx_ts_pat, out, re.M)
+ rx_ts = list(map(int, rx_ts))
+ self.error_msgs = []
+ intervals = []
+ if len(rx_ts) != 32:
+ error_msg = "Tester should recieve 32 RX timestamps."
+ self.logger.error(error_msg)
+ self.error_msgs.append(error_msg)
+ else:
+ for i in range(len(rx_ts) - 1):
+ rx = rx_ts[i + 1] - rx_ts[i]
+ intervals.append(rx)
+ self.logger.info("Rx time intervals: {}".format(intervals))
+ error = [abs((x - interval) / interval) * 100 for x in intervals]
+ error = [round(x, 2) for x in error]
+ self.logger.info("Rx time interval error: {}".format(error))
+ if all(x < 50 for x in error) == False:
+ error_msg = "The time interval error for Rx timestamp is too large."
+ self.logger.error(error_msg)
+ self.error_msgs.append(error_msg)
+ else:
+ self.logger.info(
+ "max_error={}%, min_error={}%".format(max(error), min(error))
+ )
+ self.verify(not self.error_msgs, "some subcases failed")
+
+ def handle_tx_timestamp_case(self, apps_name, param, intra, interval):
+ self.launch_test_app(apps_name)
+ self.launch_testpmd(param, intra)
+ self.check_time_interval_error(interval)
+ self.pmdout.execute_cmd("quit", "#")
+
+ def test_with_tx_timestamp_check_time_interval(self, apps_name="test-timestamp"):
+ # Subcase 1: set txtimes intra 2000ns, check the time interval of Rx timestamp is 2000ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 1: set txtimes intra 2000ns ================"
+ )
+ self.handle_tx_timestamp_case(
+ apps_name, param="--tx-offloads=0x200000", intra=2000, interval=2000
+ )
+
+ # Subcase 2: set txtimes intra 1000ns, check the time interval of Rx timestamp is 1000ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 2: set txtimes intra 1000ns ================"
+ )
+ self.handle_tx_timestamp_case(
+ apps_name, param="--tx-offloads=0x200000", intra=1000, interval=1000
+ )
+
+ # Subcase 3: set txtimes intra 400ns, check the time interval of Rx timestamp is 400ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 3: set txtimes intra 400ns ================"
+ )
+ self.handle_tx_timestamp_case(
+ apps_name, param="--tx-offloads=0x200000", intra=400, interval=400
+ )
+
+ # Subcase 4: set txtimes intra 300ns, check the time interval of Rx timestamp is 300ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 4: set txtimes intra 300ns ================"
+ )
+ self.handle_tx_timestamp_case(
+ apps_name, param="--tx-offloads=0x200000", intra=300, interval=300
+ )
+
+ # Subcase 5: set txtimes intra 200ns, check the time interval of Rx timestamp is 300ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 5: set txtimes intra 200ns ================"
+ )
+ self.handle_tx_timestamp_case(
+ apps_name, param="--tx-offloads=0x200000", intra=200, interval=300
+ )
+
+ def test_without_tx_timestamp_check_time_interval(self, apps_name="test-timestamp"):
+ # Subcase 1: set txtimes intra 2000ns, check the time interval of Rx timestamp is 300ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 1: set txtimes intra 2000ns ================"
+ )
+ self.handle_tx_timestamp_case(apps_name, param="", intra=2000, interval=300)
+
+ # Subcase 2: set txtimes intra 1000ns, check the time interval of Rx timestamp is 300ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 2: set txtimes intra 1000ns ================"
+ )
+ self.handle_tx_timestamp_case(apps_name, param="", intra=1000, interval=300)
+
+ # Subcase 3: set txtimes intra 400ns, check the time interval of Rx timestamp is 300ns and measure MAX/MIN error.
+ self.logger.info(
+ "===================Test subcase 3: set txtimes intra 400ns ================"
+ )
+ self.handle_tx_timestamp_case(apps_name, param="", intra=400, interval=300)
+
+ def tear_down(self):
+ """
+ Run after each test case.
+ """
+ # Bind the interfaces to the igc driver, then bind to the vfio-pci driver.
+ self.pmdout.execute_cmd("quit", "#")
+ self.dut.bind_interfaces_linux("igc")
+ self.dut.bind_interfaces_linux("vfio-pci")
+
+ def tear_down_all(self):
+ """
+ Run after each test suite.
+ """
+ self.dut.kill_all()