[PATCH v12 net-next 3/3] selftests: drivers/net: Implement ptp4l sync test using netdevsim

[Maciek Machnikowski <maciek@machnikowski.net>]

Add PTP synchronization test using ptp4l and netdevsim.
The test uses the NetDrvEpEnv to link a local netdevsim
device to a remote endpoint, runs ptp4l as leader and follower
on the two ends, and waits for the follower to report the
synchronized state (s2).

Reviewed-by: Mina Almasry <almasrymina@google.com>
Signed-off-by: Maciek Machnikowski <maciek@machnikowski.net>
---
 tools/testing/selftests/drivers/net/Makefile |   1 +
 tools/testing/selftests/drivers/net/config   |   1 +
 tools/testing/selftests/drivers/net/ptp.py   | 112 +++++++++++++++++++
 3 files changed, 114 insertions(+)
 create mode 100755 tools/testing/selftests/drivers/net/ptp.py

diff --git a/tools/testing/selftests/drivers/net/Makefile b/tools/testing/selftests/drivers/net/Makefile
index 79c6f998e704..ed41a69ef6ad 100644
--- a/tools/testing/selftests/drivers/net/Makefile
+++ b/tools/testing/selftests/drivers/net/Makefile
@@ -23,6 +23,7 @@ TEST_PROGS := \
 	ping.py \
 	pppoe_gro.py \
 	psp.py \
+	ptp.py \
 	queues.py \
 	ring_reconfig.py \
 	rss_key.py \
diff --git a/tools/testing/selftests/drivers/net/config b/tools/testing/selftests/drivers/net/config
index 4838adf27fa1..30e69446d41d 100644
--- a/tools/testing/selftests/drivers/net/config
+++ b/tools/testing/selftests/drivers/net/config
@@ -21,6 +21,7 @@ CONFIG_NET_SCH_INGRESS=y
 CONFIG_NET_SCH_PRIO=m
 CONFIG_PPP=y
 CONFIG_PPPOE=y
+CONFIG_PTP_1588_CLOCK_MOCK=y
 CONFIG_TLS=y
 CONFIG_VLAN_8021Q=m
 CONFIG_XDP_SOCKETS=y
diff --git a/tools/testing/selftests/drivers/net/ptp.py b/tools/testing/selftests/drivers/net/ptp.py
new file mode 100755
index 000000000000..1065bd42ba53
--- /dev/null
+++ b/tools/testing/selftests/drivers/net/ptp.py
@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# By Maciek Machnikowski <maciek@machnikowski.net> (c) 2026
+
+"""
+Test suite for PTP sync using ptp4l.
+
+Start a ptp4l leader and follower and check that the follower locks onto the
+leader (state s2)
+"""
+
+import time
+
+from lib.py import (
+    KsftSkipEx,
+    NetDrvEpEnv,
+    bkg,
+    cmd,
+    defer,
+    fd_read_timeout,
+    ksft_exit,
+    ksft_pr,
+    ksft_run,
+    ksft_true,
+)
+
+PTP4L_SYNC_TIMEOUT = 40
+
+
+def _require_hw_timestamping(ifname, host=None):
+    """Skip the test unless ifname supports HW TX and RX timestamping."""
+    end = "Remote" if host else "Local"
+    ts_info = cmd(f"ethtool -T {ifname}", host=host).stdout
+    if "hardware-transmit" not in ts_info or "hardware-receive" not in ts_info:
+        raise KsftSkipEx(
+            f"{end} device does not support hardware timestamping")
+
+
+def _poll_follower_sync(follower, timeout):
+    """Read the follower stdout pipe until ptp4l reports sync state s2.
+
+    Returns a tuple (synced, output) where output is the text read so far.
+    The synchronized ptp4l log line looks like this:
+    ptp4l[18374.558]: master offset        844 s2 freq    +822 path delay       521
+    """
+    fd_file = follower.proc.stdout
+    fd = fd_file.fileno()
+    buf = b""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if b" s2 " in buf:
+            break
+        if follower.proc.poll() is not None:
+            chunk = fd_file.read()
+            if chunk:
+                buf += chunk
+            break
+        try:
+            remaining = deadline - time.monotonic()
+            buf += fd_read_timeout(fd, max(0, min(1, remaining)))
+        except TimeoutError:
+            continue
+    return b" s2 " in buf, buf.decode("utf-8", "replace")
+
+
+def ptp_sync_test(cfg):
+    """Verify ptp4l leader/follower synchronization reaches state s2."""
+    cfg.require_cmd("ptp4l", remote=True)
+    cfg.require_cmd("ethtool", remote=True)
+
+    _require_hw_timestamping(cfg.ifname)
+    _require_hw_timestamping(cfg.remote_ifname, host=cfg.remote)
+
+    local_dir = cmd("mktemp -d").stdout.strip()
+    remote_dir = cmd("mktemp -d", host=cfg.remote).stdout.strip()
+    defer(cmd, f"rm -rf {local_dir}")
+    defer(cmd, f"rm -rf {remote_dir}", host=cfg.remote)
+
+    leader_cmd = (f"ptp4l -i {cfg.remote_ifname} -m -2 "
+                  f"--uds_address {remote_dir}/ptp4l "
+                  f"--uds_ro_address {remote_dir}/ptp4lro")
+    follower_cmd = (f"ptp4l -i {cfg.ifname} -m -s -2 "
+                    f"--uds_address {local_dir}/ptp4l "
+                    f"--uds_ro_address {local_dir}/ptp4lro")
+
+    # bkg() can't reliably signal the remote ptp4l over SSH (no pty), so
+    # kill it by pattern explicitly in addition to the normal teardown.
+    # pkill exits 1 when ptp4l already terminated, so don't treat it as fatal.
+    defer(cmd, f'pkill -INT -f "{leader_cmd}"', host=cfg.remote, fail=False)
+
+    with bkg(leader_cmd, host=cfg.remote), \
+         bkg(follower_cmd) as follower:
+        synced, output = _poll_follower_sync(follower, PTP4L_SYNC_TIMEOUT)
+
+    if not synced:
+        ksft_pr(f"ptp4l follower did not reach locked state (s2) within "
+                f"{PTP4L_SYNC_TIMEOUT}s")
+        tail = output.strip().split("\n")[-10:]
+        ksft_pr("Follower log (last 10 lines): " + " | ".join(tail))
+    ksft_true(synced, "PTP sync timeout")
+
+
+def main():
+    """Run ksft tests."""
+    with NetDrvEpEnv(__file__) as cfg:
+        ksft_run([ptp_sync_test], args=(cfg, ))
+    ksft_exit()
+
+
+if __name__ == "__main__":
+    main()
-- 
2.56.0