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)
+
+
if __name__ == '__main__':
QemuSystemTest.main()
--
2.47.0On 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)
This code will leave '/tmp/qemu_port_lock' existing forever.... which is
correct, because if you try to unlink it after closing, you'll introduce
a race because the 2nd __enter__ will now be locking an unlinked file,
and a 3rd __enter__ that comes along will create & lock an entirely new
file.
There are ways to make this safe by using stat + fstat either side of
LOCK_EX, in a loop, to detect locking of an unlinked file. That is
overkill though. It is simpler to just put the lock file in the build
directory IMHO, and thus avoid needing to care about unlinking - that'll
be done when the user purges their build dir.
> +
> + 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)
> +
> +
> if __name__ == '__main__':
> QemuSystemTest.main()
> --
> 2.47.0
>
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 :|
On 04/12/2024 11.46, 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) > > This code will leave '/tmp/qemu_port_lock' existing forever.... which is > correct, because if you try to unlink it after closing, you'll introduce > a race because the 2nd __enter__ will now be locking an unlinked file, > and a 3rd __enter__ that comes along will create & lock an entirely new > file. > > There are ways to make this safe by using stat + fstat either side of > LOCK_EX, in a loop, to detect locking of an unlinked file. That is > overkill though. It is simpler to just put the lock file in the build > directory IMHO, and thus avoid needing to care about unlinking - that'll > be done when the user purges their build dir. Putting the lock in the build directory is a nice idea - but it will fail if the user is using multiple build directory and running the test in the multiple directories in parallel. But maybe we could ease that situation by randomizing PORTS_START based on a seed calculated somehow by the build directory string? Thomas
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 :|
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
© 2016 - 2025 Red Hat, Inc.