You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

qtest.py 3.3KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116
  1. # QEMU qtest library
  2. #
  3. # Copyright (C) 2015 Red Hat Inc.
  4. #
  5. # Authors:
  6. # Fam Zheng <famz@redhat.com>
  7. #
  8. # This work is licensed under the terms of the GNU GPL, version 2. See
  9. # the COPYING file in the top-level directory.
  10. #
  11. # Based on qmp.py.
  12. #
  13. import socket
  14. import os
  15. from .machine import QEMUMachine
  16. class QEMUQtestProtocol(object):
  17. def __init__(self, address, server=False):
  18. """
  19. Create a QEMUQtestProtocol object.
  20. @param address: QEMU address, can be either a unix socket path (string)
  21. or a tuple in the form ( address, port ) for a TCP
  22. connection
  23. @param server: server mode, listens on the socket (bool)
  24. @raise socket.error on socket connection errors
  25. @note No connection is established, this is done by the connect() or
  26. accept() methods
  27. """
  28. self._address = address
  29. self._sock = self._get_sock()
  30. self._sockfile = None
  31. if server:
  32. self._sock.bind(self._address)
  33. self._sock.listen(1)
  34. def _get_sock(self):
  35. if isinstance(self._address, tuple):
  36. family = socket.AF_INET
  37. else:
  38. family = socket.AF_UNIX
  39. return socket.socket(family, socket.SOCK_STREAM)
  40. def connect(self):
  41. """
  42. Connect to the qtest socket.
  43. @raise socket.error on socket connection errors
  44. """
  45. self._sock.connect(self._address)
  46. self._sockfile = self._sock.makefile()
  47. def accept(self):
  48. """
  49. Await connection from QEMU.
  50. @raise socket.error on socket connection errors
  51. """
  52. self._sock, _ = self._sock.accept()
  53. self._sockfile = self._sock.makefile()
  54. def cmd(self, qtest_cmd):
  55. """
  56. Send a qtest command on the wire.
  57. @param qtest_cmd: qtest command text to be sent
  58. """
  59. self._sock.sendall((qtest_cmd + "\n").encode('utf-8'))
  60. resp = self._sockfile.readline()
  61. return resp
  62. def close(self):
  63. self._sock.close()
  64. self._sockfile.close()
  65. def settimeout(self, timeout):
  66. self._sock.settimeout(timeout)
  67. class QEMUQtestMachine(QEMUMachine):
  68. '''A QEMU VM'''
  69. def __init__(self, binary, args=None, name=None, test_dir="/var/tmp",
  70. socket_scm_helper=None):
  71. if name is None:
  72. name = "qemu-%d" % os.getpid()
  73. super(QEMUQtestMachine,
  74. self).__init__(binary, args, name=name, test_dir=test_dir,
  75. socket_scm_helper=socket_scm_helper)
  76. self._qtest = None
  77. self._qtest_path = os.path.join(test_dir, name + "-qtest.sock")
  78. def _base_args(self):
  79. args = super(QEMUQtestMachine, self)._base_args()
  80. args.extend(['-qtest', 'unix:path=' + self._qtest_path,
  81. '-accel', 'qtest'])
  82. return args
  83. def _pre_launch(self):
  84. super(QEMUQtestMachine, self)._pre_launch()
  85. self._qtest = QEMUQtestProtocol(self._qtest_path, server=True)
  86. def _post_launch(self):
  87. super(QEMUQtestMachine, self)._post_launch()
  88. self._qtest.accept()
  89. def _post_shutdown(self):
  90. super(QEMUQtestMachine, self)._post_shutdown()
  91. self._remove_if_exists(self._qtest_path)
  92. def qtest(self, cmd):
  93. '''Send a qtest command to guest'''
  94. return self._qtest.cmd(cmd)