| 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 | |
| 10 | import fcntl |
| 11 | import os |
| 12 | import socket |
| 13 | |
| 14 | from .config import BUILD_DIR |
| 15 | from typing import List |
| 16 | |
| 17 | |
| 18 | class Ports(): |
| 19 | |
| 20 | PORTS_ADDR = '127.0.0.1' |
| 21 | PORTS_RANGE_SIZE = 1024 |
| 22 | PORTS_START = 49152 + ((os.getpid() * PORTS_RANGE_SIZE) % 16384) |
| 23 | PORTS_END = PORTS_START + PORTS_RANGE_SIZE |
| 24 | |
| 25 | def __enter__(self): |
| 26 | lock_file = os.path.join(BUILD_DIR, "tests", "functional", |
| 27 | f".port_lock.{self.PORTS_START}") |
| 28 | self.lock_fh = os.open(lock_file, os.O_CREAT, mode=0o666) |
| 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] |