xv6, line by line
test-xv6.py

test-xv6.py

Python · 224 lines · annotated 100% · upstream

About this file

A Python 3 script that tests xv6 automatically, with no person at the keyboard. It runs make qemu as a child process, connected through pipes: whatever it writes to QEMU’s standard input arrives at xv6’s console as if typed, and everything xv6 prints comes back on QEMU’s standard output for the script to search. (With -nographic, QEMU connects the emulated UART to its own standard input and output.)

It knows two kinds of test:

  • usertests: boot xv6, type usertests into the shell, and wait up to ten minutes for the line ALL TESTS PASSED (user/usertests.c).
  • crash tests: start a program that is in the middle of changing the file system, kill QEMU abruptly (like pulling the power cord), boot again from the same fs.img, and check that the kernel’s recovery code ran: replaying the log (recover_from_log) and freeing orphaned inodes (ireclaim).

The CI job runs ./test-xv6.py usertests and ./test-xv6.py crash (.github/workflows/test.yml:11); you can run the same on your machine (Linux; the crash tests use a Linux-only ps option). A failure prints FAIL and saves the whole console output in test-xv6.out.

1#!/usr/bin/env python3
3#
4# python script that tests xv6 without having to boot it and type to its shell
5#
6# ./test-xv6.py usertests (runs usertests)
7# ./test-xv6.py -q usertests (runs the quick tests of usertests)
8# ./test-xv6.py crash (runs the crash tests)
9# ./test-xv6.py log (runs the log crash test)
11import argparse, os, inspect, re, signal, subprocess, sys, time
12from subprocess import run
14sys.stdout.reconfigure(line_buffering=True)
16parser = argparse.ArgumentParser()
17parser.add_argument('testrex', help="test name or regular expression")
18parser.add_argument("-q", action='store_true', help="usertests quick")
19args = parser.parse_args()
21class QEMU(object):
23 def __init__(self, reset=False):
24 if reset:
25 self.build_xv6()
26 self.reset_fs()
27 q = ["make", "qemu"]
28 self.proc = subprocess.Popen(q, stdin=subprocess.PIPE,
29 stdout=subprocess.PIPE,
30 stderr=subprocess.STDOUT)
31 os.set_blocking(self.proc.stdout.fileno(), False)
32 self.output = ""
33 self.outbytes = bytearray()
34 self.reported = 0
35 time.sleep(1)
37 def reset_fs(self):
38 try:
39 run(["rm", "-f", "fs.img"], check=True)
40 run(["make", "fs.img"], check=True)
41 except subprocess.CalledProcessError as e:
42 print(f"Command failed with exit code {e.returncode}")
44 def build_xv6(self):
45 try:
46 run(["make", "kernel/kernel"], check=True)
47 except subprocess.CalledProcessError as e:
48 print(f"Command failed with exit code {e.returncode}")
50 def save_output(self):
51 try:
52 with open("test-xv6.out", "w") as f:
53 f.write(self.output)
54 f.close()
55 except OSError as e:
56 print("Provided a bad results path. Error:", e)
58 def cmd(self, c):
59 if isinstance(c, str):
60 c = c.encode('utf-8')
61 self.proc.stdin.write(c)
62 self.proc.stdin.flush()
64 def crash(self):
65 ps = run(['ps', '-opid', '--no-headers', '--ppid', str(self.proc.pid)], stdout=subprocess.PIPE, encoding='utf8')
66 kids = [int(line) for line in ps.stdout.splitlines()]
67 if len(kids) == 0:
68 print("no qemu")
69 sys.exit(1)
70 print("kill", kids[0])
71 os.kill(kids[0], signal.SIGKILL)
73 def stop(self):
74 self.proc.terminate()
76 def read(self):
77 while True:
78 try:
79 buf = os.read(self.proc.stdout.fileno(), 4096)
80 except BlockingIOError:
81 break
82 if len(buf) == 0: # qemu exited
83 break
84 self.outbytes.extend(buf)
85 self.output = self.outbytes.decode("utf-8", "replace")
87 def lines(self):
88 return self.output.splitlines()
90 def error(self, *regexps):
91 print("FAIL: match failed", regexps)
92 self.save_output()
93 self.stop()
94 sys.exit(1)
96 def match(self, *regexps, exit=True):
97 found = False
98 for line in self.lines():
99 if any(re.match(r, line) for r in regexps):
100 print(line)
101 found = True
102 if not found and exit:
103 self.error(*regexps)
104 return found
106 # Print the lines matching regexp that have arrived since the last
107 # call. A trailing partial line is left for the next call, so that
108 # each line is printed once, after all of it has been read.
109 def progress(self, regexp):
110 end = self.output.rfind("\n") + 1
111 if end <= self.reported:
112 return
113 for line in self.output[self.reported:end].splitlines():
114 if re.match(regexp, line):
115 print(line)
116 self.reported = end
118 def monitor(self, *regexps, progress="", timeout):
119 deadline = time.time() + timeout
120 while True:
121 time.sleep(1)
122 timeleft = deadline - time.time()
123 if timeleft < 0:
124 self.error(*regexps)
125 self.read()
126 if progress:
127 self.progress(progress)
128 if self.match(*regexps, exit=False):
129 return
131def crash_log():
132 q = QEMU(True)
133 q.cmd("logstress f0 f1 f2 f3 f4 f5\n")
134 time.sleep(2)
135 q.crash()
136 q.stop()
138def recover_log():
139 q = QEMU()
140 time.sleep(2)
141 q.read()
142 ok = q.match('^recovering', exit=False)
143 if ok:
144 q.cmd("ls\n")
145 q.monitor('f5', timeout=30)
146 q.stop()
147 return ok
149def forphan():
150 q = QEMU(True)
151 q.cmd("forphan\n")
152 q.monitor('wait', timeout=30)
153 q.crash()
154 q.stop()
156def dorphan():
157 q = QEMU(True)
158 q.cmd("dorphan\n")
159 q.monitor('wait', timeout=30)
160 q.crash()
161 q.stop()
163def recover_orphan():
164 q = QEMU()
165 q.monitor('^ireclaim', timeout=30)
166 q.stop()
168def test_log():
169 print("Test recovery of log")
170 for i in range(20):
171 crash_log()
172 ok = recover_log()
173 if ok:
174 print("OK")
175 return
176 print("log attempt ", i+1)
177 print("FAIL")
178 sys.exit(1)
180def test_forphan():
181 print("Test recovery of an orphaned file")
182 forphan()
183 recover_orphan()
184 print("OK")
186def test_dorphan():
187 print("Test recovery of an orphaned file")
188 dorphan()
189 recover_orphan()
190 print("OK")
192def test_crash():
193 test_log()
194 test_forphan()
195 test_dorphan()
197def test_usertests(test=""):
198 timeout = 600
199 opt = ""
200 if args.q:
201 opt = " -q"
202 timeout = 300
203 elif test != "":
204 opt += " " + test
205 q = QEMU(True)
206 q.cmd("usertests" + opt + "\n")
207 q.monitor('^ALL TESTS PASSED', progress='test', timeout=timeout)
208 q.stop()
210def main():
211 print(args)
212 rex = r'%s' % args.testrex
213 funcs = [(obj,name) for name,obj in inspect.getmembers(sys.modules[__name__])
214 if (inspect.isfunction(obj) and
215 name.startswith('test'))]
216 none = True
217 for (f,n) in funcs:
218 if re.search(rex, n):
219 none = False
220 f()
221 if none:
222 test_usertests(test=args.testrex)
224main()