selftests: drv-net: try to check if port is in use

We use random ports for communication. As Willem predicted
this leads to occasional failures. Try to check if port is
already in use by opening a socket and binding to that port.

Reviewed-by: Przemek Kitszel <przemyslaw.kitszel@intel.com>
Reviewed-by: Petr Machata <petrm@nvidia.com>
Reviewed-by: Willem de Bruijn <willemb@google.com>
Link: https://patch.msgid.link/20240626012456.2326192-2-kuba@kernel.org
Signed-off-by: Jakub Kicinski <kuba@kernel.org>

diff --git a/tools/testing/selftests/net/lib/py/utils.py b/tools/testing/selftests/net/lib/py/utils.py
index 0540ea24921d5ef458e77b100d3167b04897c7f8..dd9d2b9f2b20458a68811c95e4e5cf5d09785de4 100644 (file)
--- a/tools/testing/selftests/net/lib/py/utils.py
+++ b/tools/testing/selftests/net/lib/py/utils.py
@@ -1,8 +1,10 @@
 # SPDX-License-Identifier: GPL-2.0
 
+import errno
 import json as _json
 import random
 import re
+import socket
 import subprocess
 import time
 
@@ -79,9 +81,18 @@ def ip(args, json=None, ns=None, host=None):
 
 def rand_port():
     """
-    Get unprivileged port, for now just random, one day we may decide to check if used.
+    Get a random unprivileged port, try to make sure it's not already used.
     """
-    return random.randint(10000, 65535)
+    for _ in range(1000):
+        port = random.randint(10000, 65535)
+        try:
+            with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
+                s.bind(("", port))
+            return port
+        except OSError as e:
+            if e.errno != errno.EADDRINUSE:
+                raise
+    raise Exception("Can't find any free unprivileged port")
 
 
 def wait_port_listen(port, proto="tcp", ns=None, host=None, sleep=0.005, deadline=5):