1 # SPDX-License-Identifier: GPL-2.0
3 # Runs UML kernel, collects output, and handles errors.
5 # Copyright (C) 2019, Google LLC.
6 # Author: Felix Guo <felixguoxiuping@gmail.com>
7 # Author: Brendan Higgins <brendanhiggins@google.com>
18 from typing import Iterator, List, Optional, Tuple
24 KCONFIG_PATH = '.config'
25 KUNITCONFIG_PATH = '.kunitconfig'
26 OLD_KUNITCONFIG_PATH = 'last_used_kunitconfig'
27 DEFAULT_KUNITCONFIG_PATH = 'tools/testing/kunit/configs/default.config'
28 BROKEN_ALLCONFIG_PATH = 'tools/testing/kunit/configs/broken_on_uml.config'
29 OUTFILE_PATH = 'test.log'
30 ABS_TOOL_PATH = os.path.abspath(os.path.dirname(__file__))
31 QEMU_CONFIGS_DIR = os.path.join(ABS_TOOL_PATH, 'qemu_configs')
33 class ConfigError(Exception):
34 """Represents an error trying to configure the Linux kernel."""
37 class BuildError(Exception):
38 """Represents an error trying to build the Linux kernel."""
41 class LinuxSourceTreeOperations:
42 """An abstraction over command line operations performed on a source tree."""
44 def __init__(self, linux_arch: str, cross_compile: Optional[str]):
45 self._linux_arch = linux_arch
46 self._cross_compile = cross_compile
48 def make_mrproper(self) -> None:
50 subprocess.check_output(['make', 'mrproper'], stderr=subprocess.STDOUT)
52 raise ConfigError('Could not call make command: ' + str(e))
53 except subprocess.CalledProcessError as e:
54 raise ConfigError(e.output.decode())
56 def make_arch_qemuconfig(self, base_kunitconfig: kunit_config.Kconfig) -> None:
59 def make_allyesconfig(self, build_dir: str, make_options) -> None:
60 raise ConfigError('Only the "um" arch is supported for alltests')
62 def make_olddefconfig(self, build_dir: str, make_options) -> None:
63 command = ['make', 'ARCH=' + self._linux_arch, 'O=' + build_dir, 'olddefconfig']
64 if self._cross_compile:
65 command += ['CROSS_COMPILE=' + self._cross_compile]
67 command.extend(make_options)
68 print('Populating config with:\n$', ' '.join(command))
70 subprocess.check_output(command, stderr=subprocess.STDOUT)
72 raise ConfigError('Could not call make command: ' + str(e))
73 except subprocess.CalledProcessError as e:
74 raise ConfigError(e.output.decode())
76 def make(self, jobs, build_dir: str, make_options) -> None:
77 command = ['make', 'ARCH=' + self._linux_arch, 'O=' + build_dir, '--jobs=' + str(jobs)]
79 command.extend(make_options)
80 if self._cross_compile:
81 command += ['CROSS_COMPILE=' + self._cross_compile]
82 print('Building with:\n$', ' '.join(command))
84 proc = subprocess.Popen(command,
85 stderr=subprocess.PIPE,
86 stdout=subprocess.DEVNULL)
88 raise BuildError('Could not call execute make: ' + str(e))
89 except subprocess.CalledProcessError as e:
90 raise BuildError(e.output)
91 _, stderr = proc.communicate()
92 if proc.returncode != 0:
93 raise BuildError(stderr.decode())
94 if stderr: # likely only due to build warnings
95 print(stderr.decode())
97 def start(self, params: List[str], build_dir: str) -> subprocess.Popen:
98 raise RuntimeError('not implemented!')
101 class LinuxSourceTreeOperationsQemu(LinuxSourceTreeOperations):
103 def __init__(self, qemu_arch_params: qemu_config.QemuArchParams, cross_compile: Optional[str]):
104 super().__init__(linux_arch=qemu_arch_params.linux_arch,
105 cross_compile=cross_compile)
106 self._kconfig = qemu_arch_params.kconfig
107 self._qemu_arch = qemu_arch_params.qemu_arch
108 self._kernel_path = qemu_arch_params.kernel_path
109 self._kernel_command_line = qemu_arch_params.kernel_command_line + ' kunit_shutdown=reboot'
110 self._extra_qemu_params = qemu_arch_params.extra_qemu_params
112 def make_arch_qemuconfig(self, base_kunitconfig: kunit_config.Kconfig) -> None:
113 kconfig = kunit_config.parse_from_string(self._kconfig)
114 base_kunitconfig.merge_in_entries(kconfig)
116 def start(self, params: List[str], build_dir: str) -> subprocess.Popen:
117 kernel_path = os.path.join(build_dir, self._kernel_path)
118 qemu_command = ['qemu-system-' + self._qemu_arch,
121 '-kernel', kernel_path,
122 '-append', ' '.join(params + [self._kernel_command_line]),
125 '-serial', 'stdio'] + self._extra_qemu_params
126 # Note: shlex.join() does what we want, but requires python 3.8+.
127 print('Running tests with:\n$', ' '.join(shlex.quote(arg) for arg in qemu_command))
128 return subprocess.Popen(qemu_command,
129 stdin=subprocess.PIPE,
130 stdout=subprocess.PIPE,
131 stderr=subprocess.STDOUT,
132 text=True, errors='backslashreplace')
134 class LinuxSourceTreeOperationsUml(LinuxSourceTreeOperations):
135 """An abstraction over command line operations performed on a source tree."""
137 def __init__(self, cross_compile=None):
138 super().__init__(linux_arch='um', cross_compile=cross_compile)
140 def make_allyesconfig(self, build_dir: str, make_options) -> None:
141 kunit_parser.print_with_timestamp(
142 'Enabling all CONFIGs for UML...')
143 command = ['make', 'ARCH=um', 'O=' + build_dir, 'allyesconfig']
145 command.extend(make_options)
146 process = subprocess.Popen(
148 stdout=subprocess.DEVNULL,
149 stderr=subprocess.STDOUT)
151 kunit_parser.print_with_timestamp(
152 'Disabling broken configs to run KUnit tests...')
154 with open(get_kconfig_path(build_dir), 'a') as config:
155 with open(BROKEN_ALLCONFIG_PATH, 'r') as disable:
156 config.write(disable.read())
157 kunit_parser.print_with_timestamp(
158 'Starting Kernel with all configs takes a few minutes...')
160 def start(self, params: List[str], build_dir: str) -> subprocess.Popen:
161 """Runs the Linux UML binary. Must be named 'linux'."""
162 linux_bin = os.path.join(build_dir, 'linux')
163 return subprocess.Popen([linux_bin] + params,
164 stdin=subprocess.PIPE,
165 stdout=subprocess.PIPE,
166 stderr=subprocess.STDOUT,
167 text=True, errors='backslashreplace')
169 def get_kconfig_path(build_dir: str) -> str:
170 return os.path.join(build_dir, KCONFIG_PATH)
172 def get_kunitconfig_path(build_dir: str) -> str:
173 return os.path.join(build_dir, KUNITCONFIG_PATH)
175 def get_old_kunitconfig_path(build_dir: str) -> str:
176 return os.path.join(build_dir, OLD_KUNITCONFIG_PATH)
178 def get_outfile_path(build_dir: str) -> str:
179 return os.path.join(build_dir, OUTFILE_PATH)
181 def get_source_tree_ops(arch: str, cross_compile: Optional[str]) -> LinuxSourceTreeOperations:
182 config_path = os.path.join(QEMU_CONFIGS_DIR, arch + '.py')
184 return LinuxSourceTreeOperationsUml(cross_compile=cross_compile)
185 if os.path.isfile(config_path):
186 return get_source_tree_ops_from_qemu_config(config_path, cross_compile)[1]
188 options = [f[:-3] for f in os.listdir(QEMU_CONFIGS_DIR) if f.endswith('.py')]
189 raise ConfigError(arch + ' is not a valid arch, options are ' + str(sorted(options)))
191 def get_source_tree_ops_from_qemu_config(config_path: str,
192 cross_compile: Optional[str]) -> Tuple[
193 str, LinuxSourceTreeOperations]:
194 # The module name/path has very little to do with where the actual file
195 # exists (I learned this through experimentation and could not find it
196 # anywhere in the Python documentation).
198 # Bascially, we completely ignore the actual file location of the config
199 # we are loading and just tell Python that the module lives in the
200 # QEMU_CONFIGS_DIR for import purposes regardless of where it actually
202 module_path = '.' + os.path.join(os.path.basename(QEMU_CONFIGS_DIR), os.path.basename(config_path))
203 spec = importlib.util.spec_from_file_location(module_path, config_path)
204 assert spec is not None
205 config = importlib.util.module_from_spec(spec)
206 # See https://github.com/python/typeshed/pull/2626 for context.
207 assert isinstance(spec.loader, importlib.abc.Loader)
208 spec.loader.exec_module(config)
210 if not hasattr(config, 'QEMU_ARCH'):
211 raise ValueError('qemu_config module missing "QEMU_ARCH": ' + config_path)
212 params: qemu_config.QemuArchParams = config.QEMU_ARCH # type: ignore
213 return params.linux_arch, LinuxSourceTreeOperationsQemu(
214 params, cross_compile=cross_compile)
216 class LinuxSourceTree:
217 """Represents a Linux kernel source tree with KUnit tests."""
224 kconfig_add: Optional[List[str]]=None,
227 qemu_config_path=None) -> None:
228 signal.signal(signal.SIGINT, self.signal_handler)
230 self._arch, self._ops = get_source_tree_ops_from_qemu_config(
231 qemu_config_path, cross_compile)
233 self._arch = 'um' if arch is None else arch
234 self._ops = get_source_tree_ops(self._arch, cross_compile)
240 if os.path.isdir(kunitconfig_path):
241 kunitconfig_path = os.path.join(kunitconfig_path, KUNITCONFIG_PATH)
242 if not os.path.exists(kunitconfig_path):
243 raise ConfigError(f'Specified kunitconfig ({kunitconfig_path}) does not exist')
245 kunitconfig_path = get_kunitconfig_path(build_dir)
246 if not os.path.exists(kunitconfig_path):
247 shutil.copyfile(DEFAULT_KUNITCONFIG_PATH, kunitconfig_path)
249 self._kconfig = kunit_config.parse_file(kunitconfig_path)
251 kconfig = kunit_config.parse_from_string('\n'.join(kconfig_add))
252 self._kconfig.merge_in_entries(kconfig)
254 def arch(self) -> str:
257 def clean(self) -> bool:
259 self._ops.make_mrproper()
260 except ConfigError as e:
265 def validate_config(self, build_dir: str) -> bool:
266 kconfig_path = get_kconfig_path(build_dir)
267 validated_kconfig = kunit_config.parse_file(kconfig_path)
268 if self._kconfig.is_subset_of(validated_kconfig):
270 invalid = self._kconfig.entries() - validated_kconfig.entries()
271 message = 'Not all Kconfig options selected in kunitconfig were in the generated .config.\n' \
272 'This is probably due to unsatisfied dependencies.\n' \
273 'Missing: ' + ', '.join([str(e) for e in invalid])
274 if self._arch == 'um':
275 message += '\nNote: many Kconfig options aren\'t available on UML. You can try running ' \
276 'on a different architecture with something like "--arch=x86_64".'
277 logging.error(message)
280 def build_config(self, build_dir: str, make_options) -> bool:
281 kconfig_path = get_kconfig_path(build_dir)
282 if build_dir and not os.path.exists(build_dir):
285 self._ops.make_arch_qemuconfig(self._kconfig)
286 self._kconfig.write_to_file(kconfig_path)
287 self._ops.make_olddefconfig(build_dir, make_options)
288 except ConfigError as e:
291 if not self.validate_config(build_dir):
294 old_path = get_old_kunitconfig_path(build_dir)
295 if os.path.exists(old_path):
296 os.remove(old_path) # write_to_file appends to the file
297 self._kconfig.write_to_file(old_path)
300 def _kunitconfig_changed(self, build_dir: str) -> bool:
301 old_path = get_old_kunitconfig_path(build_dir)
302 if not os.path.exists(old_path):
305 old_kconfig = kunit_config.parse_file(old_path)
306 return old_kconfig.entries() != self._kconfig.entries()
308 def build_reconfig(self, build_dir: str, make_options) -> bool:
309 """Creates a new .config if it is not a subset of the .kunitconfig."""
310 kconfig_path = get_kconfig_path(build_dir)
311 if not os.path.exists(kconfig_path):
312 print('Generating .config ...')
313 return self.build_config(build_dir, make_options)
315 existing_kconfig = kunit_config.parse_file(kconfig_path)
316 self._ops.make_arch_qemuconfig(self._kconfig)
317 if self._kconfig.is_subset_of(existing_kconfig) and not self._kunitconfig_changed(build_dir):
319 print('Regenerating .config ...')
320 os.remove(kconfig_path)
321 return self.build_config(build_dir, make_options)
323 def build_kernel(self, alltests, jobs, build_dir: str, make_options) -> bool:
326 self._ops.make_allyesconfig(build_dir, make_options)
327 self._ops.make_olddefconfig(build_dir, make_options)
328 self._ops.make(jobs, build_dir, make_options)
329 except (ConfigError, BuildError) as e:
332 return self.validate_config(build_dir)
334 def run_kernel(self, args=None, build_dir='', filter_glob='', timeout=None) -> Iterator[str]:
337 args.extend(['mem=1G', 'console=tty', 'kunit_shutdown=halt'])
339 args.append('kunit.filter_glob='+filter_glob)
341 process = self._ops.start(args, build_dir)
342 assert process.stdout is not None # tell mypy it's set
344 # Enforce the timeout in a background thread.
347 process.wait(timeout=timeout)
348 except Exception as e:
352 waiter = threading.Thread(target=_wait_proc)
355 output = open(get_outfile_path(build_dir), 'w')
357 # Tee the output to the file and to our caller in real time.
358 for line in process.stdout:
361 # This runs even if our caller doesn't consume every line.
363 # Flush any leftover output to the file
364 output.write(process.stdout.read())
366 process.stdout.close()
369 subprocess.call(['stty', 'sane'])
371 def signal_handler(self, unused_sig, unused_frame) -> None:
372 logging.error('Build interruption occurred. Cleaning console.')
373 subprocess.call(['stty', 'sane'])