xref: /qemu/tests/functional/test_vnc.py (revision b103cc6e74ac92f070a0e004bd84334e845c20b5)
1#!/usr/bin/env python3
2#
3# Simple functional tests for VNC functionality
4#
5# Copyright (c) 2018 Red Hat, Inc.
6#
7# Author:
8#  Cleber Rosa <crosa@redhat.com>
9#
10# This work is licensed under the terms of the GNU GPL, version 2 or
11# later.  See the COPYING file in the top-level directory.
12
13import socket
14
15from qemu.machine.machine import VMLaunchFailure
16from qemu_test import QemuSystemTest
17from qemu_test.ports import Ports
18
19
20VNC_ADDR = '127.0.0.1'
21
22def check_connect(port: int) -> bool:
23    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
24        try:
25            sock.connect((VNC_ADDR, port))
26        except ConnectionRefusedError:
27            return False
28
29    return True
30
31class Vnc(QemuSystemTest):
32
33    def test_no_vnc_change_password(self):
34        self.vm.add_args('-nodefaults', '-S')
35        self.vm.launch()
36
37        query_vnc_response = self.vm.qmp('query-vnc')
38        if 'error' in query_vnc_response:
39            self.assertEqual(query_vnc_response['error']['class'],
40                             'CommandNotFound')
41            self.skipTest('VNC support not available')
42        self.assertFalse(query_vnc_response['return']['enabled'])
43
44        set_password_response = self.vm.qmp('change-vnc-password',
45                                            password='new_password')
46        self.assertIn('error', set_password_response)
47        self.assertEqual(set_password_response['error']['class'],
48                         'GenericError')
49        self.assertEqual(set_password_response['error']['desc'],
50                         'Could not set password')
51
52    def launch_guarded(self):
53        try:
54            self.vm.launch()
55        except VMLaunchFailure as excp:
56            if "-vnc: invalid option" in excp.output:
57                self.skipTest("VNC support not available")
58            elif "Cipher backend does not support DES algorithm" in excp.output:
59                self.skipTest("No cryptographic backend available")
60            else:
61                self.log.info("unhandled launch failure: %s", excp.output)
62                raise excp
63
64    def test_change_password_requires_a_password(self):
65        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999')
66        self.launch_guarded()
67        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
68        set_password_response = self.vm.qmp('change-vnc-password',
69                                            password='new_password')
70        self.assertIn('error', set_password_response)
71        self.assertEqual(set_password_response['error']['class'],
72                         'GenericError')
73        self.assertEqual(set_password_response['error']['desc'],
74                         'Could not set password')
75
76    def test_change_password(self):
77        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999,password=on')
78        self.launch_guarded()
79        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
80        self.vm.cmd('change-vnc-password',
81                    password='new_password')
82
83    def do_test_change_listen(self, a, b, c):
84        self.assertFalse(check_connect(a))
85        self.assertFalse(check_connect(b))
86        self.assertFalse(check_connect(c))
87
88        self.vm.add_args('-nodefaults', '-S', '-vnc', f'{VNC_ADDR}:{a - 5900}')
89        self.launch_guarded()
90        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(a))
91        self.assertTrue(check_connect(a))
92        self.assertFalse(check_connect(b))
93        self.assertFalse(check_connect(c))
94
95        self.vm.cmd('display-update', type='vnc',
96                    addresses=[{'type': 'inet', 'host': VNC_ADDR,
97                                'port': str(b)},
98                               {'type': 'inet', 'host': VNC_ADDR,
99                                'port': str(c)}])
100        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(b))
101        self.assertFalse(check_connect(a))
102        self.assertTrue(check_connect(b))
103        self.assertTrue(check_connect(c))
104
105    def test_change_listen(self):
106        with Ports() as ports:
107            a, b, c = ports.find_free_ports(3)
108            self.do_test_change_listen(a, b, c)
109
110
111if __name__ == '__main__':
112    QemuSystemTest.main()
113