]>
Commit | Line | Data |
---|---|---|
7b1bd11c CR |
1 | # Simple functional tests for VNC functionality |
2 | # | |
3 | # Copyright (c) 2018 Red Hat, Inc. | |
4 | # | |
5 | # Author: | |
6 | # Cleber Rosa <crosa@redhat.com> | |
7 | # | |
8 | # This work is licensed under the terms of the GNU GPL, version 2 or | |
9 | # later. See the COPYING file in the top-level directory. | |
10 | ||
0d90da82 VSO |
11 | import socket |
12 | from typing import List | |
13 | ||
2283b627 | 14 | from avocado_qemu import QemuSystemTest |
7b1bd11c CR |
15 | |
16 | ||
0d90da82 VSO |
17 | VNC_ADDR = '127.0.0.1' |
18 | VNC_PORT_START = 32768 | |
19 | VNC_PORT_END = VNC_PORT_START + 1024 | |
20 | ||
21 | ||
22 | def check_bind(port: int) -> bool: | |
23 | with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: | |
24 | try: | |
25 | sock.bind((VNC_ADDR, port)) | |
26 | except OSError: | |
27 | return False | |
28 | ||
29 | return True | |
30 | ||
31 | ||
32 | def check_connect(port: int) -> bool: | |
33 | with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: | |
34 | try: | |
35 | sock.connect((VNC_ADDR, port)) | |
36 | except ConnectionRefusedError: | |
37 | return False | |
38 | ||
39 | return True | |
40 | ||
41 | ||
42 | def find_free_ports(count: int) -> List[int]: | |
43 | result = [] | |
44 | for port in range(VNC_PORT_START, VNC_PORT_END): | |
45 | if check_bind(port): | |
46 | result.append(port) | |
47 | if len(result) >= count: | |
48 | break | |
49 | assert len(result) == count | |
50 | return result | |
51 | ||
52 | ||
2283b627 | 53 | class Vnc(QemuSystemTest): |
7b1bd11c | 54 | """ |
7b1bd11c | 55 | :avocado: tags=vnc,quick |
5104b738 | 56 | :avocado: tags=machine:none |
7b1bd11c CR |
57 | """ |
58 | def test_no_vnc(self): | |
59 | self.vm.add_args('-nodefaults', '-S') | |
60 | self.vm.launch() | |
61 | self.assertFalse(self.vm.qmp('query-vnc')['return']['enabled']) | |
62 | ||
63 | def test_no_vnc_change_password(self): | |
64 | self.vm.add_args('-nodefaults', '-S') | |
65 | self.vm.launch() | |
66 | self.assertFalse(self.vm.qmp('query-vnc')['return']['enabled']) | |
3e3267a9 PB |
67 | set_password_response = self.vm.qmp('change-vnc-password', |
68 | password='new_password') | |
7b1bd11c CR |
69 | self.assertIn('error', set_password_response) |
70 | self.assertEqual(set_password_response['error']['class'], | |
71 | 'GenericError') | |
72 | self.assertEqual(set_password_response['error']['desc'], | |
73 | 'Could not set password') | |
74 | ||
41a7af1c | 75 | def test_change_password_requires_a_password(self): |
7b1bd11c CR |
76 | self.vm.add_args('-nodefaults', '-S', '-vnc', ':0') |
77 | self.vm.launch() | |
78 | self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled']) | |
3e3267a9 PB |
79 | set_password_response = self.vm.qmp('change-vnc-password', |
80 | password='new_password') | |
7b1bd11c CR |
81 | self.assertIn('error', set_password_response) |
82 | self.assertEqual(set_password_response['error']['class'], | |
83 | 'GenericError') | |
84 | self.assertEqual(set_password_response['error']['desc'], | |
85 | 'Could not set password') | |
86 | ||
41a7af1c | 87 | def test_change_password(self): |
0f981d87 | 88 | self.vm.add_args('-nodefaults', '-S', '-vnc', ':0,password=on') |
7b1bd11c CR |
89 | self.vm.launch() |
90 | self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled']) | |
3e3267a9 PB |
91 | set_password_response = self.vm.qmp('change-vnc-password', |
92 | password='new_password') | |
7b1bd11c | 93 | self.assertEqual(set_password_response['return'], {}) |
0d90da82 VSO |
94 | |
95 | def test_change_listen(self): | |
96 | a, b, c = find_free_ports(3) | |
97 | self.assertFalse(check_connect(a)) | |
98 | self.assertFalse(check_connect(b)) | |
99 | self.assertFalse(check_connect(c)) | |
100 | ||
101 | self.vm.add_args('-nodefaults', '-S', '-vnc', f'{VNC_ADDR}:{a - 5900}') | |
102 | self.vm.launch() | |
103 | self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(a)) | |
104 | self.assertTrue(check_connect(a)) | |
105 | self.assertFalse(check_connect(b)) | |
106 | self.assertFalse(check_connect(c)) | |
107 | ||
108 | res = self.vm.qmp('display-update', type='vnc', | |
109 | addresses=[{'type': 'inet', 'host': VNC_ADDR, | |
110 | 'port': str(b)}, | |
111 | {'type': 'inet', 'host': VNC_ADDR, | |
112 | 'port': str(c)}]) | |
113 | self.assertEqual(res['return'], {}) | |
114 | self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(b)) | |
115 | self.assertFalse(check_connect(a)) | |
116 | self.assertTrue(check_connect(b)) | |
117 | self.assertTrue(check_connect(c)) |