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)