[PATCH 2/5] tests/functional: Extract the find_free_ports() function into a helper file

[Thomas Huth]

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.

Signed-off-by: Thomas Huth <thuth@redhat.com>
---
 tests/functional/qemu_test/ports.py | 53 +++++++++++++++++++++++++++++
 tests/functional/test_vnc.py        | 36 +++++---------------
 2 files changed, 61 insertions(+), 28 deletions(-)
 create mode 100644 tests/functional/qemu_test/ports.py

diff --git a/tests/functional/qemu_test/ports.py 
b/tests/functional/qemu_test/ports.py
new file mode 100644
index 0000000000..d235d3432b
--- /dev/null
+++ b/tests/functional/qemu_test/ports.py
@@ -0,0 +1,53 @@
+#!/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 typing import List
+
+class Ports():
+
+    PORTS_ADDR = '127.0.0.1'
+    PORTS_START = 32768
+    PORTS_END = PORTS_START + 1024
+
+    def __enter__(self):
+        lock_file = os.path.join(tempfile.gettempdir(), "qemu_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]
diff --git a/tests/functional/test_vnc.py b/tests/functional/test_vnc.py
index b769d3b268..32a81259e4 100755
--- a/tests/functional/test_vnc.py
+++ b/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(self):
@@ -90,8 +65,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))
@@ -113,5 +87,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()
-- 
2.47.0