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: <20241218131439.255841-5-thuth@redhat.com>
Signed-off-by: Thomas Huth <thuth@redhat.com>
This commit is contained in:
Thomas Huth 2024-12-18 14:14:38 +01:00
parent 93a9fdc550
commit b7edbbf432
2 changed files with 64 additions and 28 deletions

View file

@ -14,22 +14,9 @@ import socket
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 @@ class Vnc(QemuSystemTest):
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 @@ class Vnc(QemuSystemTest):
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()