1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
# 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)
|
