From: Waldemar Kozaczuk <jwkozac...@gmail.com>
Committer: Waldemar Kozaczuk <jwkozac...@gmail.com>
Branch: master
scripts: enhance tests/testing.py
This patch enhances the script in following ways:
- makes it work regardless which place on filesystem called from
- allows piping the standard input to the guest input to help
automate testings interactive apps like cli
- adds couple of new methods to test appearance of text in the guest output
Signed-off-by: Waldemar Kozaczuk <jwkozac...@gmail.com>
---
diff --git a/scripts/tests/testing.py b/scripts/tests/testing.py
--- a/scripts/tests/testing.py
+++ b/scripts/tests/testing.py
@@ -8,7 +8,7 @@
tests = []
_verbose_output = False
-osv_base = '.'
+osv_base =
os.path.join(os.path.dirname(os.path.abspath(__file__)), '../..')
class TestFailed(Exception):
pass
@@ -93,8 +93,12 @@ def
scan_errors(s,scan_for_failed_to_load_object_error=True):
return False
class SupervisedProcess:
- def __init__(self, args, show_output=False, show_output_on_error=True,
scan_for_failed_to_load_object_error=True):
- self.process = subprocess.Popen(args, stdout=subprocess.PIPE)
+ def __init__(self, args, show_output=False, show_output_on_error=True,
scan_for_failed_to_load_object_error=True, pipe_stdin=False):
+ if pipe_stdin:
+ self.process = subprocess.Popen(args, stdout=subprocess.PIPE,
stdin=subprocess.PIPE)
+ else:
+ self.process = subprocess.Popen(args, stdout=subprocess.PIPE)
+ self.pipe_stdin = pipe_stdin
self.cv = threading.Condition()
self.lines = []
self.output_collector_done = False
@@ -105,13 +109,15 @@ def __init__(self, args, show_output=False,
show_output_on_error=True, scan_for_
self.output_collector_thread =
threading.Thread(target=self._output_collector)
self.output_collector_thread.start()
self.scan_for_failed_to_load_object_error =
scan_for_failed_to_load_object_error
+ self.line_with_err = ""
def _output_collector(self):
def append_line(line):
self.cv.acquire()
if not self.has_errors and
scan_errors(line,self.scan_for_failed_to_load_object_error):
self.has_errors = True
+ self.line_with_err = line
if self.show_output_on_error and not self.show_output:
sys.stdout.write(self.output)
sys.stdout.flush()
@@ -126,14 +132,20 @@ def append_line(line):
self.cv.release()
line = ''
+ ch_bytes = ''
while True:
- ch = self.process.stdout.read(1).decode()
- if ch == '' and self.process.poll() != None:
- break
- line += ch
- if ch == '\n':
- append_line(line)
- line = ''
+ ch_bytes = ch_bytes + self.process.stdout.read(1)
+ try:
+ ch = ch_bytes.decode('utf-8')
+ if ch == '' and self.process.poll() != None:
+ break
+ line += ch
+ if ch == '\n':
+ append_line(line)
+ line = ''
+ ch_bytes = ''
+ except UnicodeError:
+ continue
if line:
append_line(line)
@@ -165,6 +177,8 @@ def read_lines(self):
def join(self):
self.output_collector_thread.join()
+ if self.pipe_stdin:
+ self.process.stdin.close()
if self.process.returncode:
raise Exception('Guest failed (returncode=%d)' %
self.proces.returncode)
if self.failed:
@@ -185,6 +199,13 @@ def failed(self):
assert not self.output_collector_thread.isAlive()
return self.has_errors or self.process.returncode
+ def write_line_to_input(self, line):
+ self.process.stdin.write(line + "\n")
+ self.process.stdin.flush()
+
+ def line_with_error(self):
+ return self.line_with_err
+
def run_command_in_guest(command, **kwargs):
common_parameters = ["-e", "--power-off-on-abort " + command]
if 'hypervisor' in kwargs.keys() and kwargs['hypervisor']
== 'firecracker':
@@ -194,7 +215,7 @@ def run_command_in_guest(command, **kwargs):
class Guest(SupervisedProcess):
def __init__(self, args, forward=[], hold_with_poweroff=False,
show_output_on_error=True,
- scan_for_failed_to_load_object_error=True,
run_py_args=[], hypervisor='qemu'):
+ scan_for_failed_to_load_object_error=True,
run_py_args=[], hypervisor='qemu', pipe_stdin=False):
if hypervisor == 'firecracker':
run_script = os.path.join(osv_base, "scripts/firecracker.py")
@@ -215,7 +236,8 @@ def __init__(self, args, forward=[],
hold_with_poweroff=False, show_output_on_er
SupervisedProcess.__init__(self, [run_script] + run_py_args + args,
show_output=_verbose_output,
show_output_on_error=show_output_on_error,
-
scan_for_failed_to_load_object_error=scan_for_failed_to_load_object_error)
+
scan_for_failed_to_load_object_error=scan_for_failed_to_load_object_error,
+ pipe_stdin=pipe_stdin)
def kill(self):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -225,8 +247,17 @@ def kill(self):
s.close()
def wait_for_line(guest, text):
+ return _wait_for_line(guest, lambda line: line == text, text)
+
+def wait_for_line_starts(guest, text):
+ return _wait_for_line(guest, lambda line: line.startswith(text), text)
+
+def wait_for_line_contains(guest, text):
+ return _wait_for_line(guest, lambda line: text in line, text)
+
+def _wait_for_line(guest, predicate, text):
for line in guest.read_lines():
- if line == text:
+ if predicate(line):
return
raise Exception('Text not found in output: ' + text)
--
You received this message because you are subscribed to the Google Groups "OSv
Development" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to osv-dev+unsubscr...@googlegroups.com.
To view this discussion on the web visit
https://groups.google.com/d/msgid/osv-dev/000000000000480a950595931e29%40google.com.
