xref: /qemu/tests/functional/qemu_test/ports.py (revision 513823e7521a09ed7ad1e32e6454bac3b2cbf52d)
1#!/usr/bin/env python3
2#
3# Simple functional tests for VNC functionality
4#
5# Copyright 2018, 2024 Red Hat, Inc.
6#
7# This work is licensed under the terms of the GNU GPL, version 2 or
8# later.  See the COPYING file in the top-level directory.
9
10import fcntl
11import os
12import socket
13import sys
14import tempfile
15
16from .config import BUILD_DIR
17from typing import List
18
19class Ports():
20
21    PORTS_ADDR = '127.0.0.1'
22    PORTS_RANGE_SIZE = 1024
23    PORTS_START = 49152 + ((os.getpid() * PORTS_RANGE_SIZE) % 16384)
24    PORTS_END = PORTS_START + PORTS_RANGE_SIZE
25
26    def __enter__(self):
27        lock_file = os.path.join(BUILD_DIR, "tests", "functional", "port_lock")
28        self.lock_fh = os.open(lock_file, os.O_CREAT)
29        fcntl.flock(self.lock_fh, fcntl.LOCK_EX)
30        return self
31
32    def __exit__(self, exc_type, exc_value, traceback):
33        fcntl.flock(self.lock_fh, fcntl.LOCK_UN)
34        os.close(self.lock_fh)
35
36    def check_bind(self, port: int) -> bool:
37        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
38            try:
39                sock.bind((self.PORTS_ADDR, port))
40            except OSError:
41                return False
42
43        return True
44
45    def find_free_ports(self, count: int) -> List[int]:
46        result = []
47        for port in range(self.PORTS_START, self.PORTS_END):
48            if self.check_bind(port):
49                result.append(port)
50                if len(result) >= count:
51                    break
52        assert len(result) == count
53        return result
54
55    def find_free_port(self) -> int:
56        return self.find_free_ports(1)[0]
57