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 <c...@redhat.com>
Signed-off-by: Daniel P. Berrangé <berra...@redhat.com>
---
tests/functional/qemu_test/cmd.py | 63 +++++++++++++++++++++++--------
1 file changed, 48 insertions(+), 15 deletions(-)
diff --git a/tests/functional/qemu_test/cmd.py
b/tests/functional/qemu_test/cmd.py
index 76a48064cd..91267a087f 100644
--- a/tests/functional/qemu_test/cmd.py
+++ b/tests/functional/qemu_test/cmd.py
@@ -78,15 +78,58 @@ 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)
+def _console_readline(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 is None or 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
if vm is None:
vm = test.vm
- console = vm.console_file
- console_logger = logging.getLogger('console')
+
test.log.debug(f"Console interaction success:'{success_message}' " +
f"failure:'{failure_message}' send:'{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())
@@ -99,20 +142,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 is None or success_message in msg:
+ if _console_readline(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,
--
2.46.0
