xref: /qemu/tests/functional/test_vnc.py (revision d82bb3f5dd5647e0f470b4189096aced1447b09f) 
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.set_machine('none')
35        self.vm.add_args('-nodefaults', '-S')
36        self.vm.launch()
37
38        query_vnc_response = self.vm.qmp('query-vnc')
39        if 'error' in query_vnc_response:
40            self.assertEqual(query_vnc_response['error']['class'],
41                             'CommandNotFound')
42            self.skipTest('VNC support not available')
43        self.assertFalse(query_vnc_response['return']['enabled'])
44
45        set_password_response = self.vm.qmp('change-vnc-password',
46                                            password='new_password')
47        self.assertIn('error', set_password_response)
48        self.assertEqual(set_password_response['error']['class'],
49                         'GenericError')
50        self.assertEqual(set_password_response['error']['desc'],
51                         'Could not set password')
52
53    def launch_guarded(self):
54        try:
55            self.vm.launch()
56        except VMLaunchFailure as excp:
57            if "-vnc: invalid option" in excp.output:
58                self.skipTest("VNC support not available")
59            elif "Cipher backend does not support DES algorithm" in excp.output:
60                self.skipTest("No cryptographic backend available")
61            else:
62                self.log.info("unhandled launch failure: %s", excp.output)
63                raise excp
64
65    def test_change_password_requires_a_password(self):
66        self.set_machine('none')
67        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999')
68        self.launch_guarded()
69        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
70        set_password_response = self.vm.qmp('change-vnc-password',
71                                            password='new_password')
72        self.assertIn('error', set_password_response)
73        self.assertEqual(set_password_response['error']['class'],
74                         'GenericError')
75        self.assertEqual(set_password_response['error']['desc'],
76                         'Could not set password')
77
78    def test_change_password(self):
79        self.set_machine('none')
80        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999,password=on')
81        self.launch_guarded()
82        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
83        self.vm.cmd('change-vnc-password',
84                    password='new_password')
85
86    def do_test_change_listen(self, a, b, c):
87        self.assertFalse(check_connect(a))
88        self.assertFalse(check_connect(b))
89        self.assertFalse(check_connect(c))
90
91        self.vm.add_args('-nodefaults', '-S', '-vnc', f'{VNC_ADDR}:{a - 5900}')
92        self.launch_guarded()
93        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(a))
94        self.assertTrue(check_connect(a))
95        self.assertFalse(check_connect(b))
96        self.assertFalse(check_connect(c))
97
98        self.vm.cmd('display-update', type='vnc',
99                    addresses=[{'type': 'inet', 'host': VNC_ADDR,
100                                'port': str(b)},
101                               {'type': 'inet', 'host': VNC_ADDR,
102                                'port': str(c)}])
103        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(b))
104        self.assertFalse(check_connect(a))
105        self.assertTrue(check_connect(b))
106        self.assertTrue(check_connect(c))
107
108    def test_change_listen(self):
109        self.set_machine('none')
110        with Ports() as ports:
111            a, b, c = ports.find_free_ports(3)
112            self.do_test_change_listen(a, b, c)
113
114
115if __name__ == '__main__':
116    QemuSystemTest.main()
117