mirror of
https://gitlab.com/qemu-project/qemu
synced 2024-09-06 09:55:55 +00:00
avocado/vnc: add test_change_listen
Add simple test-case for new display-update qmp command. Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@openvz.org> Reviewed-by: Daniel P. Berrangé <berrange@redhat.com> Message-Id: <20220401143936.356460-4-vsementsov@openvz.org> Signed-off-by: Gerd Hoffmann <kraxel@redhat.com>
This commit is contained in:
parent
abea19468e
commit
0d90da8203
|
@ -8,9 +8,48 @@
|
|||
# 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 avocado_qemu 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):
|
||||
"""
|
||||
:avocado: tags=vnc,quick
|
||||
|
@ -51,3 +90,27 @@ def test_change_password(self):
|
|||
set_password_response = self.vm.qmp('change-vnc-password',
|
||||
password='new_password')
|
||||
self.assertEqual(set_password_response['return'], {})
|
||||
|
||||
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))
|
||||
|
||||
res = self.vm.qmp('display-update', type='vnc',
|
||||
addresses=[{'type': 'inet', 'host': VNC_ADDR,
|
||||
'port': str(b)},
|
||||
{'type': 'inet', 'host': VNC_ADDR,
|
||||
'port': str(c)}])
|
||||
self.assertEqual(res['return'], {})
|
||||
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))
|
||||
|
|
Loading…
Reference in a new issue