XenOn/kolibrios-fun
1
1
Fork 0
forked from KolibriOS/kolibrios
Code Pull Requests Wiki Activity

[KERNEL][TEST] Fail on prematurely finished qemu.

Browse Source 
git-svn-id: svn://kolibrios.org@9919 a494cfbc-eb01-0410-851d-a64ba20cac60
This commit is contained in:
Magomed Kostoev (mkostoevr) 2023-06-18 12:44:38 +00:00
parent 0a2a7aa344
commit 82d8c58e1e
1 changed files with 16 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
kernel/trunk/test/common/__init__.py
View File

@ -25,16 +25,25 @@ class TestTimeoutException(Exception):
pass pass
class TestFailureException(Exception): class TestFailureException(Exception):
pass def __init__(self, message):
self.message = message
class Qemu: class Qemu:
def __init__(self, popen, debug_log): def __init__(self, popen, debug_log):
self.popen = popen self.popen = popen
# Qemu needs time to create debug.log file # Qemu needs time to create debug.log file
while not os.path.exists(debug_log): while not os.path.exists(debug_log) and self.qemu_is_alive():
self.wait() self.wait(0.250)
self.assert_qemu_not_died("waiting for the debug log file")
self.debug = open(debug_log, "rb") self.debug = open(debug_log, "rb")
def qemu_is_alive(self):
return self.popen.poll() == None
def assert_qemu_not_died(self, while_):
if not self.qemu_is_alive():
raise TestFailureException(f"Qemu has finished while {while_}.")
def wait_for_debug_log(self, needle, timeout = 1): def wait_for_debug_log(self, needle, timeout = 1):
needle = bytes(needle, "utf-8") needle = bytes(needle, "utf-8")
start = timeit.default_timer() start = timeit.default_timer()
@ -42,6 +51,7 @@ class Qemu:
# While no timeout, read and search logs # While no timeout, read and search logs
while timeit.default_timer() - start < timeout: while timeit.default_timer() - start < timeout:
# TODO: Non-blocking read.
log += self.debug.read(1) log += self.debug.read(1)
if needle in log: if needle in log:
return return
@ -57,6 +67,8 @@ class Qemu:
if len(log) > len(needle) * 2: if len(log) > len(needle) * 2:
log = log[len(needle):] log = log[len(needle):]
self.assert_qemu_not_died("waiting for the debug log")
self.timeout() self.timeout()
def kill(self): def kill(self):
@ -66,15 +78,11 @@ class Qemu:
else: else:
os.killpg(os.getpgid(self.popen.pid), signal.SIGTERM) os.killpg(os.getpgid(self.popen.pid), signal.SIGTERM)
def failure(self):
self.kill()
raise TestFailureException()
def timeout(self): def timeout(self):
self.kill() self.kill()
raise TestTimeoutException() raise TestTimeoutException()
def wait(self, seconds = 0.25): def wait(self, seconds):
time.sleep(seconds) time.sleep(seconds)
def run_qemu(root_dir, test_dir, debug_log): def run_qemu(root_dir, test_dir, debug_log):