circuitpython/tests/run-multitests.py

436 lines
14 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
# This file is part of the MicroPython project, http://micropython.org/
# The MIT License (MIT)
# Copyright (c) 2020 Damien P. George
import sys, os, time, re, select
import argparse
import subprocess
sys.path.append("../tools")
import pyboard
if os.name == "nt":
CPYTHON3 = os.getenv("MICROPY_CPYTHON3", "python3.exe")
MICROPYTHON = os.getenv("MICROPY_MICROPYTHON", "../ports/windows/micropython.exe")
else:
CPYTHON3 = os.getenv("MICROPY_CPYTHON3", "python3")
MICROPYTHON = os.getenv("MICROPY_MICROPYTHON", "../ports/unix/micropython")
PYTHON_TRUTH = CPYTHON3
INSTANCE_READ_TIMEOUT_S = 10
APPEND_CODE_TEMPLATE = """
import sys
class multitest:
@staticmethod
def flush():
if hasattr(sys.stdout, "flush"):
sys.stdout.flush()
@staticmethod
def skip():
print("SKIP")
multitest.flush()
raise SystemExit
@staticmethod
def next():
print("NEXT")
multitest.flush()
@staticmethod
def globals(**gs):
for g in gs:
print("SET {{}} = {{!r}}".format(g, gs[g]))
multitest.flush()
@staticmethod
def get_network_ip():
try:
import network
ip = network.WLAN().ifconfig()[0]
except:
ip = "127.0.0.1"
return ip
{}
instance{}()
multitest.flush()
"""
# The btstack implementation on Unix generates some spurious output that we
# can't control.
IGNORE_OUTPUT_MATCHES = (
"libusb: error ", # It tries to open devices that it doesn't have access to (libusb prints unconditionally).
"hci_transport_h2_libusb.c", # Same issue. We enable LOG_ERROR in btstack.
"USB Path: ", # Hardcoded in btstack's libusb transport.
"hci_number_completed_packet", # Warning from btstack.
)
class PyInstance:
def __init__(self):
pass
def close(self):
pass
def prepare_script_from_file(self, filename, prepend, append):
with open(filename, "rb") as f:
script = f.read()
if prepend:
script = bytes(prepend, "ascii") + b"\n" + script
if append:
script += b"\n" + bytes(append, "ascii")
return script
def run_file(self, filename, prepend="", append=""):
return self.run_script(self.prepare_script_from_file(filename, prepend, append))
def start_file(self, filename, prepend="", append=""):
return self.start_script(self.prepare_script_from_file(filename, prepend, append))
class PyInstanceSubProcess(PyInstance):
def __init__(self, cmd):
self.cmd = cmd
self.popen = None
self.finished = True
def __str__(self):
return self.cmd[0].rsplit("/")[-1]
def run_script(self, script):
output = b""
err = None
try:
p = subprocess.run(
self.cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, input=script
)
output = p.stdout
except subprocess.CalledProcessError as er:
err = er
return str(output.strip(), "ascii"), err
def start_script(self, script):
self.popen = subprocess.Popen(
self.cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
)
self.popen.stdin.write(script)
self.popen.stdin.close()
self.finished = False
def stop(self):
if self.popen and self.popen.poll() is None:
self.popen.terminate()
def readline(self):
sel = select.select([self.popen.stdout.raw], [], [], 0.1)
if not sel[0]:
self.finished = self.popen.poll() is not None
return None, None
out = self.popen.stdout.raw.readline()
if out == b"":
self.finished = self.popen.poll() is not None
return None, None
else:
return str(out.rstrip(), "ascii"), None
def is_finished(self):
return self.finished
def wait_finished(self):
self.popen.wait()
out = self.popen.stdout.read()
return str(out, "ascii"), ""
class PyInstancePyboard(PyInstance):
def __init__(self, device):
self.device = device
self.pyb = pyboard.Pyboard(device)
self.pyb.enter_raw_repl()
self.finished = True
def __str__(self):
return self.device.rsplit("/")[-1]
def close(self):
self.pyb.exit_raw_repl()
self.pyb.close()
def run_script(self, script):
output = b""
err = None
try:
self.pyb.enter_raw_repl()
output = self.pyb.exec_(script)
except pyboard.PyboardError as er:
err = er
return str(output.strip(), "ascii"), err
def start_script(self, script):
self.pyb.enter_raw_repl()
self.pyb.exec_raw_no_follow(script)
self.finished = False
def stop(self):
self.pyb.serial.write(b"\r\x03")
def readline(self):
if self.finished:
return None, None
if self.pyb.serial.inWaiting() == 0:
return None, None
out = self.pyb.read_until(1, (b"\r\n", b"\x04"))
if out.endswith(b"\x04"):
self.finished = True
out = out[:-1]
err = str(self.pyb.read_until(1, b"\x04"), "ascii")
err = err[:-1]
if not out and not err:
return None, None
else:
err = None
return str(out.rstrip(), "ascii"), err
def is_finished(self):
return self.finished
def wait_finished(self):
out, err = self.pyb.follow(10, None)
return str(out, "ascii"), str(err, "ascii")
def prepare_test_file_list(test_files):
test_files2 = []
for test_file in sorted(test_files):
num_instances = 0
with open(test_file) as f:
for line in f:
m = re.match(r"def instance([0-9]+)\(\):", line)
if m:
num_instances = max(num_instances, int(m.group(1)) + 1)
test_files2.append((test_file, num_instances))
return test_files2
def trace_instance_output(instance_idx, line):
if cmd_args.trace_output:
t_ms = round((time.time() - trace_t0) * 1000)
print("{:6} i{} :".format(t_ms, instance_idx), line)
def run_test_on_instances(test_file, num_instances, instances):
global trace_t0
trace_t0 = time.time()
error = False
skip = False
injected_globals = ""
output = [[] for _ in range(num_instances)]
if cmd_args.trace_output:
print("TRACE {}:".format("|".join(str(i) for i in instances)))
# Start all instances running, in order, waiting until they signal they are ready
for idx in range(num_instances):
append_code = APPEND_CODE_TEMPLATE.format(injected_globals, idx)
instance = instances[idx]
instance.start_file(test_file, append=append_code)
last_read_time = time.time()
while True:
if instance.is_finished():
break
out, err = instance.readline()
if out is None and err is None:
if time.time() > last_read_time + INSTANCE_READ_TIMEOUT_S:
output[idx].append("TIMEOUT")
error = True
break
time.sleep(0.1)
continue
last_read_time = time.time()
if out is not None and not any(m in out for m in IGNORE_OUTPUT_MATCHES):
trace_instance_output(idx, out)
if out.startswith("SET "):
injected_globals += out[4:] + "\n"
elif out == "SKIP":
skip = True
break
elif out == "NEXT":
break
else:
output[idx].append(out)
if err is not None:
trace_instance_output(idx, err)
output[idx].append(err)
error = True
if error or skip:
break
if not error and not skip:
# Capture output and wait for all instances to finish running
last_read_time = [time.time() for _ in range(num_instances)]
while True:
num_running = 0
num_output = 0
for idx in range(num_instances):
instance = instances[idx]
if instance.is_finished():
continue
num_running += 1
out, err = instance.readline()
if out is None and err is None:
if time.time() > last_read_time[idx] + INSTANCE_READ_TIMEOUT_S:
output[idx].append("TIMEOUT")
error = True
continue
num_output += 1
last_read_time[idx] = time.time()
if out is not None:
trace_instance_output(idx, out)
output[idx].append(out)
if err is not None:
trace_instance_output(idx, err)
output[idx].append(err)
error = True
if not num_output:
time.sleep(0.1)
if not num_running or error:
break
# Stop all instances
for idx in range(num_instances):
instances[idx].stop()
output_str = ""
for idx, lines in enumerate(output):
output_str += "--- instance{} ---\n".format(idx)
output_str += "\n".join(lines) + "\n"
return error, skip, output_str
def run_tests(test_files, instances_truth, instances_test):
skipped_tests = []
passed_tests = []
failed_tests = []
for test_file, num_instances in test_files:
instances_str = "|".join(str(instances_test[i]) for i in range(num_instances))
print("{} on {}: ".format(test_file, instances_str), end="")
if cmd_args.show_output or cmd_args.trace_output:
print()
sys.stdout.flush()
# Run test on test instances
error, skip, output_test = run_test_on_instances(test_file, num_instances, instances_test)
if not skip:
# Check if truth exists in a file, and read it in
test_file_expected = test_file + ".exp"
if os.path.isfile(test_file_expected):
with open(test_file_expected) as f:
output_truth = f.read()
else:
# Run test on truth instances to get expected output
_, _, output_truth = run_test_on_instances(
test_file, num_instances, instances_truth
)
if cmd_args.show_output:
print("### TEST ###")
print(output_test, end="")
if not skip:
print("### TRUTH ###")
print(output_truth, end="")
# Print result of test
if skip:
print("skip")
skipped_tests.append(test_file)
elif output_test == output_truth:
print("pass")
passed_tests.append(test_file)
else:
print("FAIL")
failed_tests.append(test_file)
if not cmd_args.show_output:
print("### TEST ###")
print(output_test, end="")
print("### TRUTH ###")
print(output_truth, end="")
if cmd_args.show_output:
print()
print("{} tests performed".format(len(skipped_tests) + len(passed_tests) + len(failed_tests)))
print("{} tests passed".format(len(passed_tests)))
if skipped_tests:
print("{} tests skipped: {}".format(len(skipped_tests), " ".join(skipped_tests)))
if failed_tests:
print("{} tests failed: {}".format(len(failed_tests), " ".join(failed_tests)))
return not failed_tests
def main():
global cmd_args
cmd_parser = argparse.ArgumentParser(description="Run network tests for MicroPython")
cmd_parser.add_argument(
"-s", "--show-output", action="store_true", help="show test output after running"
)
cmd_parser.add_argument(
"-t", "--trace-output", action="store_true", help="trace test output while running"
)
cmd_parser.add_argument(
"-i", "--instance", action="append", default=[], help="instance(s) to run the tests on"
)
cmd_parser.add_argument("files", nargs="+", help="input test files")
cmd_args = cmd_parser.parse_args()
# clear search path to make sure tests use only builtin modules and those in extmod
os.environ["MICROPYPATH"] = os.pathsep + "../extmod"
test_files = prepare_test_file_list(cmd_args.files)
max_instances = max(t[1] for t in test_files)
instances_truth = [PyInstanceSubProcess([PYTHON_TRUTH]) for _ in range(max_instances)]
instances_test = []
for i in cmd_args.instance:
if i.startswith("exec:"):
instances_test.append(PyInstanceSubProcess([i[len("exec:") :]]))
elif i == "micropython":
instances_test.append(PyInstanceSubProcess([MICROPYTHON]))
elif i == "cpython":
instances_test.append(PyInstanceSubProcess([CPYTHON3]))
elif i.startswith("pyb:"):
instances_test.append(PyInstancePyboard(i[len("pyb:") :]))
else:
print("unknown instance string: {}".format(i), file=sys.stderr)
sys.exit(1)
for _ in range(max_instances - len(instances_test)):
instances_test.append(PyInstanceSubProcess([MICROPYTHON]))
try:
all_pass = run_tests(test_files, instances_truth, instances_test)
finally:
for i in instances_truth:
i.close()
for i in instances_test:
i.close()
if not all_pass:
sys.exit(1)
if __name__ == "__main__":
main()