import json import socket import tempfile import threading import time import unittest from pathlib import Path from hydra.qmp import QMPClient, QMPError class QMPServer: def __init__(self, path, handler): self.socket = socket.socket(socket.AF_UNIX) self.socket.bind(str(path)) self.socket.listen(4) self.socket.settimeout(0.1) self.handler = handler self.stop = threading.Event() self.errors = [] self.thread = threading.Thread(target=self.run, daemon=True) self.thread.start() def run(self): while not self.stop.is_set(): try: conn, _ = self.socket.accept() except socket.timeout: continue except OSError: break try: with conn: conn.settimeout(1) self.handler(conn) except (BrokenPipeError, ConnectionResetError, socket.timeout): pass except Exception as exc: self.errors.append(exc) def close(self): self.stop.set() self.socket.close() self.thread.join(2) def send(conn, message): conn.sendall(json.dumps(message).encode() + b'\r\n') def greeting(conn): send(conn, {'QMP': {'version': {'qemu': {'major': 10, 'minor': 0, 'micro': 0}, 'package': ''}, 'capabilities': []}}) class QMPTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix='hq-') self.addCleanup(self.tmp.cleanup) self.path = Path(self.tmp.name) / 'qmp' def server(self, handler): srv = QMPServer(self.path, handler) self.addCleanup(srv.close) return srv def test_handshake_ids_events_and_return(self): commands = [] def handler(conn): greeting(conn) stream = conn.makefile('rb') for index in range(2): request = json.loads(stream.readline()) commands.append(request) send(conn, {'event': 'STOP', 'data': {}}) send(conn, {'return': {'ignored': True}, 'id': 'unrelated'}) send(conn, {'return': {} if index == 0 else {'status': 'running'}, 'id': request['id']}) self.server(handler) with QMPClient(self.path, timeout=0.5) as client: self.assertEqual(client.command('query-status'), {'status': 'running'}) self.assertEqual([cmd['execute'] for cmd in commands], ['qmp_capabilities', 'query-status']) self.assertNotEqual(commands[0]['id'], commands[1]['id']) def test_command_error(self): def handler(conn): greeting(conn) stream = conn.makefile('rb') for index in range(2): request = json.loads(stream.readline()) if index == 0: send(conn, {'return': {}, 'id': request['id']}) else: send(conn, {'error': {'class': 'CommandNotFound', 'desc': 'unsupported'}, 'id': request['id']}) self.server(handler) with QMPClient(self.path, timeout=0.5) as client, self.assertRaises(QMPError): client.command('missing') def test_greeting_timeout_is_bounded(self): self.server(lambda conn: time.sleep(0.3)) started = time.monotonic() with self.assertRaises(QMPError): QMPClient(self.path, timeout=0.08) self.assertLess(time.monotonic() - started, 0.25) def test_event_stream_cannot_extend_command_deadline(self): def handler(conn): greeting(conn) request = json.loads(conn.makefile('rb').readline()) while True: send(conn, {'event': 'STOP', 'data': {}}) time.sleep(0.01) self.server(handler) started = time.monotonic() with self.assertRaises(QMPError): QMPClient(self.path, timeout=0.08) self.assertLess(time.monotonic() - started, 0.25) def test_malformed_greeting_rejected(self): self.server(lambda conn: send(conn, {'return': {}})) with self.assertRaises(QMPError): QMPClient(self.path) def test_command_timeout_after_successful_handshake(self): def handler(conn): greeting(conn) with conn.makefile('rb') as stream: request = json.loads(stream.readline()) send(conn, {'return': {}, 'id': request['id']}) stream.readline() time.sleep(0.3) self.server(handler) with QMPClient(self.path, timeout=0.08) as client: started = time.monotonic() with self.assertRaises(QMPError): client.command('query-status') self.assertLess(time.monotonic() - started, 0.25) def test_spice_events_before_greeting_still_negotiate_and_query_identity(self): commands = [] def handler(conn): send(conn, {'event': 'SPICE_CONNECTED', 'data': {'server': {'family': 'unix'}}, 'timestamp': {'seconds': 1791919209, 'microseconds': 650048}}) send(conn, {'event': 'SPICE_DISCONNECTED', 'timestamp': {'seconds': -1, 'microseconds': -1}}) greeting(conn) with conn.makefile('rb') as stream: for name in ('qmp_capabilities', 'query-name'): request = json.loads(stream.readline()) commands.append(request['execute']) send(conn, {'return': {} if name == 'qmp_capabilities' else {'name': 'owned'}, 'id': request['id']}) self.server(handler) with QMPClient(self.path, timeout=0.5) as client: self.assertEqual(client.command('query-name'), {'name': 'owned'}) self.assertEqual(commands, ['qmp_capabilities', 'query-name']) def test_before_greeting_rejects_replies_and_malformed_events(self): messages = [ {}, {'return': {}}, {'error': {}}, {'event': 'SPICE_CONNECTED'}, {'event': '', 'timestamp': {'seconds': 1, 'microseconds': 0}}, {'event': 1, 'timestamp': {'seconds': 1, 'microseconds': 0}}, {'event': 'STOP', 'data': [], 'timestamp': {'seconds': 1, 'microseconds': 0}}, {'event': 'STOP', 'timestamp': {'seconds': True, 'microseconds': 0}}, {'event': 'STOP', 'timestamp': {'seconds': 1, 'microseconds': 1000000}}, {'event': 'STOP', 'timestamp': {'seconds': 1, 'microseconds': -1}}, {'event': 'STOP', 'timestamp': {'seconds': 1}}, {'event': 'STOP', 'timestamp': {'seconds': 1, 'microseconds': 0}, 'return': {}}, ] for message in messages: with self.subTest(message=message): self.path.unlink(missing_ok=True) server = self.server(lambda conn, message=message: (send(conn, message), greeting(conn))) with self.assertRaisesRegex(QMPError, 'Invalid QMP'): QMPClient(self.path, timeout=0.2) server.close() def test_before_greeting_event_flood_cannot_extend_handshake_deadline(self): def handler(conn): while True: send(conn, {'event': 'SPICE_CONNECTED', 'timestamp': {'seconds': 1, 'microseconds': 0}}) time.sleep(0.005) self.server(handler) started = time.monotonic() with self.assertRaisesRegex(QMPError, 'deadline|receive'): QMPClient(self.path, timeout=0.08) elapsed = time.monotonic() - started self.assertGreaterEqual(elapsed, 0.06) self.assertLess(elapsed, 0.25) def test_greeting_and_capabilities_share_original_handshake_deadline(self): def handler(conn): send(conn, {'event': 'SPICE_CONNECTED', 'timestamp': {'seconds': 1, 'microseconds': 0}}) time.sleep(0.06) greeting(conn) request = json.loads(conn.makefile('rb').readline()) time.sleep(0.06) send(conn, {'return': {}, 'id': request['id']}) self.server(handler) started = time.monotonic() with self.assertRaisesRegex(QMPError, 'deadline|receive'): QMPClient(self.path, timeout=0.1) self.assertLess(time.monotonic() - started, 0.17) def test_event_then_greeting_does_not_bypass_capabilities_failure(self): def handler(conn): send(conn, {'event': 'SPICE_CONNECTED', 'timestamp': {'seconds': 1, 'microseconds': 0}}) greeting(conn) request = json.loads(conn.makefile('rb').readline()) send(conn, {'error': {'class': 'GenericError', 'desc': 'refused'}, 'id': request['id']}) self.server(handler) with self.assertRaisesRegex(QMPError, 'QMP qmp_capabilities'): QMPClient(self.path, timeout=0.2) if __name__ == '__main__': unittest.main()