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

[Thomas Huth <thuth@redhat.com>]

On 04/12/2024 11.41, Daniel P. Berrangé wrote:
> 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

Being mainly a C coder, I think I'd rather avoid getting into too much 
Python magic like such a decorator in the functional tests, to avoid scaring 
the Python ignorants (like I have been in the past) too much ;-)

  Thomas