From c82bfaf42dc6f8dbf101190c86cfa8af9ea400fd Mon Sep 17 00:00:00 2001 From: Thomas Huth Date: Wed, 18 Dec 2024 14:14:35 +0100 Subject: tests/functional: Convert the vnc test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing thrilling in here, it's just a straight forward conversion. Reviewed-by: Philippe Mathieu-Daudé Tested-by: Philippe Mathieu-Daudé Message-ID: <20241218131439.255841-2-thuth@redhat.com> Signed-off-by: Thomas Huth --- tests/functional/test_vnc.py | 117 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 117 insertions(+) create mode 100755 tests/functional/test_vnc.py (limited to 'tests/functional/test_vnc.py') diff --git a/tests/functional/test_vnc.py b/tests/functional/test_vnc.py new file mode 100755 index 0000000000..b769d3b268 --- /dev/null +++ b/tests/functional/test_vnc.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +# +# Simple functional tests for VNC functionality +# +# Copyright (c) 2018 Red Hat, Inc. +# +# Author: +# Cleber Rosa +# +# 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 socket +from typing import List + +from qemu_test import QemuSystemTest + + +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: + try: + sock.connect((VNC_ADDR, port)) + except ConnectionRefusedError: + return False + + 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): + self.vm.add_args('-nodefaults', '-S') + self.vm.launch() + self.assertFalse(self.vm.qmp('query-vnc')['return']['enabled']) + + def test_no_vnc_change_password(self): + self.vm.add_args('-nodefaults', '-S') + self.vm.launch() + self.assertFalse(self.vm.qmp('query-vnc')['return']['enabled']) + set_password_response = self.vm.qmp('change-vnc-password', + password='new_password') + self.assertIn('error', set_password_response) + self.assertEqual(set_password_response['error']['class'], + 'GenericError') + self.assertEqual(set_password_response['error']['desc'], + 'Could not set password') + + def test_change_password_requires_a_password(self): + self.vm.add_args('-nodefaults', '-S', '-vnc', ':0') + self.vm.launch() + self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled']) + set_password_response = self.vm.qmp('change-vnc-password', + password='new_password') + self.assertIn('error', set_password_response) + self.assertEqual(set_password_response['error']['class'], + 'GenericError') + self.assertEqual(set_password_response['error']['desc'], + 'Could not set password') + + def test_change_password(self): + self.vm.add_args('-nodefaults', '-S', '-vnc', ':0,password=on') + self.vm.launch() + self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled']) + self.vm.cmd('change-vnc-password', + password='new_password') + + def test_change_listen(self): + a, b, c = find_free_ports(3) + self.assertFalse(check_connect(a)) + self.assertFalse(check_connect(b)) + self.assertFalse(check_connect(c)) + + self.vm.add_args('-nodefaults', '-S', '-vnc', f'{VNC_ADDR}:{a - 5900}') + self.vm.launch() + self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(a)) + self.assertTrue(check_connect(a)) + self.assertFalse(check_connect(b)) + self.assertFalse(check_connect(c)) + + self.vm.cmd('display-update', type='vnc', + addresses=[{'type': 'inet', 'host': VNC_ADDR, + 'port': str(b)}, + {'type': 'inet', 'host': VNC_ADDR, + 'port': str(c)}]) + self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(b)) + self.assertFalse(check_connect(a)) + self.assertTrue(check_connect(b)) + self.assertTrue(check_connect(c)) + +if __name__ == '__main__': + QemuSystemTest.main() -- cgit 1.4.1 From 56d3a1482921e7e23233f3abcce9c29f3f56cb72 Mon Sep 17 00:00:00 2001 From: Thomas Huth Date: Wed, 18 Dec 2024 14:14:37 +0100 Subject: tests/functional/test_vnc: Do not use a hard-coded VNC port Two tests here are using the hard-coded VNC port :0 ... if there is already a QEMU or other program running that is using this port, the tests will be failing. Fortunately, QEMU can also auto-detect a free port with the "to=..." parameter, so let's use that for the tests to avoid the problem. Message-ID: <20241218131439.255841-4-thuth@redhat.com> Signed-off-by: Thomas Huth --- tests/functional/test_vnc.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'tests/functional/test_vnc.py') diff --git a/tests/functional/test_vnc.py b/tests/functional/test_vnc.py index b769d3b268..e6328567c7 100755 --- a/tests/functional/test_vnc.py +++ b/tests/functional/test_vnc.py @@ -72,7 +72,7 @@ class Vnc(QemuSystemTest): 'Could not set password') def test_change_password_requires_a_password(self): - self.vm.add_args('-nodefaults', '-S', '-vnc', ':0') + self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999') self.vm.launch() self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled']) set_password_response = self.vm.qmp('change-vnc-password', @@ -84,7 +84,7 @@ class Vnc(QemuSystemTest): 'Could not set password') def test_change_password(self): - self.vm.add_args('-nodefaults', '-S', '-vnc', ':0,password=on') + self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999,password=on') self.vm.launch() self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled']) self.vm.cmd('change-vnc-password', -- cgit 1.4.1 From 93a9fdc5504f15d319927c1497522cb72929d78e Mon Sep 17 00:00:00 2001 From: Thomas Huth Date: Wed, 18 Dec 2024 14:14:36 +0100 Subject: tests/functional/test_vnc: Remove the test_no_vnc test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This test matches exactly the first three lines of the following test_no_vnc_change_password test, so there is exactly zero additional test coverage in here. Reviewed-by: Daniel P. Berrangé Message-ID: <20241218131439.255841-3-thuth@redhat.com> Signed-off-by: Thomas Huth --- tests/functional/test_vnc.py | 5 ----- 1 file changed, 5 deletions(-) (limited to 'tests/functional/test_vnc.py') diff --git a/tests/functional/test_vnc.py b/tests/functional/test_vnc.py index e6328567c7..e600d75234 100755 --- a/tests/functional/test_vnc.py +++ b/tests/functional/test_vnc.py @@ -54,11 +54,6 @@ def find_free_ports(count: int) -> List[int]: class Vnc(QemuSystemTest): - def test_no_vnc(self): - self.vm.add_args('-nodefaults', '-S') - self.vm.launch() - self.assertFalse(self.vm.qmp('query-vnc')['return']['enabled']) - def test_no_vnc_change_password(self): self.vm.add_args('-nodefaults', '-S') self.vm.launch() -- cgit 1.4.1 From b7edbbf4321fea9efefda2a5d6bcea4f7140f866 Mon Sep 17 00:00:00 2001 From: Thomas Huth Date: Wed, 18 Dec 2024 14:14:38 +0100 Subject: tests/functional: Extract the find_free_ports() function into a helper file 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. Also make sure to only use ports in the "Dynamic Ports" range (see https://www.rfc-editor.org/rfc/rfc6335) and "randomize" the start of the probed range with the PID of the test process to further avoid possible clashes with other competing processes. Message-ID: <20241218131439.255841-5-thuth@redhat.com> Signed-off-by: Thomas Huth --- tests/functional/qemu_test/ports.py | 56 +++++++++++++++++++++++++++++++++++++ tests/functional/test_vnc.py | 36 ++++++------------------ 2 files changed, 64 insertions(+), 28 deletions(-) create mode 100644 tests/functional/qemu_test/ports.py (limited to 'tests/functional/test_vnc.py') diff --git a/tests/functional/qemu_test/ports.py b/tests/functional/qemu_test/ports.py new file mode 100644 index 0000000000..cc39939d48 --- /dev/null +++ b/tests/functional/qemu_test/ports.py @@ -0,0 +1,56 @@ +#!/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 .config import BUILD_DIR +from typing import List + +class Ports(): + + PORTS_ADDR = '127.0.0.1' + PORTS_RANGE_SIZE = 1024 + PORTS_START = 49152 + ((os.getpid() * PORTS_RANGE_SIZE) % 16384) + PORTS_END = PORTS_START + PORTS_RANGE_SIZE + + def __enter__(self): + lock_file = os.path.join(BUILD_DIR, "tests", "functional", "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 e600d75234..1916be0103 100755 --- a/tests/functional/test_vnc.py +++ b/tests/functional/test_vnc.py @@ -14,22 +14,9 @@ import socket 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_change_password(self): @@ -85,8 +60,7 @@ class Vnc(QemuSystemTest): 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)) @@ -108,5 +82,11 @@ class Vnc(QemuSystemTest): 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() -- cgit 1.4.1