xref: /qemu/tests/functional/test_vnc.py (revision c07cd110a1824e2d046581af7375f16dac26e96f)
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
14from typing import List
15from qemu.machine.machine import VMLaunchFailure
16
17from qemu_test import QemuSystemTest
18from qemu_test.ports import Ports
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            else:
59                self.log.info("unhandled launch failure: %s", excp.output)
60                raise excp
61
62    def test_change_password_requires_a_password(self):
63        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999')
64        self.launch_guarded()
65        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
66        set_password_response = self.vm.qmp('change-vnc-password',
67                                            password='new_password')
68        self.assertIn('error', set_password_response)
69        self.assertEqual(set_password_response['error']['class'],
70                         'GenericError')
71        self.assertEqual(set_password_response['error']['desc'],
72                         'Could not set password')
73
74    def test_change_password(self):
75        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999,password=on')
76        self.launch_guarded()
77        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
78        self.vm.cmd('change-vnc-password',
79                    password='new_password')
80
81    def do_test_change_listen(self, a, b, c):
82        self.assertFalse(check_connect(a))
83        self.assertFalse(check_connect(b))
84        self.assertFalse(check_connect(c))
85
86        self.vm.add_args('-nodefaults', '-S', '-vnc', f'{VNC_ADDR}:{a - 5900}')
87        self.launch_guarded()
88        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(a))
89        self.assertTrue(check_connect(a))
90        self.assertFalse(check_connect(b))
91        self.assertFalse(check_connect(c))
92
93        self.vm.cmd('display-update', type='vnc',
94                    addresses=[{'type': 'inet', 'host': VNC_ADDR,
95                                'port': str(b)},
96                               {'type': 'inet', 'host': VNC_ADDR,
97                                'port': str(c)}])
98        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(b))
99        self.assertFalse(check_connect(a))
100        self.assertTrue(check_connect(b))
101        self.assertTrue(check_connect(c))
102
103    def test_change_listen(self):
104        with Ports() as ports:
105            a, b, c = ports.find_free_ports(3)
106            self.do_test_change_listen(a, b, c)
107
108
109if __name__ == '__main__':
110    QemuSystemTest.main()
111