tests/functional: rewrite console handling to be bytewise

The console interaction that waits for predicted strings uses
readline(), and thus is only capable of waiting for strings
that are followed by a newline.

This is inconvenient when needing to match on some things,
particularly login prompts, or shell prompts, causing tests
to use time.sleep(...) instead, which is unreliable.

Switch to reading the console 1 byte at a time, comparing
against the success/failure messages until we see a match,
regardless of whether a newline is encountered.

The success/failure comparisons are done with the python bytes
type, rather than strings, to avoid the problem of needing to
decode partially received multibyte utf8 characters.

Heavily inspired by a patch proposed by Cédric, but written
again to work in bytes, rather than strings.

Co-developed-by: Cédric Le Goater <clg@redhat.com>
Signed-off-by: Daniel P. Berrangé <berrange@redhat.com>
Message-Id: <20241121154218.1423005-16-berrange@redhat.com>
Signed-off-by: Alex Bennée <alex.bennee@linaro.org>
Message-Id: <20241121165806.476008-16-alex.bennee@linaro.org>
This commit is contained in:
Daniel P. Berrangé 2024-11-21 16:57:42 +00:00 committed by Alex Bennée
parent f03a81897d
commit cdad03b74f
1 changed files with 64 additions and 15 deletions
tests/functional/qemu_test

View File

@ -78,6 +78,54 @@ def run_cmd(args):
def is_readable_executable_file(path):
return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK)
# @test: functional test to fail if @failure is seen
# @vm: the VM whose console to process
# @success: a non-None string to look for
# @failure: a string to look for that triggers test failure, or None
#
# Read up to 1 line of text from @vm, looking for @success
# and optionally @failure.
#
# If @success or @failure are seen, immediately return True,
# even if end of line is not yet seen. ie remainder of the
# line is left unread.
#
# If end of line is seen, with neither @success or @failure
# return False
#
# If @failure is seen, then mark @test as failed
def _console_read_line_until_match(test, vm, success, failure):
msg = bytes([])
done = False
while True:
c = vm.console_socket.recv(1)
if c is None:
done = True
test.fail(
f"EOF in console, expected '{success}'")
break
msg += c
if success in msg:
done = True
break
if failure and failure in msg:
done = True
vm.console_socket.close()
test.fail(
f"'{failure}' found in console, expected '{success}'")
if c == b'\n':
break
console_logger = logging.getLogger('console')
try:
console_logger.debug(msg.decode().strip())
except:
console_logger.debug(msg)
return done
def _console_interaction(test, success_message, failure_message,
send_string, keep_sending=False, vm=None):
assert not keep_sending or send_string
@ -85,11 +133,22 @@ def _console_interaction(test, success_message, failure_message,
if vm is None:
vm = test.vm
console = vm.console_file
console_logger = logging.getLogger('console')
test.log.debug(
f"Console interaction: success_msg='{success_message}' " +
f"failure_msg='{failure_message}' send_string='{send_string}'")
# We'll process console in bytes, to avoid having to
# deal with unicode decode errors from receiving
# partial utf8 byte sequences
success_message_b = None
if success_message is not None:
success_message_b = success_message.encode()
failure_message_b = None
if failure_message is not None:
failure_message_b = failure_message.encode()
while True:
if send_string:
vm.console_socket.sendall(send_string.encode())
@ -102,20 +161,10 @@ def _console_interaction(test, success_message, failure_message,
break
continue
try:
msg = console.readline().decode().strip()
except UnicodeDecodeError:
msg = None
if not msg:
continue
console_logger.debug(msg)
if success_message in msg:
if _console_read_line_until_match(test, vm,
success_message_b,
failure_message_b):
break
if failure_message and failure_message in msg:
console.close()
fail = 'Failure message found in console: "%s". Expected: "%s"' % \
(failure_message, success_message)
test.fail(fail)
def interrupt_interactive_console_until_pattern(test, success_message,
failure_message=None,