Re: [PATCH 2/5] tests/functional: Extract the find_free_ports() function into a helper file

[Daniel P . Berrangé]

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 <thuth@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 :|