On Wed, Dec 04, 2024 at 08:19:08AM +0100, Thomas Huth wrote:
> 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 <th...@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)
I think it would be possible to implement a decorator using "Ports"
such that we don't need to manually wrap the methods, and just receive
the list of port numbers as arguments.
eg to make this pattern with:
@findFreePorts(3)
def test_change_listen(self, ports):
....use 'ports' list ....
Fully untested, but I think something approximately like this would
work:
def findFreePorts(num)
def findFreePortsDeco(func):
def wrapper(*args, **kwargs):
with Ports() as ports:
freeports = ports.find_free_ports(num)
func(freeports, *args, **kwargs)
return wrapper
return findFreePortsDeco
With regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
