selftests: drv-net: add require_XYZ() helpers for validating env

Wrap typical checks like whether given command used by the test
is available in helpers.

Reviewed-by: Willem de Bruijn <willemb@google.com>
Link: https://lore.kernel.org/r/20240420025237.3309296-8-kuba@kernel.org
Signed-off-by: Jakub Kicinski <kuba@kernel.org>

tools/testing/selftests/drivers/net/lib/py/env.py

@@ -3,7 +3,8 @@
 import os
 import shlex
 from pathlib import Path
-from lib.py import ip
+from lib.py import KsftSkipEx
+from lib.py import cmd, ip
 from lib.py import NetNS, NetdevSimDev
 from .remote import Remote
 
@@ -118,6 +119,8 @@ class NetDrvEpEnv:
         self.ifname = self.dev['ifname']
         self.ifindex = self.dev['ifindex']
 
+        self._required_cmd = {}
+
     def create_local(self):
         self._netns = NetNS()
         self._ns = NetdevSimDev()
@@ -160,3 +163,27 @@ class NetDrvEpEnv:
         if self.remote:
             del self.remote
             self.remote = None
+
+    def require_v4(self):
+        if not self.v4 or not self.remote_v4:
+            raise KsftSkipEx("Test requires IPv4 connectivity")
+
+    def require_v6(self):
+        if not self.v6 or not self.remote_v6:
+            raise KsftSkipEx("Test requires IPv6 connectivity")
+
+    def _require_cmd(self, comm, key, host=None):
+        cached = self._required_cmd.get(comm, {})
+        if cached.get(key) is None:
+            cached[key] = cmd("command -v -- " + comm, fail=False,
+                              shell=True, host=host).ret == 0
+        self._required_cmd[comm] = cached
+        return cached[key]
+
+    def require_cmd(self, comm, local=True, remote=False):
+        if local:
+            if not self._require_cmd(comm, "local"):
+                raise KsftSkipEx("Test requires command: " + comm)
+        if remote:
+            if not self._require_cmd(comm, "remote"):
+                raise KsftSkipEx("Test requires (remote) command: " + comm)

tools/testing/selftests/drivers/net/ping.py

@@ -8,16 +8,22 @@ from lib.py import bkg, cmd, wait_port_listen, rand_port
 
 
 def test_v4(cfg) -> None:
+    cfg.require_v4()
+
     cmd(f"ping -c 1 -W0.5 {cfg.remote_v4}")
     cmd(f"ping -c 1 -W0.5 {cfg.v4}", host=cfg.remote)
 
 
 def test_v6(cfg) -> None:
+    cfg.require_v6()
+
     cmd(f"ping -c 1 -W0.5 {cfg.remote_v6}")
     cmd(f"ping -c 1 -W0.5 {cfg.v6}", host=cfg.remote)
 
 
 def test_tcp(cfg) -> None:
+    cfg.require_cmd("socat", remote=True)
+
     port = rand_port()
     listen_cmd = f"socat -{cfg.addr_ipver} -t 2 -u TCP-LISTEN:{port},reuseport STDOUT"