tests/functional: Extract the find_free_ports() function into a helper file

We'll need this functionality in other functional tests, too, so
let's extract it into the qemu_test module.
Also add  an __enter__ and __exit__ function that can be used for
using this functionality in a locked context, so that tests that
are running in parallel don't try to compete for the same ports
later.
Also make sure to only use ports in the "Dynamic Ports" range
(see https://www.rfc-editor.org/rfc/rfc6335) and "randomize" the
start of the probed range with the PID of the test process to
further avoid possible clashes with other competing processes.

Message-ID: <[email protected]>
Signed-off-by: Thomas Huth <[email protected]>

Author: huth

tests/functional/qemu_test/ports.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+#
+# Simple functional tests for VNC functionality
+#
+# Copyright 2018, 2024 Red Hat, Inc.
+#
+# This work is licensed under the terms of the GNU GPL, version 2 or
+# later.  See the COPYING file in the top-level directory.
+
+import fcntl
+import os
+import socket
+import sys
+import tempfile
+
+from .config import BUILD_DIR
+from typing import List
+
+class Ports():
+
+    PORTS_ADDR = '127.0.0.1'
+    PORTS_RANGE_SIZE = 1024
+    PORTS_START = 49152 + ((os.getpid() * PORTS_RANGE_SIZE) % 16384)
+    PORTS_END = PORTS_START + PORTS_RANGE_SIZE
+
+    def __enter__(self):
+        lock_file = os.path.join(BUILD_DIR, "tests", "functional", "port_lock")
+        self.lock_fh = os.open(lock_file, os.O_CREAT)
+        fcntl.flock(self.lock_fh, fcntl.LOCK_EX)
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        fcntl.flock(self.lock_fh, fcntl.LOCK_UN)
+        os.close(self.lock_fh)
+
+    def check_bind(self, port: int) -> bool:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            try:
+                sock.bind((self.PORTS_ADDR, port))
+            except OSError:
+                return False
+
+        return True
+
+    def find_free_ports(self, count: int) -> List[int]:
+        result = []
+        for port in range(self.PORTS_START, self.PORTS_END):
+            if self.check_bind(port):
+                result.append(port)
+                if len(result) >= count:
+                    break
+        assert len(result) == count
+        return result
+
+    def find_free_port(self) -> int:
+        return self.find_free_ports(1)[0]

tests/functional/test_vnc.py

@@ -14,22 +14,9 @@
 from typing import List
 
 from qemu_test import QemuSystemTest
-
+from qemu_test.ports import Ports
 
 VNC_ADDR = '127.0.0.1'
-VNC_PORT_START = 32768
-VNC_PORT_END = VNC_PORT_START + 1024
-
-
-def check_bind(port: int) -> bool:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-        try:
-            sock.bind((VNC_ADDR, port))
-        except OSError:
-            return False
-
-    return True
-
 
 def check_connect(port: int) -> bool:
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
@@ -40,18 +27,6 @@ def check_connect(port: int) -> bool:
 
     return True
 
-
-def find_free_ports(count: int) -> List[int]:
-    result = []
-    for port in range(VNC_PORT_START, VNC_PORT_END):
-        if check_bind(port):
-            result.append(port)
-            if len(result) >= count:
-                break
-    assert len(result) == count
-    return result
-
-
 class Vnc(QemuSystemTest):
 
     def test_no_vnc_change_password(self):
@@ -85,8 +60,7 @@ def test_change_password(self):
         self.vm.cmd('change-vnc-password',
                     password='new_password')
 
-    def test_change_listen(self):
-        a, b, c = find_free_ports(3)
+    def do_test_change_listen(self, a, b, c):
         self.assertFalse(check_connect(a))
         self.assertFalse(check_connect(b))
         self.assertFalse(check_connect(c))
@@ -108,5 +82,11 @@ def test_change_listen(self):
         self.assertTrue(check_connect(b))
         self.assertTrue(check_connect(c))
 
+    def test_change_listen(self):
+        with Ports() as ports:
+            a, b, c = ports.find_free_ports(3)
+            self.do_test_change_listen(a, b, c)
+
+
 if __name__ == '__main__':
     QemuSystemTest.main()