[Devel] [PATCH hci-8.1 v2 2/3] selftests: drv-net: run ping in its own netns

[Eva Kurchatova]

In netdevsim mode ping.py sends its traffic to itself through the host
namespace, so a host firewall that drops it fails the test:

  ok 1 ping.test_v4
  ok 2 ping.test_v6
  not ok 3 ping.test_tcp
  # Stopping tests due to KsftTerminate.

Re-exec the test in a namespace of its own, the way nft_audit.sh does
since commit 2e2d9c7d4d37 ("selftests: netfilter: Run nft_audit.sh in
its own netns"). A test given a real interface through NETIF stays
where it is, that interface is not reachable from a new namespace.

https://virtuozzo.atlassian.net/browse/VSTOR-139651
Feature: fix selftests
Signed-off-by: Eva Kurchatova <eva.kurchatova at virtuozzo.com>
---
 .../selftests/drivers/net/lib/py/__init__.py  |  2 +-
 tools/testing/selftests/drivers/net/ping.py   |  4 ++
 tools/testing/selftests/net/lib/py/utils.py   | 44 +++++++++++++++++++
 3 files changed, 49 insertions(+), 1 deletion(-)

diff --git a/tools/testing/selftests/drivers/net/lib/py/__init__.py b/tools/testing/selftests/drivers/net/lib/py/__init__.py
index fce5d9218f1d..2d3fdaa81550 100644
--- a/tools/testing/selftests/drivers/net/lib/py/__init__.py
+++ b/tools/testing/selftests/drivers/net/lib/py/__init__.py
@@ -15,7 +15,7 @@ try:
         NlError, RtnlFamily, DevlinkFamily
     from net.lib.py import CmdExitFailure
     from net.lib.py import bkg, cmd, defer, ethtool, fd_read_timeout, ip, \
-        rand_port, tool, wait_port_listen
+        rand_port, rexec_in_netns, tool, wait_port_listen
     from net.lib.py import fd_read_timeout
     from net.lib.py import KsftSkipEx, KsftFailEx, KsftXfailEx
     from net.lib.py import ksft_disruptive, ksft_exit, ksft_pr, ksft_run, \
diff --git a/tools/testing/selftests/drivers/net/ping.py b/tools/testing/selftests/drivers/net/ping.py
index eb83e7b48797..46da331158ba 100755
--- a/tools/testing/selftests/drivers/net/ping.py
+++ b/tools/testing/selftests/drivers/net/ping.py
@@ -5,6 +5,7 @@ from lib.py import ksft_run, ksft_exit
 from lib.py import ksft_eq
 from lib.py import NetDrvEpEnv
 from lib.py import bkg, cmd, wait_port_listen, rand_port
+from lib.py import rexec_in_netns
 
 
 def test_v4(cfg) -> None:
@@ -42,6 +43,9 @@ def test_tcp(cfg) -> None:
 
 
 def main() -> None:
+    # In netdevsim mode both ends would sit in the host's namespace
+    rexec_in_netns(__file__)
+
     with NetDrvEpEnv(__file__) as cfg:
         ksft_run(globs=globals(), case_pfx={"test_"}, args=(cfg, ))
     ksft_exit()
diff --git a/tools/testing/selftests/net/lib/py/utils.py b/tools/testing/selftests/net/lib/py/utils.py
index afce547bcab0..ab91358cca9c 100644
--- a/tools/testing/selftests/net/lib/py/utils.py
+++ b/tools/testing/selftests/net/lib/py/utils.py
@@ -8,6 +8,7 @@ import re
 import select
 import socket
 import subprocess
+import sys
 import time
 
 
@@ -131,6 +132,49 @@ def ethtool(args, json=None, ns=None, host=None):
     return tool('ethtool', args, json=json, ns=ns, host=host)
 
 
+def _netif_configured(path):
+    """
+    Whether a real interface was named, in the environment or net.config.
+    """
+    if os.environ.get("NETIF"):
+        return True
+
+    cfg = os.path.join(os.path.dirname(os.path.abspath(path)), "net.config")
+    try:
+        with open(cfg, 'r') as fp:
+            for line in fp.readlines():
+                pos = line.find("#")
+                if pos >= 0:
+                    line = line[:pos]
+                pair = line.strip().split('=', maxsplit=1)
+                if len(pair) == 2 and pair[0] == "NETIF" and pair[1]:
+                    return True
+    except OSError:
+        pass
+    return False
+
+
+def rexec_in_netns(path):
+    """
+    Re-exec the test in a network namespace of its own, with sysfs
+    mounted again. Does nothing where a real interface is named, or
+    where unshare is not there to move the test.
+    """
+    if os.environ.get("KSFT_IN_NETNS") or _netif_configured(path):
+        return
+
+    os.environ["KSFT_IN_NETNS"] = "1"
+    try:
+        os.execvp("unshare",
+                  ["unshare", "-n", "-m",
+                   "sh", "-c",
+                   'mount -t sysfs sysfs /sys && ip link set lo up && '
+                   'exec "$@"',
+                   "sh", sys.executable, os.path.abspath(path)] + sys.argv[1:])
+    except OSError:
+        del os.environ["KSFT_IN_NETNS"]
+
+
 def rand_port(type=socket.SOCK_STREAM):
     """
     Get a random unprivileged port.
-- 
2.55.0