@@ -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,
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> --- tests/functional/qemu_test/cmd.py | 79 +++++++++++++++++++++++++------ 1 file changed, 64 insertions(+), 15 deletions(-)