xref: /openbmc/qemu/tests/functional/qemu_test/linuxkernel.py (revision 7ae004869aff46fc3195d280b25dc9b94a447be7) 
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
6import hashlib
7import urllib.request
8import logging
9import re
10import time
11
12from .cmd import wait_for_console_pattern, exec_command_and_wait_for_pattern
13from .testcase import QemuSystemTest
14from .utils import get_usernet_hostfwd_port
15
16
17class 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)
112