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