mirror of
https://github.com/mborgerson/xemu.git
synced 2026-03-21 09:44:38 +00:00
0cdabf8adb
Currently, we have one lock that is held while a test is looking for free ports. However, we are also using different ranges for looking for free ports nowadays (PORTS_START is based on the PID of the process), so instead of using only one lock, we should rather use a lock per range instead. This should help to allow running more tests in parallel. While we're at it, also create the lock files without executable bit (mode is 0o777 by default). Reviewed-by: Daniel P. Berrangé <berrange@redhat.com> Signed-off-by: Thomas Huth <thuth@redhat.com> Message-ID: <20250821094735.804210-1-thuth@redhat.com>
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
#!/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
|
|
|
|
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",
|
|
f".port_lock.{self.PORTS_START}")
|
|
self.lock_fh = os.open(lock_file, os.O_CREAT, mode=0o666)
|
|
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]
|