fridtjof/qemu-cr16
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests/functional: return output from cmd.py helpers

Browse source 
Tests might want to look at the whole output from a command execution,
as well as just logging it. Add support for this.

Signed-off-by: John Levon <john.levon@nutanix.com>
Reviewed-by: Daniel P. Berrangé <berrange@redhat.com>
Message-ID: <20250903201931.168317-2-john.levon@nutanix.com>
Signed-off-by: Thomas Huth <thuth@redhat.com>
This commit is contained in:
John Levon 2025-09-03 22:19:29 +02:00 • committed by Thomas Huth
parent f547894f68
commit 595ba81145
1 changed files with 44 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

53
tests/functional/qemu_test/cmd.py
View file

@ -45,6 +45,9 @@ def is_readable_executable_file(path):
# If end of line is seen, with neither @success or @failure
# return False
#
# In both cases, also return the contents of the line (in bytes)
# up to that point.
#
# If @failure is seen, then mark @test as failed
def _console_read_line_until_match(test, vm, success, failure):
msg = bytes([])
@ -76,10 +79,23 @@ def _console_read_line_until_match(test, vm, success, failure):
except:
console_logger.debug(msg)
return done
return done, msg
def _console_interaction(test, success_message, failure_message,
send_string, keep_sending=False, vm=None):
"""
Interact with the console until either message is seen.
:param success_message: if this message appears, finish interaction
:param failure_message: if this message appears, test fails
:param send_string: a string to send to the console before trying
to read a new line
:param keep_sending: keep sending the send string each time
:param vm: the VM to interact with
:return: The collected output (in bytes form).
"""
assert not keep_sending or send_string
assert success_message or send_string
@ -101,6 +117,8 @@ def _console_interaction(test, success_message, failure_message,
if failure_message is not None:
failure_message_b = failure_message.encode()
out = bytes([])
while True:
if send_string:
vm.console_socket.sendall(send_string.encode())
@ -113,11 +131,17 @@ def _console_interaction(test, success_message, failure_message,
break
continue
if _console_read_line_until_match(test, vm,
success_message_b,
failure_message_b):
done, line = _console_read_line_until_match(test, vm,
success_message_b,
failure_message_b)
out += line
if done:
break
return out
def interrupt_interactive_console_until_pattern(test, success_message,
failure_message=None,
interrupt_string='\r'):
@ -140,10 +164,12 @@ def interrupt_interactive_console_until_pattern(test, success_message,
:param failure_message: if this message appears, test fails
:param interrupt_string: a string to send to the console before trying
to read a new line
:return: The collected output (in bytes form).
"""
assert success_message
_console_interaction(test, success_message, failure_message,
interrupt_string, True)
return _console_interaction(test, success_message, failure_message,
interrupt_string, True)
def wait_for_console_pattern(test, success_message, failure_message=None,
vm=None):
@ -155,9 +181,12 @@ def wait_for_console_pattern(test, success_message, failure_message=None,
:type test: :class:`qemu_test.QemuSystemTest`
:param success_message: if this message appears, test succeeds
:param failure_message: if this message appears, test fails
:return: The collected output (in bytes form).
"""
assert success_message
_console_interaction(test, success_message, failure_message, None, vm=vm)
return _console_interaction(test, success_message, failure_message,
None, vm=vm)
def exec_command(test, command):
"""
@ -168,8 +197,10 @@ def exec_command(test, command):
:type test: :class:`qemu_test.QemuSystemTest`
:param command: the command to send
:type command: str
:return: The collected output (in bytes form).
"""
_console_interaction(test, None, None, command + '\r')
return _console_interaction(test, None, None, command + '\r')
def exec_command_and_wait_for_pattern(test, command,
success_message, failure_message=None):
@ -184,9 +215,13 @@ def exec_command_and_wait_for_pattern(test, command,
:param command: the command to send
:param success_message: if this message appears, test succeeds
:param failure_message: if this message appears, test fails
:return: The collected output (in bytes form).
"""
assert success_message
_console_interaction(test, success_message, failure_message, command + '\r')
return _console_interaction(test, success_message, failure_message,
command + '\r')
def get_qemu_img(test):
test.log.debug('Looking for and selecting a qemu-img binary')