945fd61c06
Add testcases for testing fadump with PSeries and PSeries+KVM
combinations
It tests if fadump is successfully detected and registered in the first
kernel boot. Then crashes the kernel, and verifies whether we have a
/proc/vmcore in the 2nd boot
Also introduce 'wait_for_regex_console_pattern' to check for cases where
there is a single success message, but can have multiple failure
messages.
This is particularly useful for cases such as fadump, where the
success message is
"Reserved 1024MB ... successfully"
But at the same point, it can fail with multiple errors such as
"Not supported" or "Allocation failed"
'wait_for_regex_console_pattern' also has a timeout, for cases when we
know the success/failure should appear in a short amount of time,
instead of waiting for the much longer test timeout, such as kernels
with support of fadump will print the success/failure in earlyboot of
the kernel, while kernel without support of fadump won't print anything
for long time, and without a timeout the testcase keeps waiting till
longer test timeout
Signed-off-by: Aditya Gupta <adityag@linux.ibm.com>
Tested-by: Shivang Upadhyay <shivangu@linux.ibm.com>
Link: https://lore.kernel.org/qemu-devel/20251021134823.1861675-8-adityag@linux.ibm.com
Signed-off-by: Harsh Prateek Bora <harshpb@linux.ibm.com>
111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
# Test class for testing the boot process of a Linux kernel
|
|
#
|
|
# This work is licensed under the terms of the GNU GPL, version 2 or
|
|
# later. See the COPYING file in the top-level directory.
|
|
|
|
import hashlib
|
|
import urllib.request
|
|
import logging
|
|
import re
|
|
import time
|
|
|
|
from .cmd import wait_for_console_pattern, exec_command_and_wait_for_pattern
|
|
from .testcase import QemuSystemTest
|
|
from .utils import get_usernet_hostfwd_port
|
|
|
|
|
|
class LinuxKernelTest(QemuSystemTest):
|
|
KERNEL_COMMON_COMMAND_LINE = 'printk.time=0 '
|
|
|
|
def wait_for_console_pattern(self, success_message, vm=None):
|
|
wait_for_console_pattern(self, success_message,
|
|
failure_message='Kernel panic - not syncing',
|
|
vm=vm)
|
|
|
|
def wait_for_regex_console_pattern(self, success_pattern,
|
|
failure_pattern=None,
|
|
timeout=None):
|
|
"""
|
|
Similar to 'wait_for_console_pattern', but supports regex patterns,
|
|
hence multiple failure/success patterns can be detected at a time.
|
|
|
|
Args:
|
|
success_pattern (str | re.Pattern): A regex pattern that indicates
|
|
a successful event. If found, the method exits normally.
|
|
failure_pattern (str | re.Pattern, optional): A regex pattern that
|
|
indicates a failure event. If found, the test fails
|
|
timeout (int, optional): The maximum time (in seconds) to wait for
|
|
a match.
|
|
If exceeded, the test fails.
|
|
"""
|
|
|
|
console = self.vm.console_file
|
|
console_logger = logging.getLogger('console')
|
|
|
|
self.log.debug(
|
|
f"Console interaction: success_msg='{success_pattern}' " +
|
|
f"failure_msg='{failure_pattern}' timeout='{timeout}s'")
|
|
|
|
# Only consume console output if waiting for something
|
|
if success_pattern is None and failure_pattern is None:
|
|
return
|
|
|
|
start_time = time.time()
|
|
|
|
while time.time() - start_time < timeout:
|
|
try:
|
|
msg = console.readline().decode().strip()
|
|
except UnicodeDecodeError:
|
|
msg = None
|
|
if not msg:
|
|
continue
|
|
console_logger.debug(msg)
|
|
if success_pattern is None or re.search(success_pattern, msg):
|
|
break
|
|
if failure_pattern:
|
|
# Find the matching error to print in log
|
|
match = re.search(failure_pattern, msg)
|
|
if not match:
|
|
continue
|
|
|
|
console.close()
|
|
fail = 'Failure message found in console: "%s".' \
|
|
' Expected: "%s"' % \
|
|
(match.group(), success_pattern)
|
|
self.fail(fail)
|
|
|
|
if time.time() - start_time >= timeout:
|
|
fail = f"Timeout ({timeout}s) while trying to search pattern"
|
|
self.fail(fail)
|
|
|
|
def launch_kernel(self, kernel, initrd=None, dtb=None, console_index=0,
|
|
wait_for=None):
|
|
self.vm.set_console(console_index=console_index)
|
|
self.vm.add_args('-kernel', kernel)
|
|
if initrd:
|
|
self.vm.add_args('-initrd', initrd)
|
|
if dtb:
|
|
self.vm.add_args('-dtb', dtb)
|
|
self.vm.launch()
|
|
if wait_for:
|
|
self.wait_for_console_pattern(wait_for)
|
|
|
|
def check_http_download(self, filename, hashsum, guestport=8080,
|
|
pythoncmd='python3 -m http.server'):
|
|
exec_command_and_wait_for_pattern(self,
|
|
f'{pythoncmd} {guestport} & sleep 1',
|
|
f'Serving HTTP on 0.0.0.0 port {guestport}')
|
|
hl = hashlib.sha256()
|
|
hostport = get_usernet_hostfwd_port(self.vm)
|
|
url = f'http://localhost:{hostport}{filename}'
|
|
self.log.info(f'Downloading {url} ...')
|
|
with urllib.request.urlopen(url) as response:
|
|
while True:
|
|
chunk = response.read(1 << 20)
|
|
if not chunk:
|
|
break
|
|
hl.update(chunk)
|
|
|
|
digest = hl.hexdigest()
|
|
self.log.info(f'sha256sum of download is {digest}.')
|
|
self.assertEqual(digest, hashsum)
|