[scripts] add do-qemu-boot-tests.py
This will boot a list of qemu emulated devices with the option to run all of the unittests at boot. Will be automatically enabled on the build servers soon which will fail any pending CLs if it breaks.
This commit is contained in:
Travis Geiselbrecht
222
scripts/do-qemu-boot-tests.py
Executable file
222
scripts/do-qemu-boot-tests.py
Executable file
@@ -0,0 +1,222 @@
|
||||
#!/usr/bin/env python3
|
||||
import subprocess
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
class QEMUTestRunner:
|
||||
def __init__(self, lk_root):
|
||||
self.lk_root = Path(lk_root)
|
||||
self.architectures = {
|
||||
'arm': {
|
||||
'script': 'do-qemuarm',
|
||||
'args': '',
|
||||
'timeout': 30
|
||||
},
|
||||
'arm64': {
|
||||
'script': 'do-qemuarm',
|
||||
'args': '-6',
|
||||
'timeout': 30
|
||||
},
|
||||
'm68k': {
|
||||
'script': 'do-qemum68k',
|
||||
'args': '',
|
||||
'timeout': 30
|
||||
},
|
||||
'riscv32': {
|
||||
'script': 'do-qemuriscv',
|
||||
'args': '',
|
||||
'timeout': 30
|
||||
},
|
||||
'riscv64': {
|
||||
'script': 'do-qemuriscv',
|
||||
'args': '-6S',
|
||||
'timeout': 30
|
||||
},
|
||||
'x86': {
|
||||
'script': 'do-qemux86',
|
||||
'args': '',
|
||||
'timeout': 30
|
||||
},
|
||||
'x86-64': {
|
||||
'script': 'do-qemux86',
|
||||
'args': '-6',
|
||||
'timeout': 30
|
||||
}
|
||||
}
|
||||
|
||||
def run_qemu_test(self, arch, arch_config, quiet=False):
|
||||
"""Run QEMU for the specified architecture and monitor for test completion"""
|
||||
print(f"\nRunning QEMU test for {arch}...")
|
||||
|
||||
script_path = self.lk_root / 'scripts' / arch_config['script']
|
||||
if not script_path.exists():
|
||||
print(f"Script {script_path} not found")
|
||||
return False
|
||||
|
||||
# Create the QEMU commandline
|
||||
qemu_cmdline = [str(script_path)]
|
||||
if arch_config['args']:
|
||||
qemu_cmdline.append(str(arch_config['args']))
|
||||
|
||||
if not quiet:
|
||||
print(f"Executing command: {' '.join(qemu_cmdline)}")
|
||||
|
||||
# Run the QEMU script
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
qemu_cmdline,
|
||||
cwd=self.lk_root,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
universal_newlines=True,
|
||||
env={**os.environ, 'LK_ROOT': str(self.lk_root), 'RUN_UNITTESTS_AT_BOOT': '1'},
|
||||
)
|
||||
|
||||
stdout = process.stdout
|
||||
if stdout is None:
|
||||
print("Failed to capture stdout")
|
||||
return False
|
||||
|
||||
# Monitor output for success/failure indicators
|
||||
success_indicators = [
|
||||
"SUCCESS! All test cases passed",
|
||||
]
|
||||
|
||||
failure_indicators = [
|
||||
"FAILURE! Some test cases failed",
|
||||
"panic (caller"
|
||||
"CRASH: starting debug shell"
|
||||
]
|
||||
|
||||
start_time = time.time()
|
||||
output_lines = []
|
||||
test_passed = False
|
||||
test_failed = False
|
||||
|
||||
while True:
|
||||
# Check timeout
|
||||
if time.time() - start_time > arch_config['timeout']:
|
||||
print(f"Timeout reached for {arch}")
|
||||
process.terminate()
|
||||
break
|
||||
|
||||
# Read a line of output
|
||||
# TODO: fix the situation where readline() blocks indefinitely because the process
|
||||
# has printed a partial line without a newline.
|
||||
line = stdout.readline()
|
||||
if not line and process.poll() is not None:
|
||||
break
|
||||
|
||||
if line:
|
||||
line = line.removesuffix('\n')
|
||||
output_lines.append(line)
|
||||
if not quiet:
|
||||
print(f"[{arch}] {line}")
|
||||
|
||||
# Check for success indicators
|
||||
for indicator in success_indicators:
|
||||
if indicator.lower() in line.lower():
|
||||
test_passed = True
|
||||
print(f"✓ Test success detected for {arch}")
|
||||
break
|
||||
|
||||
# Check for failure indicators
|
||||
for indicator in failure_indicators:
|
||||
if indicator.lower() in line.lower():
|
||||
test_failed = True
|
||||
print(f"✗ Test failure detected for {arch}")
|
||||
break
|
||||
|
||||
if test_passed or test_failed:
|
||||
break
|
||||
|
||||
# Clean up process
|
||||
if not quiet:
|
||||
print("Terminating QEMU process...")
|
||||
if process.poll() is None:
|
||||
process.terminate()
|
||||
time.sleep(1)
|
||||
if process.poll() is None:
|
||||
process.kill()
|
||||
|
||||
return test_passed and not test_failed
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error running QEMU for {arch}: {e}")
|
||||
return False
|
||||
|
||||
def run_all_tests(self, selected_archs=None, quiet=False):
|
||||
"""Run tests for all or selected architectures"""
|
||||
if selected_archs is None:
|
||||
selected_archs = list(self.architectures.keys())
|
||||
|
||||
results = {}
|
||||
|
||||
for arch in selected_archs:
|
||||
if arch not in self.architectures:
|
||||
print(f"Unknown architecture: {arch}")
|
||||
continue
|
||||
|
||||
arch_config = self.architectures[arch]
|
||||
|
||||
# Run the test
|
||||
results[arch] = self.run_qemu_test(arch, arch_config, quiet)
|
||||
|
||||
return results
|
||||
|
||||
def print_summary(self, results):
|
||||
"""Print a summary of test results"""
|
||||
print("\n" + "="*50)
|
||||
print("TEST SUMMARY")
|
||||
print("="*50)
|
||||
|
||||
total_tests = len(results)
|
||||
passed_tests = sum(1 for result in results.values() if result)
|
||||
|
||||
for arch, passed in results.items():
|
||||
status = "PASSED" if passed else "FAILED"
|
||||
symbol = "✓" if passed else "✗"
|
||||
print(f"{symbol} {arch:10} {status}")
|
||||
|
||||
print("-"*50)
|
||||
print(f"Total: {passed_tests}/{total_tests} architectures passed")
|
||||
|
||||
if passed_tests == total_tests:
|
||||
print("🎉 All architectures passed!")
|
||||
return 0
|
||||
else:
|
||||
print("❌ Some architectures failed!")
|
||||
return 1
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Run LK QEMU tests for multiple architectures')
|
||||
parser.add_argument('--arch', choices=['arm', 'arm64', 'm68k', 'riscv32', 'riscv64', 'x86', 'x86-64'], action='append',
|
||||
help='Architecture to test (can be specified multiple times)')
|
||||
parser.add_argument('--lk-root', default='.',
|
||||
help='Path to LK root directory (default: current directory)')
|
||||
parser.add_argument('--quiet', action='store_true',
|
||||
help='Run tests in quiet mode (suppress output)')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Change to LK root directory
|
||||
lk_root = os.path.abspath(args.lk_root)
|
||||
if not os.path.exists(os.path.join(lk_root, 'makefile')) and not os.path.exists(os.path.join(lk_root, 'Makefile')):
|
||||
print(f"Error: {lk_root} doesn't appear to be the LK root directory")
|
||||
return 1
|
||||
|
||||
runner = QEMUTestRunner(lk_root)
|
||||
|
||||
# Run tests
|
||||
results = runner.run_all_tests(args.arch, args.quiet)
|
||||
|
||||
# Print summary and return appropriate exit code
|
||||
return runner.print_summary(results)
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user