| 1 | # Test class for testing the boot process of a Linux kernel |
| 2 | # |
| 3 | # This work is licensed under the terms of the GNU GPL, version 2 or |
| 4 | # later. See the COPYING file in the top-level directory. |
| 5 | |
| 6 | import hashlib |
| 7 | import urllib.request |
| 8 | import logging |
| 9 | import re |
| 10 | import time |
| 11 | |
| 12 | from .cmd import wait_for_console_pattern, exec_command_and_wait_for_pattern |
| 13 | from .testcase import QemuSystemTest |
| 14 | from .utils import get_usernet_hostfwd_port |
| 15 | |
| 16 | |
| 17 | class LinuxKernelTest(QemuSystemTest): |
| 18 | KERNEL_COMMON_COMMAND_LINE = 'printk.time=0 ' |
| 19 | |
| 20 | def wait_for_console_pattern(self, success_message, vm=None): |
| 21 | wait_for_console_pattern(self, success_message, |
| 22 | failure_message='Kernel panic - not syncing', |
| 23 | vm=vm) |
| 24 | |
| 25 | def wait_for_regex_console_pattern(self, success_pattern, |
| 26 | failure_pattern=None, |
| 27 | timeout=None): |
| 28 | """ |
| 29 | Similar to 'wait_for_console_pattern', but supports regex patterns, |
| 30 | hence multiple failure/success patterns can be detected at a time. |
| 31 | |
| 32 | Args: |
| 33 | success_pattern (str | re.Pattern): A regex pattern that indicates |
| 34 | a successful event. If found, the method exits normally. |
| 35 | failure_pattern (str | re.Pattern, optional): A regex pattern that |
| 36 | indicates a failure event. If found, the test fails |
| 37 | timeout (int, optional): The maximum time (in seconds) to wait for |
| 38 | a match. |
| 39 | If exceeded, the test fails. |
| 40 | """ |
| 41 | |
| 42 | console = self.vm.console_file |
| 43 | console_logger = logging.getLogger('console') |
| 44 | |
| 45 | self.log.debug( |
| 46 | f"Console interaction: success_msg='{success_pattern}' " + |
| 47 | f"failure_msg='{failure_pattern}' timeout='{timeout}s'") |
| 48 | |
| 49 | # Only consume console output if waiting for something |
| 50 | if success_pattern is None and failure_pattern is None: |
| 51 | return |
| 52 | |
| 53 | start_time = time.time() |
| 54 | |
| 55 | while time.time() - start_time < timeout: |
| 56 | try: |
| 57 | msg = console.readline().decode().strip() |
| 58 | except UnicodeDecodeError: |
| 59 | msg = None |
| 60 | if not msg: |
| 61 | continue |
| 62 | console_logger.debug(msg) |
| 63 | if success_pattern is None or re.search(success_pattern, msg): |
| 64 | break |
| 65 | if failure_pattern: |
| 66 | # Find the matching error to print in log |
| 67 | match = re.search(failure_pattern, msg) |
| 68 | if not match: |
| 69 | continue |
| 70 | |
| 71 | console.close() |
| 72 | fail = 'Failure message found in console: "%s".' \ |
| 73 | ' Expected: "%s"' % \ |
| 74 | (match.group(), success_pattern) |
| 75 | self.fail(fail) |
| 76 | |
| 77 | if time.time() - start_time >= timeout: |
| 78 | fail = f"Timeout ({timeout}s) while trying to search pattern" |
| 79 | self.fail(fail) |
| 80 | |
| 81 | def launch_kernel(self, kernel, initrd=None, dtb=None, console_index=0, |
| 82 | wait_for=None): |
| 83 | self.vm.set_console(console_index=console_index) |
| 84 | self.vm.add_args('-kernel', kernel) |
| 85 | if initrd: |
| 86 | self.vm.add_args('-initrd', initrd) |
| 87 | if dtb: |
| 88 | self.vm.add_args('-dtb', dtb) |
| 89 | self.vm.launch() |
| 90 | if wait_for: |
| 91 | self.wait_for_console_pattern(wait_for) |
| 92 | |
| 93 | def check_http_download(self, filename, hashsum, guestport=8080, |
| 94 | pythoncmd='python3 -m http.server'): |
| 95 | exec_command_and_wait_for_pattern(self, |
| 96 | f'{pythoncmd} {guestport} & sleep 1', |
| 97 | f'Serving HTTP on 0.0.0.0 port {guestport}') |
| 98 | hl = hashlib.sha256() |
| 99 | hostport = get_usernet_hostfwd_port(self.vm) |
| 100 | url = f'http://localhost:{hostport}{filename}' |
| 101 | self.log.info(f'Downloading {url} ...') |
| 102 | with urllib.request.urlopen(url) as response: |
| 103 | while True: |
| 104 | chunk = response.read(1 << 20) |
| 105 | if not chunk: |
| 106 | break |
| 107 | hl.update(chunk) |
| 108 | |
| 109 | digest = hl.hexdigest() |
| 110 | self.log.info(f'sha256sum of download is {digest}.') |
| 111 | self.assertEqual(digest, hashsum) |