sanitycheck: Complete overhaul and job handling rework
A complete overhaul of the sanitycheck script and how we build and run
tests. This new version of sanitycheck uses python for job distribution
and drop use of Make.
In addition to the move to python threading library, the following has
been changed:
- All handlers now run in parallel, meaning that any simulator will run
in parallel and when testing on multiple devices (using
--device-testing) the tests are run in parallel.
- Lexicial filtering (using the filter keyword in yaml files) is now
evaluated at runtime and is no long being pre-processed. This will allow
us to immediately start executing tests and skip the wait time that was
needed for filtering.
- Device testing now supports multiple devices connected at the same
time and is managed using a hardware map that needs to be generated and
maintained for every test environment. (using --generate-hardware-map
option).
- Reports are not long stored in the Zephyr tree and instead stored in
the output directory where all build artifacts are generated.
- Each tested target now has a junit report in the output directory.
- Recording option for performance data and other metrics is now
available. This will allow us to record the output from the console and
store the data for later processing. For example benchmark data can be
captured and uploaded to a tracking server.
- Test configurations (or instances) are no longer being sorted, this
will help with balancing the load when we run sanitycheck on multiple
hosts (as we do in CI).
And many other cleanups and improvements...
Signed-off-by: Anas Nashif <anas.nashif@intel.com>
diff --git a/scripts/sanitycheck b/scripts/sanitycheck
index fd213f1..e90c029 100755
--- a/scripts/sanitycheck
+++ b/scripts/sanitycheck
@@ -9,7 +9,7 @@
repository and attempts to execute them. By default, it tries to
build each test case on one platform per architecture, using a precedence
list defined in an architecture configuration file, and if possible
-run the tests in the QEMU emulator.
+run the tests in any available emulators or simulators on the system.
Test cases are detected by the presence of a 'testcase.yaml' or a sample.yaml
files in the application's project directory. This file may contain one or more
@@ -63,7 +63,7 @@
compared with information provided by the board metadata.
timeout: <number of seconds>
- Length of time to run test in QEMU before automatically killing it.
+ Length of time to run test in emulator before automatically killing it.
Default to 60 seconds.
arch_whitelist: <list of arches, such as x86, arm, arc>
@@ -183,25 +183,27 @@
import shlex
import signal
import threading
+import concurrent.futures
+from threading import BoundedSemaphore
+import queue
import time
import datetime
import csv
+import yaml
import glob
import serial
import concurrent
-import concurrent.futures
import xml.etree.ElementTree as ET
-import resource
-from xml.sax.saxutils import escape
from collections import OrderedDict
from itertools import islice
-from functools import cmp_to_key
from pathlib import Path
from distutils.spawn import find_executable
import logging
-from sanity_chk import scl
-from sanity_chk import expr_parser
+
+
+hw_map_local = threading.Lock()
+
log_format = "%(levelname)s %(name)s::%(module)s.%(funcName)s():%(lineno)d: %(message)s"
logging.basicConfig(format=log_format, level=30)
@@ -221,12 +223,12 @@
sys.path.insert(0, os.path.join(ZEPHYR_BASE, "scripts/"))
+from sanity_chk import scl
+from sanity_chk import expr_parser
+
VERBOSE = 0
-LAST_SANITY = os.path.join(ZEPHYR_BASE, "scripts", "sanity_chk",
- "last_sanity.csv")
-LAST_SANITY_XUNIT = os.path.join(ZEPHYR_BASE, "scripts", "sanity_chk",
- "last_sanity.xml")
+
RELEASE_DATA = os.path.join(ZEPHYR_BASE, "scripts", "sanity_chk",
"sanity_last_release.csv")
@@ -320,7 +322,7 @@
except ValueError as exc:
args = exc.args + ('on line {}: {}'.format(line_no, line),)
raise ValueError(args) from exc
- elif type_ == 'STRING' or type_ == 'INTERNAL':
+ elif type_ in ['STRING','INTERNAL']:
# If the value is a CMake list (i.e. is a string which
# contains a ';'), convert to a Python list.
if ';' in value:
@@ -405,25 +407,15 @@
class SanityRuntimeError(SanityCheckException):
pass
-
class ConfigurationError(SanityCheckException):
def __init__(self, cfile, message):
- self.cfile = cfile
- self.message = message
+ SanityCheckException.__init__(self, cfile + ": " + message)
- def __str__(self):
- return repr(self.cfile + ": " + self.message)
-
-
-class MakeError(SanityCheckException):
+class BuildError(SanityCheckException):
pass
-class BuildError(MakeError):
- pass
-
-
-class ExecutionError(MakeError):
+class ExecutionError(SanityCheckException):
pass
@@ -476,20 +468,12 @@
def __init__(self, instance, type_str="build"):
"""Constructor
- @param name Arbitrary name of the created thread
- @param outdir Working directory, should be where handler pid file (qemu.pid for example)
- gets created by the build system
- @param log_fn Absolute path to write out handler's log data
- @param timeout Kill the handler process if it doesn't finish up within
- the given number of seconds
"""
self.lock = threading.Lock()
+
self.state = "waiting"
self.run = False
- self.metrics = {}
- self.metrics["handler_time"] = 0
- self.metrics["ram_size"] = 0
- self.metrics["rom_size"] = 0
+ self.duration = 0
self.type_str = type_str
self.binary = None
@@ -498,25 +482,38 @@
self.name = instance.name
self.instance = instance
- self.timeout = instance.test.timeout
- self.sourcedir = instance.test.test_path
- self.outdir = instance.outdir
- self.log = os.path.join(self.outdir, "handler.log")
+ self.timeout = instance.testcase.timeout
+ self.sourcedir = instance.testcase.source_dir
+ self.build_dir = instance.build_dir
+ self.log = os.path.join(self.build_dir, "handler.log")
self.returncode = 0
- self.set_state("running", {})
+ self.set_state("running", self.duration)
- def set_state(self, state, metrics):
+ self.args = []
+
+ def set_state(self, state, duration):
self.lock.acquire()
self.state = state
- self.metrics.update(metrics)
+ self.duration = duration
self.lock.release()
def get_state(self):
self.lock.acquire()
- ret = (self.state, self.metrics)
+ ret = (self.state, self.duration)
self.lock.release()
return ret
+ def record(self, harness):
+ if harness.recording:
+ filename = os.path.join(options.outdir,
+ self.instance.platform.name,
+ self.instance.testcase.name, "recording.csv")
+ with open(filename, "at") as csvfile:
+ cw = csv.writer(csvfile, harness.fieldnames, lineterminator=os.linesep)
+ cw.writerow(harness.fieldnames)
+ for instance in harness.recording:
+ cw.writerow(instance)
+
class BinaryHandler(Handler):
def __init__(self, instance, type_str):
"""Constructor
@@ -529,7 +526,7 @@
self.terminated = False
def try_kill_process_by_pid(self):
- if self.pid_fn != None:
+ if self.pid_fn:
pid = int(open(self.pid_fn).read())
os.unlink(self.pid_fn)
self.pid_fn = None # clear so we don't try to kill the binary twice
@@ -559,17 +556,13 @@
def handle(self):
- harness_name = self.instance.test.harness.capitalize()
+ harness_name = self.instance.testcase.harness.capitalize()
harness_import = HarnessImporter(harness_name)
harness = harness_import.instance
harness.configure(self.instance)
if self.call_make_run:
- if options.ninja:
- generator_cmd = "ninja"
- else:
- generator_cmd = "make"
- command = [generator_cmd, "run"]
+ command = [get_generator()[0], "run"]
else:
command = [self.binary]
@@ -577,16 +570,16 @@
command = ["valgrind", "--error-exitcode=2",
"--leak-check=full",
"--suppressions="+ZEPHYR_BASE+"/scripts/valgrind.supp",
- "--log-file="+self.outdir+"/valgrind.log"
+ "--log-file="+self.build_dir+"/valgrind.log"
] + command
verbose("Spawning process: " +
" ".join(shlex.quote(word) for word in command) + os.linesep +
- "Spawning process in directory: " + self.outdir)
+ "Spawning process in directory: " + self.build_dir)
- start_time = time.time();
+ start_time = time.time()
- with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self.outdir) as proc:
+ with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self.build_dir) as proc:
verbose("Spawning BinaryHandler Thread for %s" % self.name)
t = threading.Thread(target=self._output_reader, args=(proc, harness, ))
t.start()
@@ -599,27 +592,31 @@
proc.wait()
self.returncode = proc.returncode
- self.metrics["handler_time"] = time.time() - start_time;
+ handler_time = time.time() - start_time
if options.enable_coverage:
- returncode = subprocess.call(["GCOV_PREFIX=" + self.outdir,
- "gcov", self.sourcedir, "-b", "-s", self.outdir], shell=True)
+ subprocess.call(["GCOV_PREFIX=" + self.build_dir,
+ "gcov", self.sourcedir, "-b", "-s", self.build_dir], shell=True)
self.try_kill_process_by_pid()
- # FIME: This is needed when killing the simulator, the console is
+ # FIXME: This is needed when killing the simulator, the console is
# garbled and needs to be reset. Did not find a better way to do that.
subprocess.call(["stty", "sane"])
self.instance.results = harness.tests
- if self.terminated==False and self.returncode != 0:
+ if not self.terminated and self.returncode != 0:
#When a process is killed, the default handler returns 128 + SIGTERM
#so in that case the return code itself is not meaningful
- self.set_state("error", {})
+ self.set_state("error", handler_time)
+ self.instance.reason = "Handler error"
elif harness.state:
- self.set_state(harness.state, {})
+ self.set_state(harness.state, handler_time)
else:
- self.set_state("timeout", {})
+ self.set_state("timeout", handler_time)
+ self.instance.reason = "Handler timeout"
+
+ self.record(harness)
class DeviceHandler(Handler):
@@ -630,6 +627,8 @@
"""
super().__init__(instance, type_str)
+ self.suite = None
+
def monitor_serial(self, ser, halt_fileno, harness):
log_out_fp = open(self.log, "wt")
@@ -651,7 +650,7 @@
serial_line = ser.readline()
except TypeError:
pass
- except serial.serialutil.SerialException:
+ except serial.SerialException:
ser.close()
break
@@ -671,12 +670,34 @@
log_out_fp.close()
+ def device_is_available(self, device):
+ for i in self.suite.connected_hardware:
+ if i['platform'] == device and i['available'] and i['connected']:
+ return True
+
+ return False
+
+ def get_available_device(self, device):
+ for i in self.suite.connected_hardware:
+ if i['platform'] == device and i['available']:
+ i['available'] = False
+ i['counter'] += 1
+ return i
+
+ return None
+
+ def make_device_available(self, serial):
+ with hw_map_local:
+ for i in self.suite.connected_hardware:
+ if i['serial'] == serial:
+ i['available'] = True
+
def handle(self):
out_state = "failed"
- if options.west_flash is not None:
- command = ["west", "flash", "--skip-rebuild", "-d", self.outdir]
- if options.west_runner is not None:
+ if options.west_flash:
+ command = ["west", "flash", "--skip-rebuild", "-d", self.build_dir]
+ if options.west_runner:
command.append("--runner")
command.append(options.west_runner)
# There are three ways this option is used.
@@ -690,43 +711,78 @@
command.append('--')
command.extend(options.west_flash.split(','))
else:
- if options.ninja:
- generator_cmd = "ninja"
- else:
- generator_cmd = "make"
+ command = [get_generator()[0], "-C", self.build_dir, "flash"]
- command = [generator_cmd, "-C", self.outdir, "flash"]
- device = options.device_serial
- ser = serial.Serial(
- device,
- baudrate=115200,
- parity=serial.PARITY_NONE,
- stopbits=serial.STOPBITS_ONE,
- bytesize=serial.EIGHTBITS,
- timeout=self.timeout
- )
+ while not self.device_is_available(self.instance.platform.name):
+ time.sleep(1)
+
+ hardware = self.get_available_device(self.instance.platform.name)
+
+ runner = hardware.get('runner', None)
+ if runner:
+ board_id = hardware.get("id", None)
+ product = hardware.get("product", None)
+ command = ["west", "flash", "--skip-rebuild", "-d", self.build_dir]
+ command.append("--runner")
+ command.append(hardware.get('runner', None))
+ if runner == "pyocd":
+ command.append("--board-id")
+ command.append(board_id)
+ elif runner == "nrfjprog":
+ command.append('--')
+ command.append("--snr")
+ command.append(board_id)
+ elif runner == "openocd" and product == "STM32 STLink":
+ command.append('--')
+ command.append("--cmd-pre-init")
+ command.append("hla_serial %s" %(board_id))
+ elif runner == "openocd" and product == "EDBG CMSIS-DAP":
+ command.append('--')
+ command.append("--cmd-pre-init")
+ command.append("cmsis_dap_serial %s" %(board_id))
+ elif runner == "jlink":
+ command.append("--tool-opt=-SelectEmuBySN %s" %(board_id))
+
+ serial_device = hardware['serial']
+
+ try:
+ ser = serial.Serial(
+ serial_device,
+ baudrate=115200,
+ parity=serial.PARITY_NONE,
+ stopbits=serial.STOPBITS_ONE,
+ bytesize=serial.EIGHTBITS,
+ timeout=self.timeout
+ )
+ except serial.SerialException as e:
+ self.set_state("failed", 0)
+ error("Serial device err: %s" %(str(e)))
+ self.make_device_available(serial_device)
+ return
ser.flush()
- harness_name = self.instance.test.harness.capitalize()
+ harness_name = self.instance.testcase.harness.capitalize()
harness_import = HarnessImporter(harness_name)
harness = harness_import.instance
harness.configure(self.instance)
- rpipe, wpipe = os.pipe()
+ read_pipe, write_pipe = os.pipe()
+ start_time = time.time()
t = threading.Thread(target=self.monitor_serial, daemon=True,
- args=(ser, rpipe, harness))
+ args=(ser, read_pipe, harness))
t.start()
logging.debug('Flash command: %s', command)
try:
- if VERBOSE:
+ if VERBOSE and not runner:
subprocess.check_call(command)
else:
subprocess.check_output(command, stderr=subprocess.PIPE)
+
except subprocess.CalledProcessError:
- os.write(wpipe, b'x') # halt the thread
+ os.write(write_pipe, b'x') # halt the thread
t.join(self.timeout)
if t.is_alive():
@@ -736,16 +792,20 @@
ser.close()
if out_state == "timeout":
- for c in self.instance.test.cases:
+ for c in self.instance.testcase.cases:
if c not in harness.tests:
harness.tests[c] = "BLOCK"
+
+ handler_time = time.time() - start_time
+
self.instance.results = harness.tests
if harness.state:
- self.set_state(harness.state, {})
+ self.set_state(harness.state, handler_time)
else:
- self.set_state(out_state, {})
+ self.set_state(out_state, handler_time)
+ self.make_device_available(serial_device)
class QEMUHandler(Handler):
"""Spawns a thread to monitor QEMU output from pipes
@@ -756,6 +816,19 @@
for these to collect whether the test passed or failed.
"""
+
+ def __init__(self, instance, type_str):
+ """Constructor
+
+ @param instance Test instance
+ """
+
+ super().__init__(instance, type_str)
+ self.fifo_fn = os.path.join(instance.build_dir, "qemu-fifo")
+
+ self.pid_fn = os.path.join(instance.build_dir, "qemu.pid")
+
+
@staticmethod
def _thread(handler, timeout, outdir, logfile, fifo_fn, pid_fn, results, harness):
fifo_in = fifo_fn + ".in"
@@ -783,7 +856,6 @@
p.register(in_fp, select.POLLIN)
out_state = None
- metrics = {}
line = ""
timeout_extended = False
while True:
@@ -833,16 +905,14 @@
timeout_time = time.time() + 30
else:
timeout_time = time.time() + 2
-
- # TODO: Add support for getting numerical performance data
- # from test cases. Will involve extending test case reporting
- # APIs. Add whatever gets reported to the metrics dictionary
line = ""
- metrics["handler_time"] = time.time() - start_time
+ handler.record(harness)
+
+ handler_time = time.time() - start_time
verbose("QEMU complete (%s) after %f seconds" %
- (out_state, metrics["handler_time"]))
- handler.set_state(out_state, metrics)
+ (out_state, handler_time))
+ handler.set_state(out_state, handler_time)
log_out_fp.close()
out_fp.close()
@@ -861,70 +931,115 @@
os.unlink(fifo_in)
os.unlink(fifo_out)
- def __init__(self, instance, type_str):
- """Constructor
-
- @param instance Test instance
- """
-
- super().__init__(instance, type_str)
-
+ def handle(self):
self.results = {}
self.run = True
# We pass this to QEMU which looks for fifos with .in and .out
# suffixes.
- self.fifo_fn = os.path.join(instance.outdir, "qemu-fifo")
+ self.fifo_fn = os.path.join(self.instance.build_dir, "qemu-fifo")
- self.pid_fn = os.path.join(instance.outdir, "qemu.pid")
+ self.pid_fn = os.path.join(self.instance.build_dir, "qemu.pid")
if os.path.exists(self.pid_fn):
os.unlink(self.pid_fn)
self.log_fn = self.log
- harness_import = HarnessImporter(instance.test.harness.capitalize())
+ harness_import = HarnessImporter(self.instance.testcase.harness.capitalize())
harness = harness_import.instance
harness.configure(self.instance)
self.thread = threading.Thread(name=self.name, target=QEMUHandler._thread,
- args=(self, self.timeout, self.outdir,
+ args=(self, self.timeout, self.build_dir,
self.log_fn, self.fifo_fn,
self.pid_fn, self.results, harness))
self.instance.results = harness.tests
self.thread.daemon = True
- verbose("Spawning QEMUHandler Thread for %s 'make run'" % self.name)
+ verbose("Spawning QEMUHandler Thread for %s" % self.name)
self.thread.start()
+ subprocess.call(["stty", "sane"])
def get_fifo(self):
return self.fifo_fn
-
class SizeCalculator:
- alloc_sections = ["bss", "noinit", "app_bss", "app_noinit", "ccm_bss",
- "ccm_noinit"]
- rw_sections = ["datas", "initlevel", "exceptions", "initshell",
- "_static_thread_area", "_k_timer_area",
- "_k_mem_slab_area", "_k_mem_pool_area", "sw_isr_table",
- "_k_sem_area", "_k_mutex_area", "app_shmem_regions",
- "_k_fifo_area", "_k_lifo_area", "_k_stack_area",
- "_k_msgq_area", "_k_mbox_area", "_k_pipe_area",
- "net_if", "net_if_dev", "net_stack", "net_l2_data",
- "_k_queue_area", "_net_buf_pool_area", "app_datas",
- "kobject_data", "mmu_tables", "app_pad", "priv_stacks",
- "ccm_data", "usb_descriptor", "usb_data", "usb_bos_desc",
- 'log_backends_sections', 'log_dynamic_sections',
- 'log_const_sections',"app_smem", 'shell_root_cmds_sections',
- 'log_const_sections',"app_smem", "font_entry_sections",
- "priv_stacks_noinit", "_TEXT_SECTION_NAME_2",
- '_GCOV_BSS_SECTION_NAME', 'gcov', 'nocache']
+ alloc_sections = [
+ "bss",
+ "noinit",
+ "app_bss",
+ "app_noinit",
+ "ccm_bss",
+ "ccm_noinit"
+ ]
+
+ rw_sections = [
+ "datas",
+ "initlevel",
+ "exceptions",
+ "initshell",
+ "_static_thread_area",
+ "_k_timer_area",
+ "_k_mem_slab_area",
+ "_k_mem_pool_area",
+ "sw_isr_table",
+ "_k_sem_area",
+ "_k_mutex_area",
+ "app_shmem_regions",
+ "_k_fifo_area",
+ "_k_lifo_area",
+ "_k_stack_area",
+ "_k_msgq_area",
+ "_k_mbox_area",
+ "_k_pipe_area",
+ "net_if",
+ "net_if_dev",
+ "net_stack",
+ "net_l2_data",
+ "_k_queue_area",
+ "_net_buf_pool_area",
+ "app_datas",
+ "kobject_data",
+ "mmu_tables",
+ "app_pad",
+ "priv_stacks",
+ "ccm_data",
+ "usb_descriptor",
+ "usb_data", "usb_bos_desc",
+ 'log_backends_sections',
+ 'log_dynamic_sections',
+ 'log_const_sections',
+ "app_smem",
+ 'shell_root_cmds_sections',
+ 'log_const_sections',
+ "font_entry_sections",
+ "priv_stacks_noinit",
+ "_TEXT_SECTION_NAME_2",
+ "_GCOV_BSS_SECTION_NAME",
+ "gcov",
+ "nocache"
+ ]
# These get copied into RAM only on non-XIP
- ro_sections = ["text", "ctors", "init_array", "reset", "object_access",
- "rodata", "devconfig", "net_l2", "vector", "sw_isr_table",
- "_settings_handlers_area", "_bt_channels_area",
- "_bt_br_channels_area", "_bt_services_area",
- "vectors", "net_socket_register", "net_ppp_proto"]
+ ro_sections = [
+ "text",
+ "ctors",
+ "init_array",
+ "reset",
+ "object_access",
+ "rodata",
+ "devconfig",
+ "net_l2",
+ "vector",
+ "sw_isr_table",
+ "_settings_handlers_area",
+ "_bt_channels_area",
+ "_bt_br_channels_area",
+ "_bt_services_area",
+ "vectors",
+ "net_socket_register",
+ "net_ppp_proto"
+ ]
def __init__(self, filename, extra_sections):
"""Constructor
@@ -937,7 +1052,7 @@
magic = f.read(4)
try:
- if (magic != b'\x7fELF'):
+ if magic != b'\x7fELF':
raise SanityRuntimeError("%s is not an ELF binary" % filename)
except Exception as e:
print(str(e))
@@ -1002,15 +1117,15 @@
for line in objdump_output:
words = line.split()
- if (len(words) == 0): # Skip lines that are too short
+ if not words: # Skip lines that are too short
continue
index = words[0]
- if (not index[0].isdigit()): # Skip lines that do not start
+ if not index[0].isdigit(): # Skip lines that do not start
continue # with a digit
name = words[1] # Skip lines with section names
- if (name[0] == '.'): # starting with '.'
+ if name[0] == '.': # starting with '.'
continue
# TODO this doesn't actually reflect the size in flash or RAM as
@@ -1048,438 +1163,6 @@
"type": stype, "recognized": recognized})
-class MakeGoal:
- """Metadata class representing one of the sub-makes called by MakeGenerator
-
- MakeGenerator returns a dictionary of these which can then be associated
- with TestInstances to get a complete picture of what happened during a test.
- MakeGenerator is used for tasks outside of building tests (such as
- defconfigs) which is why MakeGoal is a separate class from TestInstance.
- """
-
- def __init__(self, name, text, handler, make_log, build_log, run_log, handler_log):
- self.name = name
- self.text = text
- self.handler = handler
- self.make_log = make_log
- self.build_log = build_log
- self.run_log = run_log
- self.handler_log = handler_log
- self.make_state = "waiting"
- self.failed = False
- self.finished = False
- self.reason = None
- self.metrics = {}
-
- def get_error_log(self):
- if self.make_state == "waiting":
- # Shouldn't ever see this; breakage in the main Makefile itself.
- return self.make_log
- elif self.make_state == "building":
- # Failure when calling the sub-make to build the code
- return self.build_log
- elif self.make_state == "running":
- # Failure in sub-make for "make run", qemu probably failed.
- # Return qemu's output if there is one, otherwise make's.
- h = Path(self.handler_log)
- if h.exists() and h.stat().st_size > 0:
- return self.handler_log
- else:
- return self.run_log
- elif self.make_state == "finished":
- # Execution handler finished, but timed out or otherwise wasn't successful
- return self.handler_log
-
- def fail(self, reason):
- self.failed = True
- self.finished = True
- self.reason = reason
-
- def success(self):
- self.finished = True
-
- def __str__(self):
- if self.finished:
- if self.failed:
- return "[%s] failed (%s: see %s)" % (self.name, self.reason,
- self.get_error_log())
- else:
- return "[%s] passed" % self.name
- else:
- return "[%s] in progress (%s)" % (self.name, self.make_state)
-
-
-class MakeGenerator:
- """Generates a Makefile which just calls a bunch of sub-make sessions
-
- In any given test suite we may need to build dozens if not hundreds of
- test cases. The cleanest way to parallelize this is to just let Make
- do the parallelization, sharing the jobserver among all the different
- sub-make targets.
- """
-
- GOAL_HEADER_TMPL = """.PHONY: {goal}
-{goal}:
-"""
-
- MAKE_RULE_TMPL_CMAKE = """\t@echo sanity_test_{phase} {goal} >&2
-\tcmake \\
-\t\t-G"{generator}"\\
-\t\t-S{directory}\\
-\t\t-B{outdir}\\
-\t\t-DEXTRA_CFLAGS="-Werror {cflags}"\\
-\t\t-DEXTRA_AFLAGS=-Wa,--fatal-warnings\\
-\t\t-DEXTRA_LDFLAGS="{ldflags}"\\
-\t\t{args}\\
-\t\t>{logfile} 2>&1
-"""
- MAKE_RULE_TMPL_BLD = """\t{generator_cmd} -C {outdir}\\
-\t\t{verb} {make_args}\\
-\t\t>>{logfile} 2>&1
-"""
- MAKE_RULE_TMPL_RUN = """\t@echo sanity_test_{phase} {goal} >&2
-\t{generator_cmd} -C {outdir}\\
-\t\t{verb} {make_args}\\
-\t\t>>{logfile} 2>&1
-"""
-
- GOAL_FOOTER_TMPL = "\t@echo sanity_test_finished {goal} >&2\n\n"
-
- re_make = re.compile(
- "sanity_test_([A-Za-z0-9]+) (.+)|$|make[:] \*\*\* \[(.+:.+: )?(.+)\] Error.+$")
-
- def __init__(self, base_outdir):
- """MakeGenerator constructor
-
- @param base_outdir Intended to be the base out directory. A make.log
- file will be created here which contains the output of the
- top-level Make session, as well as the dynamic control Makefile
- @param verbose If true, pass V=1 to all the sub-makes which greatly
- increases their verbosity
- """
- self.goals = {}
- if not os.path.exists(base_outdir):
- os.makedirs(base_outdir)
- self.logfile = os.path.join(base_outdir, "make.log")
- self.makefile = os.path.join(base_outdir, "Makefile")
- self.deprecations = options.error_on_deprecations
-
- def _get_rule_header(self, name):
- return MakeGenerator.GOAL_HEADER_TMPL.format(goal=name)
-
- def _get_sub_make(self, name, phase, workdir, outdir,
- logfile, args, make_args=""):
- """
- @param args Arguments given to CMake
- @param make_args Arguments given to the Makefile generated by CMake
- """
- args = " ".join(["-D{}".format(a) for a in args])
- ldflags = ""
- cflags = ""
-
- if self.deprecations:
- cflags = cflags + " -Wno-deprecated-declarations"
-
- ldflags="-Wl,--fatal-warnings"
-
- if options.ninja:
- generator = "Ninja"
- generator_cmd = "ninja -j1"
- verb = "-v" if VERBOSE else ""
- else:
- generator = "Unix Makefiles"
- generator_cmd = "$(MAKE)"
- verb = "VERBOSE=1" if VERBOSE else ""
-
- if phase == 'running':
- return MakeGenerator.MAKE_RULE_TMPL_RUN.format(
- generator_cmd=generator_cmd,
- phase=phase,
- goal=name,
- outdir=outdir,
- verb=verb,
- logfile=logfile,
- make_args=make_args
- )
- else:
- cmake_rule = MakeGenerator.MAKE_RULE_TMPL_CMAKE.format(
- generator=generator,
- phase=phase,
- goal=name,
- outdir=outdir,
- cflags=cflags,
- ldflags=ldflags,
- directory=workdir,
- args=args,
- logfile=logfile,
- )
-
- if options.cmake_only:
- build_rule = ""
- else:
- build_rule = MakeGenerator.MAKE_RULE_TMPL_BLD.format(
- generator_cmd=generator_cmd,
- outdir=outdir,
- verb=verb,
- make_args=make_args,
- logfile=logfile,
- )
-
- return cmake_rule + build_rule
-
- def _get_rule_footer(self, name):
- return MakeGenerator.GOAL_FOOTER_TMPL.format(goal=name)
-
- def add_build_goal(self, name, directory, outdir,
- args, buildlog, make_args=""):
- """Add a goal to invoke a build session
-
- @param name A unique string name for this build goal. The results
- dictionary returned by execute() will be keyed by this name.
- @param directory Absolute path to working directory, will be passed
- to make -C
- @param outdir Absolute path to output directory, will be passed to
- cmake via -B=<path>
- @param args Extra command line arguments to pass to 'cmake', typically
- environment variables or specific Make goals
- """
-
- if not os.path.exists(outdir):
- os.makedirs(outdir)
-
- build_logfile = os.path.join(outdir, buildlog)
- text = self._get_rule_header(name)
- text += self._get_sub_make(name, "building", directory, outdir, build_logfile,
- args, make_args=make_args)
- text += self._get_rule_footer(name)
-
- self.goals[name] = MakeGoal( name, text, None, self.logfile, build_logfile, None, None)
-
- def add_goal(self, instance, type, args, make_args=""):
-
- """Add a goal to build a Zephyr project and then run it using a handler
-
- The generated make goal invokes Make twice, the first time it will
- build the default goal, and the second will invoke the 'qemu' goal.
- The output of the handler session will be monitored, and terminated
- either upon pass/fail result of the test program, or the timeout
- is reached.
-
- @param args Extra cache entries to define in CMake.
- """
-
- name = instance.name
- directory = instance.test.test_path
- outdir = instance.outdir
-
- build_logfile = os.path.join(outdir, "build.log")
- run_logfile = os.path.join(outdir, "run.log")
-
- if not os.path.exists(outdir):
- os.makedirs(outdir)
-
- handler = None
- if type == "qemu":
- handler = QEMUHandler(instance, "qemu")
- elif type == "native":
- handler = BinaryHandler(instance, "native")
- # defined by __build_dir in cmake/boilerplate.cmake
- handler.binary = os.path.join(outdir, "zephyr", "zephyr.exe")
- elif type == "nsim":
- handler = BinaryHandler(instance, "nsim")
- handler.call_make_run = True
- elif type == "unit":
- handler = BinaryHandler(instance, "unit")
- handler.binary = os.path.join(outdir, "testbinary")
- if options.enable_coverage:
- args += ["EXTRA_LDFLAGS=--coverage"]
- elif type == "device":
- handler = DeviceHandler(instance, "device")
- elif type == "renode":
- handler = BinaryHandler(instance, "renode")
- handler.pid_fn = os.path.join(instance.outdir, "renode.pid")
- handler.call_make_run = True
-
- if type == 'qemu':
- args.append("QEMU_PIPE=%s" % handler.get_fifo())
-
- text = self._get_rule_header(name)
- if not options.test_only:
- text += self._get_sub_make(name, "building", directory, outdir,
- build_logfile, args, make_args=make_args)
- if handler and handler.run:
- text += self._get_sub_make(name, "running", directory,
- outdir, run_logfile,
- args, make_args="run")
-
- text += self._get_rule_footer(name)
-
- self.goals[name] = MakeGoal(name, text, handler, self.logfile, build_logfile,
- run_logfile, handler.log if handler else None)
-
-
- def add_test_instance(self, ti, extra_args=[]):
- """Add a goal to build/test a TestInstance object
-
- @param ti TestInstance object to build. The status dictionary returned
- by execute() will be keyed by its .name field.
- """
- args = ti.test.extra_args[:]
-
- # merge overlay files into one variable
- overlays = ""
- idx = 0
- for a in args:
- m = re.search('OVERLAY_CONFIG="(.*)"', a)
- if m:
- overlays += m.group(1)
- del args[idx]
- idx += 1
-
- if len(ti.test.extra_configs) > 0 or options.coverage:
- args.append("OVERLAY_CONFIG=\"%s %s\"" %(overlays,
- os.path.join(ti.outdir,
- "sanitycheck", "testcase_extra.conf")))
-
- if ti.test.type == "unit" and options.enable_coverage:
- args.append("COVERAGE=1")
-
- args.append("BOARD={}".format(ti.platform.name))
- args.extend(extra_args)
-
- do_build_only = ti.build_only or options.build_only
- do_run = not do_build_only and not options.cmake_only
- skip_slow = ti.test.slow and not options.enable_slow
-
- # FIXME: Need refactoring and cleanup
- type = None
- if ti.platform.qemu_support and do_run:
- type = "qemu"
- elif ti.test.type == "unit":
- type = "unit"
- elif ti.platform.type == "native" and do_run:
- type = "native"
- elif ti.platform.simulation == "nsim" and do_run:
- if find_executable("nsimdrv"):
- type = "nsim"
- elif ti.platform.simulation == "renode" and do_run:
- if find_executable("renode"):
- type = "renode"
- elif options.device_testing and (not ti.build_only) and (not options.build_only):
- type = "device"
-
- if not skip_slow:
- self.add_goal(ti, type, args)
- else:
- verbose("Skipping slow test: " + ti.name)
-
- def _remove_unkept_files(self, base_path, keep_paths):
- for dirpath, dirnames, filenames in os.walk(base_path, topdown=False):
- for name in filenames:
- path = os.path.join(dirpath, name)
- if path not in keep_paths:
- os.remove(path)
- # Remove empty directories and symbolic links to directories
- for dir in dirnames:
- path = os.path.join(dirpath, dir)
- if os.path.islink(path):
- os.remove(path)
- elif not os.listdir(path):
- os.rmdir(path)
-
-
- def execute(self, callback_fn=None, context=None, keep_files=None):
- """Execute all the registered build goals
-
- @param callback_fn If not None, a callback function will be called
- as individual goals transition between states. This function
- should accept two parameters: a string state and an arbitrary
- context object, supplied here
- @param context Context object to pass to the callback function.
- Type and semantics are specific to that callback function.
- @param keep_files If not None, we will remove all files except those
- defined in keep_files. keep_files is a dictionary with the key
- being the 'goal' and the value is a pair. The pair first element
- is the dir path of the test and the second is a list of file paths
- that we should keep.
- @return A dictionary mapping goal names to final status.
- """
-
- with open(self.makefile, "wt") as tf, \
- open(os.devnull, "wb") as devnull, \
- open(self.logfile, "wt") as make_log:
- # Create our dynamic Makefile and execute it.
- # Watch stderr output which is where we will keep
- # track of build state
- tf.write('\n# Generated by %s which is expected\n' % __file__)
- tf.write('# to create QEMU_PIPE, spawn zephyr.exe, etc. \n\n')
- for name, goal in sorted(self.goals.items()):
- tf.write(goal.text)
- tf.write("all: %s\n" % (" ".join(sorted(self.goals.keys()))))
- tf.flush()
-
- cmd = ["make", "-k", "-j", str(JOBS), "-f", tf.name, "all"]
-
- # assure language neutral environment
- make_env = os.environ.copy()
- make_env['LC_MESSAGES'] = 'C.UTF-8'
- verbose("In %s, spawning: " % os.getcwd()
- + " ".join(shlex.quote(word)for word in cmd))
- p = subprocess.Popen(cmd, stderr=subprocess.PIPE,
- stdout=devnull, env=make_env)
-
- for line in iter(p.stderr.readline, b''):
- line = line.decode("utf-8")
- make_log.write(line)
- verbose("MAKE: " + repr(line.strip()))
- m = MakeGenerator.re_make.match(line)
- if not m:
- continue
-
- state, name, _, error = m.groups()
- if error:
- goal = self.goals[error]
- # Sometimes QEMU will run an image and then crash out, which
- # will cause the 'make run' invocation to exit with
- # nonzero status.
- # Need to distinguish this case from a compilation failure.
- if goal.make_state == "building":
- goal.fail("build_error")
- elif goal.handler:
- goal.fail("handler_crash")
- else:
- goal.fail("unknown_error")
-
- else:
- goal = self.goals[name]
- goal.make_state = state
-
- if state == "finished":
- if goal.handler and not options.cmake_only:
- if hasattr(goal.handler, "handle"):
- goal.handler.handle()
- goal.handler_log = goal.handler.log
-
- thread_status, metrics = goal.handler.get_state()
- goal.metrics.update(metrics)
- if thread_status == "passed":
- goal.success()
- else:
- goal.fail(thread_status)
- else:
- goal.success()
-
- if keep_files is not None:
- tc_path, keep_paths = keep_files[name]
- self._remove_unkept_files(tc_path, keep_paths)
-
- if callback_fn:
- callback_fn(context, self.goals, goal)
-
- p.wait()
- return self.goals
-
-
# "list" - List of strings
# "list:<type>" - List of <type>
# "set" - Set of unordered, unique strings
@@ -1491,7 +1174,7 @@
# XXX Be sure to update __doc__ if you change any of this!!
-platform_valid_keys = {"qemu_support": {"type": "bool", "default": False},
+platform_valid_keys = {
"supported_toolchains": {"type": "list", "default": []},
"env": {"type": "list", "default": []}
}
@@ -1520,7 +1203,6 @@
"harness_config": {"type": "map", "default": {}}
}
-
class SanityConfigParser:
"""Class to read test case files with semantic checking
"""
@@ -1530,15 +1212,21 @@
@param filename Source .yaml file to read
"""
- self.data = scl.yaml_load_verify(filename, schema)
+ self.data = {}
+ self.schema = schema
self.filename = filename
self.tests = {}
self.common = {}
+
+ def load(self):
+ self.data = scl.yaml_load_verify(self.filename, self.schema)
+
if 'tests' in self.data:
self.tests = self.data['tests']
if 'common' in self.data:
self.common = self.data['common']
+
def _cast_value(self, value, typestr):
if isinstance(value, str):
v = value.strip()
@@ -1566,7 +1254,7 @@
elif typestr.startswith("set"):
vs = v.split()
if len(typestr) > 3 and typestr[3] == ":":
- return set([self._cast_value(vsi, typestr[4:]) for vsi in vs])
+ return {self._cast_value(vsi, typestr[4:]) for vsi in vs}
else:
return set(vs)
@@ -1646,7 +1334,7 @@
else:
try:
d[k] = self._cast_value(d[k], kinfo["type"])
- except ValueError as ve:
+ except ValueError:
raise ConfigurationError(
self.filename, "bad %s value '%s' for key '%s' in name '%s'" %
(kinfo["type"], d[k], k, name))
@@ -1659,20 +1347,36 @@
Maps directly to BOARD when building"""
- yaml_platform_schema = scl.yaml_load(
- os.path.join(
- ZEPHYR_BASE,
- "scripts",
- "sanity_chk",
- "sanitycheck-platform-schema.yaml"))
+ platform_schema = scl.yaml_load(os.path.join(ZEPHYR_BASE,
+ "scripts","sanity_chk","platform-schema.yaml"))
- def __init__(self, cfile):
+ def __init__(self):
"""Constructor.
- @param cfile Path to platform configuration file, which gives
- info about the platform to be added.
"""
- scp = SanityConfigParser(cfile, self.yaml_platform_schema)
+
+ self.name = ""
+ self.sanitycheck = True
+ # if no RAM size is specified by the board, take a default of 128K
+ self.ram = 128
+
+ self.ignore_tags = []
+ self.default = False
+ # if no flash size is specified by the board, take a default of 512K
+ self.flash = 512
+ self.supported = set()
+
+ self.arch = ""
+ self.type = "na"
+ self.simulation = "na"
+ self.supported_toolchains = []
+ self.env = []
+ self.env_satisfied = True
+ self.filter_data = dict()
+
+ def load(self, platform_file):
+ scp = SanityConfigParser(platform_file, self.platform_schema)
+ scp.load()
data = scp.data
self.name = data['identifier']
@@ -1689,7 +1393,6 @@
for item in supp_feature.split(":"):
self.supported.add(item)
- self.qemu_support = True if data.get('simulation', "na") == 'qemu' else False
self.arch = data['arch']
self.type = data.get('type', "na")
self.simulation = data.get('simulation', "na")
@@ -1697,38 +1400,19 @@
self.env = data.get("env", [])
self.env_satisfied = True
for env in self.env:
- if os.environ.get(env, None) == None:
+ if not os.environ.get(env, None):
self.env_satisfied = False
- self.defconfig = None
- pass
+
def __repr__(self):
return "<%s on %s>" % (self.name, self.arch)
-
-class Architecture:
- """Class representing metadata for a particular architecture
- """
-
- def __init__(self, name, platforms):
- """Architecture constructor
-
- @param name String name for this architecture
- @param platforms list of platforms belonging to this architecture
- """
- self.platforms = platforms
-
- self.name = name
-
- def __repr__(self):
- return "<arch %s>" % self.name
-
-
-class TestCase:
+class TestCase(object):
"""Class representing a test application
"""
- def __init__(self, testcase_root, workdir, name, tc_dict, yamlfile):
+
+ def __init__(self):
"""TestCase constructor.
This gets called by TestSuite as it finds and reads test yaml files.
@@ -1750,42 +1434,39 @@
@param tc_dict Dictionary with test values for this test case
from the testcase.yaml file
"""
- self.test_path = os.path.join(testcase_root, workdir)
- self.id = name
+ self.id = ""
+ self.source_dir = ""
+ self.yamlfile = ""
self.cases = []
- self.type = tc_dict["type"]
- self.tags = tc_dict["tags"]
- self.extra_args = tc_dict["extra_args"]
- self.extra_configs = tc_dict["extra_configs"]
- self.arch_whitelist = tc_dict["arch_whitelist"]
- self.arch_exclude = tc_dict["arch_exclude"]
- self.skip = tc_dict["skip"]
- self.platform_exclude = tc_dict["platform_exclude"]
- self.platform_whitelist = tc_dict["platform_whitelist"]
- self.toolchain_exclude = tc_dict["toolchain_exclude"]
- self.toolchain_whitelist = tc_dict["toolchain_whitelist"]
- self.tc_filter = tc_dict["filter"]
- self.timeout = tc_dict["timeout"]
- self.harness = tc_dict["harness"]
- self.harness_config = tc_dict["harness_config"]
- self.build_only = tc_dict["build_only"]
- self.build_on_all = tc_dict["build_on_all"]
- self.slow = tc_dict["slow"]
- self.min_ram = tc_dict["min_ram"]
- self.depends_on = tc_dict["depends_on"]
- self.min_flash = tc_dict["min_flash"]
- self.extra_sections = tc_dict["extra_sections"]
+ self.name = ""
- self.name = self.get_unique(testcase_root, workdir, name)
-
- self.defconfig = {}
- self.dt_config = {}
- self.cmake_cache = {}
- self.yamlfile = yamlfile
+ self.type = None
+ self.tags = None
+ self.extra_args = None
+ self.extra_configs = None
+ self.arch_whitelist = None
+ self.arch_exclude = None
+ self.skip = None
+ self.platform_exclude = None
+ self.platform_whitelist = None
+ self.toolchain_exclude = None
+ self.toolchain_whitelist = None
+ self.tc_filter = None
+ self.timeout = 60
+ self.harness = ""
+ self.harness_config = {}
+ self.build_only = True
+ self.build_on_all = False
+ self.slow = False
+ self.min_ram = None
+ self.depends_on = None
+ self.min_flash = None
+ self.extra_sections = None
- def get_unique(self, testcase_root, workdir, name):
+ @staticmethod
+ def get_unique(testcase_root, workdir, name):
canonical_testcase_root = os.path.realpath(testcase_root)
if Path(canonical_zephyr_base) in Path(canonical_testcase_root).parents:
@@ -1800,7 +1481,8 @@
unique = os.path.normpath(os.path.join(relative_tc_root, workdir, name))
return unique
- def scan_file(self, inf_name):
+ @staticmethod
+ def scan_file(inf_name):
suite_regex = re.compile(
# do not match until end-of-line, otherwise we won't allow
# stc_regex below to catch the ones that are declared in the same
@@ -1849,9 +1531,7 @@
main_c[suite_regex_match.end():suite_run_match.start()])
if achtung_matches:
warnings = "found invalid %s in ztest_test_suite()" \
- % ", ".join(set([
- match.decode() for match in achtung_matches
- ]))
+ % ", ".join({match.decode() for match in achtung_matches})
_matches = re.findall(
stc_regex,
main_c[suite_regex_match.end():suite_run_match.start()])
@@ -1871,9 +1551,8 @@
error("%s: can't find: %s" % (filename, e))
return subcases
-
- def parse_subcases(self):
- results = self.scan_path(self.test_path)
+ def parse_subcases(self, test_path):
+ results = self.scan_path(os.path.dirname(test_path))
for sub in results:
name = "{}.{}".format(self.id, sub)
self.cases.append(name)
@@ -1895,44 +1574,91 @@
out directory used is <outdir>/<platform>/<test case name>
"""
- def __init__(self, test, platform, base_outdir):
- self.test = test
- self.platform = platform
- self.name = os.path.join(platform.name, test.name)
- self.outdir = os.path.join(base_outdir, platform.name, test.name)
+ def __init__(self, testcase, platform, base_outdir):
- self.build_only = options.build_only or test.build_only \
- or self.check_dependency() or options.cmake_only
+ self.testcase = testcase
+ self.platform = platform
+
+ self.status = None
+ self.reason = "N/A"
+ self.metrics = dict()
+ self.handler = None
+
+
+ self.name = os.path.join(platform.name, testcase.name)
+ self.build_dir = os.path.join(base_outdir, platform.name, testcase.name)
+
+ self.build_only = self.check_build_or_run()
+ self.run = not self.build_only
+
self.results = {}
def __lt__(self, other):
return self.name < other.name
- def check_dependency(self):
- build_only = False
- if self.test.harness == 'console':
- if "fixture" in self.test.harness_config:
- fixture = self.test.harness_config['fixture']
- if fixture not in options.fixture:
- build_only = True
- elif self.test.harness:
- build_only = True
+ def check_build_or_run(self):
- return build_only
+ build_only = True
+
+ # we asked for build-only on the command line
+ if options.build_only:
+ return True
+
+ # The testcase is designed to be build only.
+ if self.testcase.build_only:
+ return True
+
+ # Do not run slow tests:
+ skip_slow = self.testcase.slow and not options.enable_slow
+ if skip_slow:
+ return True
+
+ runnable =bool(self.testcase.type == "unit" or \
+ self.platform.type == "native" or \
+ self.platform.simulation in ["nsim", "renode", "qemu"] or \
+ options.device_testing)
+
+ if self.platform.simulation == "nsim":
+ if not find_executable("nsimdrv"):
+ runnable = False
+
+ if self.platform.simulation == "renode":
+ if not find_executable("renode"):
+ runnable = False
+
+ # console harness allows us to run the test and capture data.
+ if self.testcase.harness == 'console':
+
+ # if we have a fixture that is also being supplied on the
+ # command-line, then we need to run the test, not just build it.
+ if "fixture" in self.testcase.harness_config:
+ fixture = self.testcase.harness_config['fixture']
+ if fixture in options.fixture:
+ build_only = False
+ else:
+ build_only = True
+ else:
+ build_only = False
+ elif self.testcase.harness:
+ build_only = True
+ else:
+ build_only = False
+
+ return not (not build_only and runnable)
def create_overlay(self, platform):
# Create this in a "sanitycheck/" subdirectory otherwise this
# will pass this overlay to kconfig.py *twice* and kconfig.cmake
# will silently give that second time precedence over any
# --extra-args=CONFIG_*
- subdir = os.path.join(self.outdir, "sanitycheck")
+ subdir = os.path.join(self.build_dir, "sanitycheck")
os.makedirs(subdir, exist_ok=True)
file = os.path.join(subdir, "testcase_extra.conf")
with open(file, "w") as f:
content = ""
- if len(self.test.extra_configs) > 0:
- content = "\n".join(self.test.extra_configs)
+ if self.testcase.extra_configs:
+ content = "\n".join(self.testcase.extra_configs)
if options.enable_coverage:
if platform in options.coverage_platform:
@@ -1948,593 +1674,545 @@
@return A SizeCalculator object
"""
- fns = glob.glob(os.path.join(self.outdir, "zephyr", "*.elf"))
- fns.extend(glob.glob(os.path.join(self.outdir, "zephyr", "*.exe")))
+ fns = glob.glob(os.path.join(self.build_dir, "zephyr", "*.elf"))
+ fns.extend(glob.glob(os.path.join(self.build_dir, "zephyr", "*.exe")))
fns = [x for x in fns if not x.endswith('_prebuilt.elf')]
- if (len(fns) != 1):
+ if len(fns) != 1:
raise BuildError("Missing/multiple output ELF binary")
- return SizeCalculator(fns[0], self.test.extra_sections)
+
+ return SizeCalculator(fns[0], self.testcase.extra_sections)
def __repr__(self):
- return "<TestCase %s on %s>" % (self.test.name, self.platform.name)
+ return "<TestCase %s on %s>" % (self.testcase.name, self.platform.name)
-def defconfig_cb(context, goals, goal):
- if not goal.failed:
- return
+class CMake():
- info("%sCould not build defconfig for %s%s" %
- (COLOR_RED, goal.name, COLOR_NORMAL))
- if INLINE_LOGS:
- with open(goal.get_error_log()) as fp:
- data = fp.read()
- sys.stdout.write(data)
- if log_file:
- log_file.write(data)
- else:
- info("\tsee: " + COLOR_YELLOW + goal.get_error_log() + COLOR_NORMAL)
+ config_re = re.compile('(CONFIG_[A-Za-z0-9_]+)[=]\"?([^\"]*)\"?$')
+ dt_re = re.compile('([A-Za-z0-9_]+)[=]\"?([^\"]*)\"?$')
+
+ def __init__(self, testcase, platform, source_dir, build_dir):
+
+ self.cwd = None
+ self.capture_output = True
+
+ self.defconfig = {}
+ self.cmake_cache = {}
+ self.devicetree = {}
+
+ self.instance = None
+ self.testcase = testcase
+ self.platform = platform
+ self.source_dir = source_dir
+ self.build_dir = build_dir
+ self.log = "build.log"
+
+ def parse_generated(self):
+ self.defconfig = {}
+ return {}
+
+ def run_build(self, args=[]):
+
+ verbose("Building %s for %s" % (self.source_dir, self.platform.name))
+
+ cmake_args = []
+ cmake_args.extend(args)
+ cmake = shutil.which('cmake')
+ cmd = [cmake] + cmake_args
+ kwargs = dict()
+
+ if self.capture_output:
+ kwargs['stdout'] = subprocess.PIPE
+ # CMake sends the output of message() to stderr unless it's STATUS
+ kwargs['stderr'] = subprocess.STDOUT
+
+ if self.cwd:
+ kwargs['cwd'] = self.cwd
+
+ p = subprocess.Popen(cmd, **kwargs)
+ out, _ = p.communicate()
+
+ results = {}
+ if p.returncode == 0:
+ msg = "Finished building %s for %s" %(self.source_dir, self.platform.name)
+
+ self.instance.status = "passed"
+ self.instance.reason = ""
+ results = {'msg': msg, "returncode": p.returncode, "instance": self.instance}
+
+ if out:
+ log_msg = out.decode(sys.getdefaultencoding())
+ with open(os.path.join(self.build_dir, self.log), "a") as log:
+ log.write(log_msg)
+
+ else:
+ return None
+ else:
+ # A real error occurred, raise an exception
+ if out:
+ log_msg = out.decode(sys.getdefaultencoding())
+ with open(os.path.join(self.build_dir, self.log), "a") as log:
+ log.write(log_msg)
+
+ overflow_flash = "region `FLASH' overflowed by"
+ overflow_ram = "region `RAM' overflowed by"
+
+ if log_msg:
+ if log_msg.find(overflow_flash) > 0 or log_msg.find(overflow_ram) > 0:
+ verbose("RAM/ROM Overflow")
+ self.instance.status = "skipped"
+ self.instance.reason = "overflow"
+ else:
+ self.instance.status = "failed"
+ self.instance.reason = "Build failure"
+
+ results = {
+ "returncode": p.returncode,
+ "instance": self.instance,
+ }
+
+ return results
+
+ def run_cmake(self, args=[]):
+
+ verbose("Running cmake on %s for %s" %(self.source_dir, self.platform.name))
+
+ cmake_args = ['-B{}'.format(self.build_dir),
+ '-S{}'.format(self.source_dir),
+ '-G{}'.format(get_generator()[1])]
+
+ args = ["-D{}".format(a.replace('"', '')) for a in args]
+ cmake_args.extend(args)
+
+ cmake_opts = ['-DBOARD={}'.format(self.platform.name)]
+ cmake_args.extend(cmake_opts)
+
+ cmake = shutil.which('cmake')
+ cmd = [cmake] + cmake_args
+ kwargs = dict()
+
+ if self.capture_output:
+ kwargs['stdout'] = subprocess.PIPE
+ # CMake sends the output of message() to stderr unless it's STATUS
+ kwargs['stderr'] = subprocess.STDOUT
+
+ if self.cwd:
+ kwargs['cwd'] = self.cwd
+
+ p = subprocess.Popen(cmd, **kwargs)
+ out, _ = p.communicate()
+
+ if p.returncode == 0:
+ filter_results = self.parse_generated()
+ msg = "Finished building %s for %s" %(self.source_dir, self.platform.name)
+
+ results = {'msg': msg, 'filter': filter_results}
+
+ else:
+ self.instance.status = "failed"
+ self.instance.reason = "Cmake build failure"
+ results = {"returncode": p.returncode}
+
+
+ if out:
+ with open(os.path.join(self.build_dir, self.log), "a") as log:
+ log_msg = out.decode(sys.getdefaultencoding())
+ log.write(log_msg)
+
+ return results
+
+
+class FilterBuilder(CMake):
+
+ def __init__(self, testcase, platform, source_dir, build_dir):
+ super().__init__(testcase, platform, source_dir, build_dir)
+
+ self.log = "config-sanitycheck.log"
+
+ def parse_generated(self):
+
+ if self.platform.name == "unit_testing":
+ return {}
+
+ _generated_dt_confg = "include/generated/generated_dts_board.conf"
+
+ cmake_cache_path = os.path.join(self.build_dir, "CMakeCache.txt")
+ dt_config_path = os.path.join(self.build_dir, "zephyr", _generated_dt_confg)
+ defconfig_path = os.path.join(self.build_dir, "zephyr", ".config")
+
+ with open(defconfig_path, "r") as fp:
+ defconfig = {}
+ for line in fp.readlines():
+ m = self.config_re.match(line)
+ if not m:
+ if line.strip() and not line.startswith("#"):
+ sys.stderr.write("Unrecognized line %s\n" % line)
+ continue
+ defconfig[m.group(1)] = m.group(2).strip()
+
+ self.defconfig = defconfig
+
+ cmake_conf = {}
+ try:
+ cache = CMakeCache.from_file(cmake_cache_path)
+ except FileNotFoundError:
+ cache = {}
+
+ for k in iter(cache):
+ cmake_conf[k.name] = k.value
+
+ self.cmake_cache = cmake_conf
+
+ dt_conf = {}
+ if os.path.exists(dt_config_path):
+ with open(dt_config_path, "r") as fp:
+ for line in fp.readlines():
+ m = self.dt_re.match(line)
+ if not m:
+ if line.strip() and not line.startswith("#"):
+ sys.stderr.write("Unrecognized line %s\n" % line)
+ continue
+ dt_conf[m.group(1)] = m.group(2).strip()
+ self.devicetree = dt_conf
+
+ filter_data = {
+ "ARCH": self.platform.arch,
+ "PLATFORM": self.platform.name
+ }
+ filter_data.update(os.environ)
+ filter_data.update(self.defconfig)
+ filter_data.update(self.cmake_cache)
+ filter_data.update(self.devicetree)
+
+ if self.testcase and self.testcase.tc_filter:
+ try:
+ res = expr_parser.parse(self.testcase.tc_filter, filter_data)
+ except (ValueError, SyntaxError) as se:
+ sys.stderr.write(
+ "Failed processing %s\n" % self.testcase.yamlfile)
+ raise se
+
+ if not res:
+ return {os.path.join(self.platform.name, self.testcase.name): True}
+ else:
+ return {os.path.join(self.platform.name, self.testcase.name): False}
+ else:
+ self.platform.filter_data = filter_data
+ return filter_data
+
+
+class ProjectBuilder(FilterBuilder):
+
+ def __init__(self, suite, instance):
+ super().__init__(instance.testcase, instance.platform, instance.testcase.source_dir, instance.build_dir)
+
+ self.log = "build.log"
+ self.instance = instance
+ self.suite = suite
+
+ def setup_handler(self):
+
+ instance = self.instance
+ args = []
+
+ # FIXME: Needs simplification
+ if instance.platform.simulation == "qemu":
+ instance.handler = QEMUHandler(instance, "qemu")
+ args.append("QEMU_PIPE=%s" % instance.handler.get_fifo())
+ instance.handler.call_make_run = True
+ elif instance.testcase.type == "unit":
+ instance.handler = BinaryHandler(instance, "unit")
+ instance.handler.binary = os.path.join(instance.build_dir, "testbinary")
+ elif instance.platform.type == "native":
+ instance.handler = BinaryHandler(instance, "native")
+ instance.handler.binary = os.path.join(instance.build_dir, "zephyr", "zephyr.exe")
+ elif instance.platform.simulation == "nsim":
+ if find_executable("nsimdrv"):
+ instance.handler = BinaryHandler(instance, "nsim")
+ instance.handler.call_make_run = True
+ elif instance.platform.simulation == "renode":
+ if find_executable("renode"):
+ instance.handler = BinaryHandler(instance, "renode")
+ instance.handler.pid_fn = os.path.join(instance.build_dir, "renode.pid")
+ instance.handler.call_make_run = True
+ elif options.device_testing:
+ instance.handler = DeviceHandler(instance, "device")
+
+ if instance.handler:
+ instance.handler.args = args
+
+ def process(self, message):
+ op = message.get('op')
+
+ if not self.instance.handler:
+ self.setup_handler()
+
+ # The build process, call cmake and build with configured generator
+ if op == "cmake":
+ results = self.cmake()
+ if self.instance.status == "failed":
+ pipeline.put({"op": "report", "test": self.instance})
+ elif options.cmake_only:
+ pipeline.put({"op": "report", "test": self.instance})
+ else:
+ if self.instance.name in results['filter'] and results['filter'][self.instance.name]:
+ verbose("filtering %s" % self.instance.name)
+ self.instance.status = "skipped"
+ self.instance.reason = "filter"
+ pipeline.put({"op": "report", "test": self.instance})
+ else:
+ pipeline.put({"op": "build", "test": self.instance})
+
+
+ elif op == "build":
+ verbose("build test: %s" %self.instance.name)
+ results = self.build()
+
+ if results.get('returncode', 1) > 0:
+ pipeline.put({"op": "report", "test": self.instance})
+ else:
+ if self.instance.run:
+ pipeline.put({"op": "run", "test": self.instance})
+ else:
+ pipeline.put({"op": "report", "test": self.instance})
+ # Run the generated binary using one of the supported handlers
+ elif op == "run":
+ verbose("run test: %s" %self.instance.name)
+ self.run()
+ self.instance.status, _ = self.instance.handler.get_state()
+ self.instance.reason = ""
+ pipeline.put({
+ "op": "report",
+ "test": self.instance,
+ "state": "executed",
+ "status": self.instance.status,
+ "reason": self.instance.status}
+ )
+
+ # Report results and output progress to screen
+ elif op == "report":
+ self.report_out()
+
+ def report_out(self):
+ total_tests_width = len(str(self.suite.total_tests))
+ self.suite.total_done += 1
+ instance = self.instance
+
+ if instance.status in ["failed", "timeout"]:
+ self.suite.total_failed += 1
+ if VERBOSE or not TERMINAL:
+ status = COLOR_RED + "FAILED " + COLOR_NORMAL + instance.reason
+ else:
+ info(
+ "{:<25} {:<50} {}FAILED{}: {}".format(
+ instance.platform.name,
+ instance.testcase.name,
+ COLOR_RED,
+ COLOR_NORMAL,
+ instance.reason), False)
+
+ # FIXME
+ h_log = "{}/handler.log".format(instance.build_dir)
+ b_log = "{}/build.log".format(instance.build_dir)
+ if os.path.exists(h_log):
+ log_info("{}".format(h_log))
+ else:
+ log_info("{}".format(b_log))
+
+ elif instance.status == "skipped":
+ self.suite.total_skipped += 1
+ status = COLOR_YELLOW + "SKIPPED" + COLOR_NORMAL
+
+ else:
+ status = COLOR_GREEN + "PASSED" + COLOR_NORMAL
+
+ if VERBOSE or not TERMINAL:
+ if options.cmake_only:
+ more_info = "cmake"
+ elif instance.status == "skipped":
+ more_info = instance.reason
+ else:
+ if instance.handler and instance.run:
+ more_info = instance.handler.type_str
+ htime = instance.handler.duration
+ if htime:
+ more_info += " {:.3f}s".format(htime)
+ else:
+ more_info = "build"
+
+ info("{:>{}}/{} {:<25} {:<50} {} ({})".format(
+ self.suite.total_done, total_tests_width, self.suite.total_tests, instance.platform.name,
+ instance.testcase.name, status, more_info))
+
+ if instance.status in ["failed", "timeout"]:
+ h_log = "{}/handler.log".format(instance.build_dir)
+ b_log = "{}/build.log".format(instance.build_dir)
+ if os.path.exists(h_log):
+ log_info("{}".format(h_log))
+ else:
+ log_info("{}".format(b_log))
+
+ else:
+ sys.stdout.write("\rtotal complete: %s%4d/%4d%s %2d%% skipped: %s%4d%s, failed: %s%4d%s" % (
+ COLOR_GREEN,
+ self.suite.total_done,
+ self.suite.total_tests,
+ COLOR_NORMAL,
+ int((float(self.suite.total_done) / self.suite.total_tests) * 100),
+ COLOR_YELLOW if self.suite.total_skipped > 0 else COLOR_NORMAL,
+ self.suite.total_skipped,
+ COLOR_NORMAL,
+ COLOR_RED if self.suite.total_failed > 0 else COLOR_NORMAL,
+ self.suite.total_failed,
+ COLOR_NORMAL
+ )
+ )
+ sys.stdout.flush()
+
+ def cmake(self):
+
+ instance = self.instance
+ args = self.testcase.extra_args[:]
+
+ if options.extra_args:
+ args += options.extra_args
+
+ if instance.handler:
+ args += instance.handler.args
+
+ # merge overlay files into one variable
+ overlays = ""
+ idx = 0
+ for arg in args:
+ match = re.search('OVERLAY_CONFIG="(.*)"', arg)
+ if match:
+ overlays += match.group(1)
+ del args[idx]
+ idx += 1
+
+ if self.testcase.extra_configs or options.coverage:
+ args.append("OVERLAY_CONFIG=\"%s %s\"" %(overlays,
+ os.path.join(instance.build_dir,
+ "sanitycheck", "testcase_extra.conf")))
+
+ results = self.run_cmake(args)
+ return results
+
+ def build(self):
+ results = self.run_build(['--build', self.build_dir])
+ return results
+
+ def run(self):
+
+ instance = self.instance
+
+ if instance.handler.type_str == "device":
+ instance.handler.suite = self.suite
+
+ instance.handler.handle()
+
+ if instance.handler.type_str == "qemu":
+ verbose("Running %s (%s)" %(instance.name, instance.handler.type_str))
+ command = [get_generator()[0]]
+ command += ["-C", self.build_dir, "run"]
+
+ with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self.build_dir) as proc:
+ verbose("Spawning QEMUHandler Thread for %s" % instance.name)
+ proc.wait()
+ self.returncode = proc.returncode
+
+ sys.stdout.flush()
+
+
+pipeline = queue.LifoQueue()
+
+class BoundedExecutor(concurrent.futures.ThreadPoolExecutor):
+ """BoundedExecutor behaves as a ThreadPoolExecutor which will block on
+ calls to submit() once the limit given as "bound" work items are queued for
+ execution.
+ :param bound: Integer - the maximum number of items in the work queue
+ :param max_workers: Integer - the size of the thread pool
+ """
+ def __init__(self, bound, max_workers, **kwargs):
+ super().__init__(max_workers)
+ #self.executor = ThreadPoolExecutor(max_workers=max_workers)
+ self.semaphore = BoundedSemaphore(bound + max_workers)
+
+ def submit(self, fn, *args, **kwargs):
+ self.semaphore.acquire()
+ try:
+ future = super().submit(fn, *args, **kwargs)
+ except:
+ self.semaphore.release()
+ raise
+ else:
+ future.add_done_callback(lambda x: self.semaphore.release())
+ return future
+
+ def shutdown(self, wait=True):
+ super().shutdown(wait)
class TestSuite:
config_re = re.compile('(CONFIG_[A-Za-z0-9_]+)[=]\"?([^\"]*)\"?$')
dt_re = re.compile('([A-Za-z0-9_]+)[=]\"?([^\"]*)\"?$')
- yaml_tc_schema = scl.yaml_load(
+ tc_schema = scl.yaml_load(
os.path.join(ZEPHYR_BASE,
- "scripts", "sanity_chk", "sanitycheck-tc-schema.yaml"))
+ "scripts", "sanity_chk", "testcase-schema.yaml"))
def __init__(self, board_root_list, testcase_roots, outdir):
+
+ self.roots = testcase_roots
+ if not isinstance(board_root_list, list):
+ self.board_roots= [board_root_list]
+ else:
+ self.board_roots = board_root_list
+
# Keep track of which test cases we've filtered out and why
- self.arches = {}
self.testcases = {}
self.platforms = []
+ self.default_platforms = []
self.outdir = os.path.abspath(outdir)
- self.instances = {}
- self.goals = None
self.discards = None
self.load_errors = 0
+ self.instances = dict()
- for testcase_root in testcase_roots:
- testcase_root = os.path.abspath(testcase_root)
+ self.total_tests = 0 # number of test instances
+ self.total_cases = 0 # number of test cases
+ self.total_done = 0 # tests completed
+ self.total_failed = 0
+ self.total_skipped = 0
- debug("Reading test case configuration files under %s..." %
- testcase_root)
- for dirpath, dirnames, filenames in os.walk(testcase_root,
- topdown=True):
- verbose("scanning %s" % dirpath)
- if 'sample.yaml' in filenames:
- filename = 'sample.yaml'
- elif 'testcase.yaml' in filenames:
- filename = 'testcase.yaml'
- else:
- continue
+ self.total_platforms = 0
+ self.start_time = 0
+ self.duration = 0
+ self.warnings = 0
+ self.cv = threading.Condition()
- verbose("Found possible test case in " + dirpath)
- dirnames[:] = []
- yaml_path = os.path.join(dirpath, filename)
- try:
- parsed_data = SanityConfigParser(
- yaml_path, self.yaml_tc_schema)
+ # hardcoded for now
+ self.connected_hardware = []
- workdir = os.path.relpath(dirpath, testcase_root)
- for name in parsed_data.tests.keys():
- tc_dict = parsed_data.get_test(name, testcase_valid_keys)
- tc = TestCase(testcase_root, workdir, name, tc_dict,
- yaml_path)
- tc.parse_subcases()
- self.testcases[tc.name] = tc
-
- except Exception as e:
- error("E: %s: can't load (skipping): %s" % (yaml_path, e))
- self.load_errors += 1
-
-
- for board_root in board_root_list:
- board_root = os.path.abspath(board_root)
-
- debug(
- "Reading platform configuration files under %s..." %
- board_root)
- for fn in glob.glob(os.path.join(board_root, "*", "*", "*.yaml")):
- verbose("Found plaform configuration " + fn)
- try:
- platform = Platform(fn)
- if platform.sanitycheck:
- self.platforms.append(platform)
- except RuntimeError as e:
- error("E: %s: can't load: %s" % (fn, e))
- self.load_errors += 1
-
- arches = []
- for p in self.platforms:
- arches.append(p.arch)
- for a in list(set(arches)):
- aplatforms = [p for p in self.platforms if p.arch == a]
- arch = Architecture(a, aplatforms)
- self.arches[a] = arch
-
- self.instances = {}
-
- def get_platform(self, name):
- selected_platform = None
- for platform in self.platforms:
- if platform.name == name:
- selected_platform = platform
- break
- return selected_platform
-
- def get_last_failed(self):
- try:
- if not os.path.exists(LAST_SANITY):
- raise SanityRuntimeError("Couldn't find last sanity run.")
- except Exception as e:
- print(str(e))
- sys.exit(2)
-
- result = []
- with open(LAST_SANITY, "r") as fp:
- cr = csv.DictReader(fp)
- instance_list = []
- for row in cr:
- if row["passed"] == "True":
- continue
- test = row["test"]
- platform = self.get_platform(row["platform"])
- instance = TestInstance(self.testcases[test], platform, self.outdir)
- instance.create_overlay(platform.name)
- instance_list.append(instance)
- self.add_instances(instance_list)
-
- def load_from_file(self, file):
- try:
- if not os.path.exists(file):
- raise SanityRuntimeError(
- "Couldn't find input file with list of tests.")
- except Exception as e:
- print(str(e))
- sys.exit(2)
-
- with open(file, "r") as fp:
- cr = csv.reader(fp)
- instance_list = []
- for row in cr:
- name = os.path.join(row[0], row[1])
- platform = self.get_platform(row[2])
- instance = TestInstance(self.testcases[name], platform, self.outdir)
- instance.create_overlay(platform.name)
- instance_list.append(instance)
- self.add_instances(instance_list)
-
- def apply_filters(self):
-
- toolchain = os.environ.get("ZEPHYR_TOOLCHAIN_VARIANT", None) or \
- os.environ.get("ZEPHYR_GCC_VARIANT", None)
-
- if toolchain == "gccarmemb":
- # Remove this translation when gccarmemb is no longer supported.
- toolchain = "gnuarmemb"
-
- try:
- if not toolchain:
- raise SanityRuntimeError("E: Variable ZEPHYR_TOOLCHAIN_VARIANT is not defined")
- except Exception as e:
- print(str(e))
- sys.exit(2)
-
-
- instances = []
- discards = {}
- platform_filter = options.platform
- testcase_filter = run_individual_tests
- arch_filter = options.arch
- tag_filter = options.tag
- exclude_tag = options.exclude_tag
- config_filter = options.config
- extra_args = options.extra_args
- all_plats = options.all
-
- verbose("platform filter: " + str(platform_filter))
- verbose(" arch_filter: " + str(arch_filter))
- verbose(" tag_filter: " + str(tag_filter))
- verbose(" exclude_tag: " + str(exclude_tag))
- verbose(" config_filter: " + str(config_filter))
-
- default_platforms = False
-
- if all_plats:
- info("Selecting all possible platforms per test case")
- # When --all used, any --platform arguments ignored
- platform_filter = []
- elif not platform_filter:
- info("Selecting default platforms per test case")
- default_platforms = True
-
- mg = MakeGenerator(self.outdir)
- defconfig_list = {}
- dt_list = {}
- cmake_list = {}
- tc_file_list = {}
- for tc_name, tc in self.testcases.items():
- for arch_name, arch in self.arches.items():
- for plat in arch.platforms:
- instance = TestInstance(tc, plat, self.outdir)
-
- if (arch_name == "unit") != (tc.type == "unit"):
- continue
-
- if tc.build_on_all and not platform_filter:
- platform_filter = []
-
- if tc.skip:
- continue
-
- if tag_filter and not tc.tags.intersection(tag_filter):
- continue
-
- if exclude_tag and tc.tags.intersection(exclude_tag):
- continue
-
- if testcase_filter and tc_name not in testcase_filter:
- continue
-
- if arch_filter and arch_name not in arch_filter:
- continue
-
- if tc.arch_whitelist and arch.name not in tc.arch_whitelist:
- continue
-
- if tc.arch_exclude and arch.name in tc.arch_exclude:
- continue
-
- if tc.platform_exclude and plat.name in tc.platform_exclude:
- continue
-
- if tc.toolchain_exclude and toolchain in tc.toolchain_exclude:
- continue
-
- if platform_filter and plat.name not in platform_filter:
- continue
-
- if plat.ram < tc.min_ram:
- continue
-
- if set(plat.ignore_tags) & tc.tags:
- continue
-
- if tc.depends_on:
- dep_intersection = tc.depends_on.intersection(
- set(plat.supported))
- if dep_intersection != set(tc.depends_on):
- continue
-
- if plat.flash < tc.min_flash:
- continue
-
- if tc.platform_whitelist and plat.name not in tc.platform_whitelist:
- continue
-
- if tc.toolchain_whitelist and toolchain not in tc.toolchain_whitelist:
- continue
-
- if (plat.env_satisfied and tc.tc_filter
- and (plat.default or all_plats or platform_filter)
- and (toolchain in plat.supported_toolchains or options.force_toolchain)):
- args = tc.extra_args[:]
- args.append("BOARD={}".format(plat.name))
- args.extend(extra_args)
- # FIXME would be nice to use a common outdir for this so that
- # conf, gen_idt, etc aren't rebuilt for every combination,
- # need a way to avoid different Make processes from clobbering
- # each other since they all try to build them
- # simultaneously
-
- o = os.path.join(self.outdir, plat.name, tc.name)
- cmake_cache_path = os.path.join(o, "CMakeCache.txt")
- generated_dt_confg = "include/generated/generated_dts_board.conf"
- dt_config_path = os.path.join(o, "zephyr", generated_dt_confg)
- defconfig_path = os.path.join(o, "zephyr", ".config")
- dt_list[tc, plat, tc.name.split("/")[-1]] = dt_config_path
- cmake_list[tc, plat, tc.name.split("/")[-1]] = cmake_cache_path
- defconfig_list[tc, plat, tc.name.split("/")[-1]] = defconfig_path
- goal = "_".join([plat.name, "_".join(tc.name.split("/")), "config-sanitycheck"])
- tc_file_list[goal] = (o, [dt_config_path, cmake_cache_path, defconfig_path])
- mg.add_build_goal(goal, os.path.join(ZEPHYR_BASE, tc.test_path),
- o, args, "config-sanitycheck.log", make_args="config-sanitycheck")
-
- info("Building testcase defconfigs...")
- kept_files = tc_file_list if options.save_tests else None
- results = mg.execute(defconfig_cb, keep_files=kept_files)
-
- info("Filtering test cases...")
- for name, goal in results.items():
- try:
- if goal.failed:
- raise SanityRuntimeError("Couldn't build some defconfigs")
- except Exception as e:
- error(str(e))
- sys.exit(2)
-
-
- for k, out_config in defconfig_list.items():
- test, plat, name = k
- defconfig = {}
- with open(out_config, "r") as fp:
- for line in fp.readlines():
- m = TestSuite.config_re.match(line)
- if not m:
- if line.strip() and not line.startswith("#"):
- sys.stderr.write("Unrecognized line %s\n" % line)
- continue
- defconfig[m.group(1)] = m.group(2).strip()
- test.defconfig[plat] = defconfig
-
- for k, cache_file in cmake_list.items():
- if not os.path.exists(out_config):
- continue
-
- test, plat, name = k
- cmake_conf = {}
- try:
- cache = CMakeCache.from_file(cache_file)
- except FileNotFoundError:
- cache = {}
-
- for k in iter(cache):
- cmake_conf[k.name] = k.value
-
- test.cmake_cache[plat] = cmake_conf
-
- for k, out_config in dt_list.items():
- if not os.path.exists(out_config):
- continue
-
- test, plat, name = k
- dt_conf = {}
- with open(out_config, "r") as fp:
- for line in fp.readlines():
- m = TestSuite.dt_re.match(line)
- if not m:
- if line.strip() and not line.startswith("#"):
- sys.stderr.write("Unrecognized line %s\n" % line)
- continue
- dt_conf[m.group(1)] = m.group(2).strip()
- test.dt_config[plat] = dt_conf
-
- for tc_name, tc in self.testcases.items():
- for arch_name, arch in self.arches.items():
- instance_list = []
- for plat in arch.platforms:
- instance = TestInstance(tc, plat, self.outdir)
-
- if (arch_name == "unit") != (tc.type == "unit"):
- # Discard silently
- continue
-
- if tc.skip:
- discards[instance] = "Skip filter"
- continue
-
- if tc.build_on_all and not platform_filter:
- platform_filter = []
-
- if tag_filter and not tc.tags.intersection(tag_filter):
- discards[instance] = "Command line testcase tag filter"
- continue
-
- if exclude_tag and tc.tags.intersection(exclude_tag):
- discards[instance] = "Command line testcase exclude filter"
- continue
-
- if testcase_filter and tc_name not in testcase_filter:
- discards[instance] = "Testcase name filter"
- continue
-
- if arch_filter and arch_name not in arch_filter:
- discards[instance] = "Command line testcase arch filter"
- continue
-
- if tc.arch_whitelist and arch.name not in tc.arch_whitelist:
- discards[instance] = "Not in test case arch whitelist"
- continue
-
- if tc.arch_exclude and arch.name in tc.arch_exclude:
- discards[instance] = "In test case arch exclude"
- continue
-
- if tc.platform_exclude and plat.name in tc.platform_exclude:
- discards[instance] = "In test case platform exclude"
- continue
-
- if tc.toolchain_exclude and toolchain in tc.toolchain_exclude:
- discards[instance] = "In test case toolchain exclude"
- continue
-
- if platform_filter and plat.name not in platform_filter:
- discards[instance] = "Command line platform filter"
- continue
-
- if tc.platform_whitelist and plat.name not in tc.platform_whitelist:
- discards[instance] = "Not in testcase platform whitelist"
- continue
-
- if tc.toolchain_whitelist and toolchain not in tc.toolchain_whitelist:
- discards[instance] = "Not in testcase toolchain whitelist"
- continue
-
- if not plat.env_satisfied:
- discards[instance] = "Environment ({}) not satisfied".format(", ".join(plat.env))
- continue
-
- if not options.force_toolchain \
- and toolchain and (toolchain not in plat.supported_toolchains) \
- and tc.type != 'unit':
- discards[instance] = "Not supported by the toolchain"
- continue
-
- if plat.ram < tc.min_ram:
- discards[instance] = "Not enough RAM"
- continue
-
- if tc.depends_on:
- dep_intersection = tc.depends_on.intersection(set(plat.supported))
- if dep_intersection != set(tc.depends_on):
- discards[instance] = "No hardware support"
- continue
-
- if plat.flash < tc.min_flash:
- discards[instance] = "Not enough FLASH"
- continue
-
- if set(plat.ignore_tags) & tc.tags:
- discards[instance] = "Excluded tags per platform"
- continue
-
- defconfig = {
- "ARCH": arch.name,
- "PLATFORM": plat.name
- }
- defconfig.update(os.environ)
- for p, tdefconfig in tc.defconfig.items():
- if p == plat:
- defconfig.update(tdefconfig)
- break
-
- for p, tdefconfig in tc.dt_config.items():
- if p == plat:
- defconfig.update(tdefconfig)
- break
-
- for p, tdefconfig in tc.cmake_cache.items():
- if p == plat:
- defconfig.update(tdefconfig)
- break
-
- if tc.tc_filter:
- try:
- res = expr_parser.parse(tc.tc_filter, defconfig)
- except (ValueError, SyntaxError) as se:
- sys.stderr.write(
- "Failed processing %s\n" % tc.yamlfile)
- raise se
- if not res:
- discards[instance] = (
- "defconfig doesn't satisfy expression '%s'" %
- tc.tc_filter)
- continue
-
- instance_list.append(instance)
-
- if not instance_list:
- # Every platform in this arch was rejected already
- continue
-
- if default_platforms and not tc.build_on_all:
- if not tc.platform_whitelist:
- instances = list(
- filter(
- lambda tc: tc.platform.default,
- instance_list))
- self.add_instances(instances)
- else:
- self.add_instances(instance_list[:1])
-
- for instance in list(
- filter(lambda tc: not tc.platform.default, instance_list)):
- discards[instance] = "Not a default test platform"
- else:
- self.add_instances(instance_list)
-
- for name, case in self.instances.items():
- case.create_overlay(case.platform.name)
-
- self.discards = discards
-
- return discards
-
- def add_instances(self, ti_list):
- for ti in ti_list:
- self.instances[ti.name] = ti
-
- def execute(self, cb, cb_context):
-
- def calc_one_elf_size(name, goal):
- if not goal.failed:
- if self.instances[name].platform.type != "native":
- i = self.instances[name]
- sc = i.calculate_sizes()
- goal.metrics["ram_size"] = sc.get_ram_size()
- goal.metrics["rom_size"] = sc.get_rom_size()
- goal.metrics["unrecognized"] = sc.unrecognized_sections()
- else:
- goal.metrics["ram_size"] = 0
- goal.metrics["rom_size"] = 0
- goal.metrics["unrecognized"] = []
-
- mg = MakeGenerator(self.outdir)
- for i in self.instances.values():
- mg.add_test_instance(i, options.extra_args)
- self.goals = mg.execute(cb, cb_context)
-
- if not options.disable_size_report and not options.cmake_only:
- # Parallelize size calculation
- executor = concurrent.futures.ThreadPoolExecutor(JOBS)
- futures = [executor.submit(calc_one_elf_size, name, goal)
- for name, goal in self.goals.items()]
- concurrent.futures.wait(futures)
+ if options.jobs:
+ self.jobs = options.jobs
+ elif options.build_only:
+ self.jobs = multiprocessing.cpu_count() * 2
else:
- for goal in self.goals.values():
- goal.metrics["ram_size"] = 0
- goal.metrics["rom_size"] = 0
- goal.metrics["handler_time"] = 0
- goal.metrics["unrecognized"] = []
+ self.jobs = multiprocessing.cpu_count()
- return self.goals
+ info("JOBS: %d" % self.jobs)
- def save_tests(self, filename):
- with open(filename, "at") as csvfile:
- fieldnames = ['path', 'test', 'platform', 'arch']
- cw = csv.DictWriter(csvfile, fieldnames, lineterminator=os.linesep)
- for instance in self.instances.values():
- rowdict = {
- "path": os.path.dirname(instance.test.name),
- "test": os.path.basename(instance.test.name),
- "platform": instance.platform.name,
- "arch": instance.platform.arch
- }
- cw.writerow(rowdict)
+ def update(self):
+ self.total_tests = len(self.instances)
+ self.total_cases = len(self.testcases)
- def discard_report(self, filename):
-
- try:
- if self.discards is None:
- raise SanityRuntimeError("apply_filters() hasn't been run!")
- except Exception as e:
- error(str(e))
- sys.exit(2)
-
- with open(filename, "wt") as csvfile:
- fieldnames = ["test", "arch", "platform", "reason"]
- cw = csv.DictWriter(csvfile, fieldnames, lineterminator=os.linesep)
- cw.writeheader()
- for instance, reason in sorted(self.discards.items()):
- rowdict = {"test": instance.test.name,
- "arch": instance.platform.arch,
- "platform": instance.platform.name,
- "reason": reason}
- cw.writerow(rowdict)
def compare_metrics(self, filename):
# name, datatype, lower results better
interesting_metrics = [("ram_size", int, True),
("rom_size", int, True)]
- try:
- if self.goals is None:
- raise SanityRuntimeError("execute() hasn't been run!")
- except Exception as e:
- print(str(e))
- sys.exit(2)
if not os.path.exists(filename):
info("Cannot compare metrics, %s not found" % filename)
@@ -2550,118 +2228,681 @@
d[m] = row[m]
saved_metrics[(row["test"], row["platform"])] = d
- for name, goal in self.goals.items():
- i = self.instances[name]
- mkey = (i.test.name, i.platform.name)
+ for instance in self.instances.values():
+ mkey = (instance.testcase.name, instance.platform.name)
if mkey not in saved_metrics:
continue
sm = saved_metrics[mkey]
for metric, mtype, lower_better in interesting_metrics:
- if metric not in goal.metrics:
+ if metric not in instance.metrics:
continue
if sm[metric] == "":
continue
- delta = goal.metrics[metric] - mtype(sm[metric])
+ delta = instance.metrics.get(metric, 0) - mtype(sm[metric])
if delta == 0:
continue
- results.append((i, metric, goal.metrics[metric], delta,
+ results.append((instance, metric, instance.metrics.get(metric, 0 ), delta,
lower_better))
return results
- def testcase_target_report(self, report_file):
+ def misc_reports(self, report, show_footprint, all_deltas,
+ footprint_threshold, last_metrics):
- run = "Sanitycheck"
- eleTestsuite = None
- append = options.only_failed
+ if not report:
+ return
- errors = 0
- passes = 0
- fails = 0
- duration = 0
- skips = 0
+ deltas = self.compare_metrics(report)
+ warnings = 0
+ if deltas and show_footprint:
+ for i, metric, value, delta, lower_better in deltas:
+ if not all_deltas and ((delta < 0 and lower_better) or
+ (delta > 0 and not lower_better)):
+ continue
- for identifier, ti in self.instances.items():
- for k in ti.results.keys():
- if ti.results[k] == 'PASS':
- passes += 1
- elif ti.results[k] == 'BLOCK':
- errors += 1
- elif ti.results[k] == 'SKIP':
- skips += 1
- else:
- fails += 1
+ percentage = (float(delta) / float(value - delta))
+ if not all_deltas and (percentage <
+ (footprint_threshold / 100.0)):
+ continue
- eleTestsuites = ET.Element('testsuites')
- eleTestsuite = ET.SubElement(eleTestsuites, 'testsuite',
- name=run, time="%d" % duration,
- tests="%d" % (errors + passes + fails),
- failures="%d" % fails,
- errors="%d" % errors, skipped="%d" %skips)
+ info("{:<25} {:<60} {}{}{}: {} {:<+4}, is now {:6} {:+.2%}".format(
+ i.platform.name, i.testcase.name, COLOR_YELLOW,
+ "INFO" if all_deltas else "WARNING", COLOR_NORMAL,
+ metric, delta, value, percentage))
+ warnings += 1
- handler_time = "0"
+ if warnings:
+ info("Deltas based on metrics from last %s" %
+ ("release" if not last_metrics else "run"))
- # print out test results
- for identifier, ti in self.instances.items():
- for k in ti.results.keys():
+ def summary(self, unrecognized_sections):
+ failed = 0
+ for instance in self.instances.values():
+ if instance.status == "failed":
+ failed += 1
+ elif instance.metrics.get("unrecognized") and not unrecognized_sections:
+ info("%sFAILED%s: %s has unrecognized binary sections: %s" %
+ (COLOR_RED, COLOR_NORMAL, instance.name,
+ str(instance.metrics.get("unrecognized", []))))
+ failed += 1
- eleTestcase = ET.SubElement(
- eleTestsuite, 'testcase', classname="%s:%s" %(ti.platform.name, os.path.basename(ti.test.name)),
- name="%s" % (k), time=handler_time)
- if ti.results[k] in ['FAIL', 'BLOCK']:
- el = None
+ if self.total_tests and self.total_tests != self.total_skipped:
+ pass_rate = (float(self.total_tests - self.total_failed - self.total_skipped)/ float(self.total_tests - self.total_skipped))
+ else:
+ pass_rate = 0
- if ti.results[k] == 'FAIL':
- el = ET.SubElement(
- eleTestcase,
- 'failure',
- type="failure",
- message="failed")
- elif ti.results[k] == 'BLOCK':
- el = ET.SubElement(
- eleTestcase,
- 'error',
- type="failure",
- message="failed")
- p = os.path.join(options.outdir, ti.platform.name, ti.test.name)
- log_file = os.path.join(p, "handler.log")
+ info("{}{} of {}{} tests passed ({:.2%}), {}{}{} failed, {} skipped with {}{}{} warnings in {:.2f} seconds".format(
+ COLOR_RED if failed else COLOR_GREEN,
+ self.total_tests - self.total_failed - self.total_skipped,
+ self.total_tests,
+ COLOR_NORMAL,
+ pass_rate,
+ COLOR_RED if self.total_failed else COLOR_NORMAL,
+ self.total_failed,
+ COLOR_NORMAL,
+ self.total_skipped,
+ COLOR_YELLOW if self.warnings else COLOR_NORMAL,
+ self.warnings,
+ COLOR_NORMAL,
+ self.duration))
- if os.path.exists(log_file):
- with open(log_file, "rb") as f:
- log = f.read().decode("utf-8")
- filtered_string = ''.join(filter(lambda x: x in string.printable, log))
- el.text = filtered_string
+ platforms = set(p.platform for p in self.instances.values())
+ self.total_platforms = len(self.platforms)
+ if self.platforms:
+ info("In total {} test cases were executed on {} out of total {} platforms ({:02.2f}%)".format(
+ self.total_cases,
+ len(platforms),
+ self.total_platforms,
+ (100 * len(platforms) / len(self.platforms))
+ ))
- elif ti.results[k] == 'SKIP':
- el = ET.SubElement(
- eleTestcase,
- 'skipped',
- type="skipped",
- message="Skipped")
+ def save_reports(self):
+ if not self.instances:
+ return
- result = ET.tostring(eleTestsuites)
- f = open(report_file, 'wb')
- f.write(result)
- f.close()
+ report_name = "sanitycheck"
+ if options.report_name:
+ report_name = options.report_name
+ if options.report_dir:
+ os.mkdir(options.report_dir)
+ filename = os.path.join(options.report_dir, report_name)
+ outdir = options.report_dir
+ else:
+ filename = os.path.join(options.outdir, report_name)
+ outdir = options.outdir
- def testcase_xunit_report(self, filename, duration):
+ if not options.no_update:
+ self.xunit_report(filename + ".xml")
+ self.csv_report(filename + ".csv")
+ self.target_report(outdir)
+ if self.discards:
+ self.discard_report(filename + "_discard.csv")
+
+ if options.release:
+ self.csv_report(RELEASE_DATA)
+
+ if log_file:
+ log_file.close()
+
+ def load_hardware_map_from_cmdline(self, serial, platform):
+ device = {
+ "serial": serial,
+ "platform": platform,
+ "counter": 0,
+ "available": True,
+ "connected": True
+ }
+ self.connected_hardware = [device]
+
+ def load_hardware_map(self, map_file):
+ with open(map_file, 'r') as stream:
+ try:
+ self.connected_hardware = yaml.safe_load(stream)
+ except yaml.YAMLError as exc:
+ print(exc)
+ for i in self.connected_hardware:
+ i['counter'] = 0
+
+ def add_configurations(self):
+
+ for board_root in self.board_roots:
+ board_root = os.path.abspath(board_root)
+
+ debug("Reading platform configuration files under %s..." %
+ board_root)
+
+ for file in glob.glob(os.path.join(board_root, "*", "*", "*.yaml")):
+ verbose("Found plaform configuration " + file)
+ try:
+ platform = Platform()
+ platform.load(file)
+ if platform.sanitycheck:
+ self.platforms.append(platform)
+ if platform.default:
+ self.default_platforms.append(platform.name)
+
+ except RuntimeError as e:
+ error("E: %s: can't load: %s" % (file, e))
+ self.load_errors += 1
+
+ @staticmethod
+ def get_toolchain():
+ toolchain = os.environ.get("ZEPHYR_TOOLCHAIN_VARIANT", None) or \
+ os.environ.get("ZEPHYR_GCC_VARIANT", None)
+
+ if toolchain == "gccarmemb":
+ # Remove this translation when gccarmemb is no longer supported.
+ toolchain = "gnuarmemb"
+
try:
- if self.goals is None:
- raise SanityRuntimeError("execute() hasn't been run!")
+ if not toolchain:
+ raise SanityRuntimeError("E: Variable ZEPHYR_TOOLCHAIN_VARIANT is not defined")
except Exception as e:
print(str(e))
sys.exit(2)
+ return toolchain
+
+
+ def add_testcases(self):
+ for root in self.roots:
+ root = os.path.abspath(root)
+
+ debug("Reading test case configuration files under %s..." %root)
+
+ for dirpath, dirnames, filenames in os.walk(root, topdown=True):
+ verbose("scanning %s" % dirpath)
+ if 'sample.yaml' in filenames:
+ filename = 'sample.yaml'
+ elif 'testcase.yaml' in filenames:
+ filename = 'testcase.yaml'
+ else:
+ continue
+
+ verbose("Found possible test case in " + dirpath)
+
+ dirnames[:] = []
+ tc_path = os.path.join(dirpath, filename)
+ self.add_testcase(tc_path, root)
+
+ def add_testcase(self, tc_data_file, root):
+ try:
+ parsed_data = SanityConfigParser(tc_data_file, self.tc_schema)
+ parsed_data.load()
+
+ tc_path = os.path.dirname(tc_data_file)
+ workdir = os.path.relpath(tc_path, root)
+
+ for name in parsed_data.tests.keys():
+ tc = TestCase()
+ tc.name = tc.get_unique(root, workdir, name)
+
+ tc_dict = parsed_data.get_test(name, testcase_valid_keys)
+
+ tc.source_dir = tc_path
+ tc.yamlfile = tc_data_file
+
+ tc.id = name
+ tc.type = tc_dict["type"]
+ tc.tags = tc_dict["tags"]
+ tc.extra_args = tc_dict["extra_args"]
+ tc.extra_configs = tc_dict["extra_configs"]
+ tc.arch_whitelist = tc_dict["arch_whitelist"]
+ tc.arch_exclude = tc_dict["arch_exclude"]
+ tc.skip = tc_dict["skip"]
+ tc.platform_exclude = tc_dict["platform_exclude"]
+ tc.platform_whitelist = tc_dict["platform_whitelist"]
+ tc.toolchain_exclude = tc_dict["toolchain_exclude"]
+ tc.toolchain_whitelist = tc_dict["toolchain_whitelist"]
+ tc.tc_filter = tc_dict["filter"]
+ tc.timeout = tc_dict["timeout"]
+ tc.harness = tc_dict["harness"]
+ tc.harness_config = tc_dict["harness_config"]
+ tc.build_only = tc_dict["build_only"]
+ tc.build_on_all = tc_dict["build_on_all"]
+ tc.slow = tc_dict["slow"]
+ tc.min_ram = tc_dict["min_ram"]
+ tc.depends_on = tc_dict["depends_on"]
+ tc.min_flash = tc_dict["min_flash"]
+ tc.extra_sections = tc_dict["extra_sections"]
+
+ tc.parse_subcases(tc_path)
+
+ if tc.name:
+ self.testcases[tc.name] = tc
+
+ except Exception as e:
+ error("E: %s: can't load (skipping): %s" % (tc_data_file, e))
+ self.load_errors += 1
+ return False
+
+ return True
+
+
+ def get_platform(self, name):
+ selected_platform = None
+ for platform in self.platforms:
+ if platform.name == name:
+ selected_platform = platform
+ break
+ return selected_platform
+
+ def get_last_failed(self):
+ last_run = os.path.join(options.outdir, "sanitycheck.csv")
+ try:
+ if not os.path.exists(last_run):
+ raise SanityRuntimeError("Couldn't find last sanitycheck run.: %s" %last_run)
+ except Exception as e:
+ print(str(e))
+ sys.exit(2)
+
+ total_tests = 0
+ with open(last_run, "r") as fp:
+ cr = csv.DictReader(fp)
+ instance_list = []
+ for row in cr:
+ total_tests += 1
+ if row["passed"] == "True":
+ continue
+ test = row["test"]
+ platform = self.get_platform(row["platform"])
+ instance = TestInstance(self.testcases[test], platform, self.outdir)
+ instance.create_overlay(platform.name)
+ instance_list.append(instance)
+ self.add_instances(instance_list)
+
+ tests_to_run = len(self.instances)
+ info("%d tests passed already, retyring %d tests" %(total_tests - tests_to_run, tests_to_run))
+
+ def load_from_file(self, file):
+ try:
+ if not os.path.exists(file):
+ raise SanityRuntimeError(
+ "Couldn't find input file with list of tests.")
+ except Exception as e:
+ print(str(e))
+ sys.exit(2)
+
+ with open(file, "r") as fp:
+ cr = csv.DictReader(fp)
+ instance_list = []
+ for row in cr:
+ if row["arch"] == "arch":
+ continue
+ test = row["test"]
+ platform = self.get_platform(row["platform"])
+ instance = TestInstance(self.testcases[test], platform, self.outdir)
+ instance.create_overlay(platform.name)
+ instance_list.append(instance)
+ self.add_instances(instance_list)
+
+
+ def apply_filters(self):
+
+ toolchain = self.get_toolchain()
+
+ discards = {}
+ platform_filter = options.platform
+ testcase_filter = run_individual_tests
+ arch_filter = options.arch
+ tag_filter = options.tag
+ exclude_tag = options.exclude_tag
+
+ verbose("platform filter: " + str(platform_filter))
+ verbose(" arch_filter: " + str(arch_filter))
+ verbose(" tag_filter: " + str(tag_filter))
+ verbose(" exclude_tag: " + str(exclude_tag))
+
+ default_platforms = False
+
+ if platform_filter:
+ platforms = list(filter(lambda p: p.name in platform_filter, self.platforms))
+ else:
+ platforms = self.platforms
+
+ if options.all:
+ info("Selecting all possible platforms per test case")
+ # When --all used, any --platform arguments ignored
+ platform_filter = []
+ elif not platform_filter:
+ info("Selecting default platforms per test case")
+ default_platforms = True
+
+ info("Building initial testcase list...")
+
+ for tc_name, tc in self.testcases.items():
+ # list of instances per testcase, aka configurations.
+ instance_list = []
+ for plat in platforms:
+ instance = TestInstance(tc, plat, self.outdir)
+
+ if (plat.arch == "unit") != (tc.type == "unit"):
+ # Discard silently
+ continue
+
+ if options.device_testing and instance.build_only:
+ discards[instance] = "Not runnable on device"
+ continue
+
+ if tc.skip:
+ discards[instance] = "Skip filter"
+ continue
+
+ if tc.build_on_all and not platform_filter:
+ platform_filter = []
+
+ if tag_filter and not tc.tags.intersection(tag_filter):
+ discards[instance] = "Command line testcase tag filter"
+ continue
+
+ if exclude_tag and tc.tags.intersection(exclude_tag):
+ discards[instance] = "Command line testcase exclude filter"
+ continue
+
+ if testcase_filter and tc_name not in testcase_filter:
+ discards[instance] = "Testcase name filter"
+ continue
+
+ if arch_filter and plat.arch not in arch_filter:
+ discards[instance] = "Command line testcase arch filter"
+ continue
+
+ if tc.arch_whitelist and plat.arch not in tc.arch_whitelist:
+ discards[instance] = "Not in test case arch whitelist"
+ continue
+
+ if tc.arch_exclude and plat.arch in tc.arch_exclude:
+ discards[instance] = "In test case arch exclude"
+ continue
+
+ if tc.platform_exclude and plat.name in tc.platform_exclude:
+ discards[instance] = "In test case platform exclude"
+ continue
+
+ if tc.toolchain_exclude and toolchain in tc.toolchain_exclude:
+ discards[instance] = "In test case toolchain exclude"
+ continue
+
+ if platform_filter and plat.name not in platform_filter:
+ discards[instance] = "Command line platform filter"
+ continue
+
+ if tc.platform_whitelist and plat.name not in tc.platform_whitelist:
+ discards[instance] = "Not in testcase platform whitelist"
+ continue
+
+ if tc.toolchain_whitelist and toolchain not in tc.toolchain_whitelist:
+ discards[instance] = "Not in testcase toolchain whitelist"
+ continue
+
+ if not plat.env_satisfied:
+ discards[instance] = "Environment ({}) not satisfied".format(", ".join(plat.env))
+ continue
+
+ if not options.force_toolchain \
+ and toolchain and (toolchain not in plat.supported_toolchains) \
+ and tc.type != 'unit':
+ discards[instance] = "Not supported by the toolchain"
+ continue
+
+ if plat.ram < tc.min_ram:
+ discards[instance] = "Not enough RAM"
+ continue
+
+ if tc.depends_on:
+ dep_intersection = tc.depends_on.intersection(set(plat.supported))
+ if dep_intersection != set(tc.depends_on):
+ discards[instance] = "No hardware support"
+ continue
+
+ if plat.flash < tc.min_flash:
+ discards[instance] = "Not enough FLASH"
+ continue
+
+ if set(plat.ignore_tags) & tc.tags:
+ discards[instance] = "Excluded tags per platform"
+ continue
+
+ # if nothing stopped us until now, it means this configuration
+ # needs to be added.
+ instance_list.append(instance)
+
+ # no configurations, so jump to next testcase
+ if not instance_list:
+ continue
+
+ # if sanitycheck was launched with no platform options at all, we
+ # take all default platforms
+ if default_platforms and not tc.build_on_all:
+ if tc.platform_whitelist:
+ a = set(self.default_platforms)
+ b = set(tc.platform_whitelist)
+ c = a.intersection(b)
+ if c:
+ aa = list( filter( lambda tc: tc.platform.name in c, instance_list))
+ self.add_instances(aa)
+ else:
+ self.add_instances(instance_list[:1])
+ else:
+ instances = list( filter( lambda tc: tc.platform.default, instance_list))
+ self.add_instances(instances)
+
+ for instance in list(filter(lambda tc: not tc.platform.default, instance_list)):
+ discards[instance] = "Not a default test platform"
+
+ else:
+ self.add_instances(instance_list)
+
+ for _, case in self.instances.items():
+ case.create_overlay(case.platform.name)
+
+ self.discards = discards
+
+ return discards
+
+ def add_instances(self, instance_list):
+ for instance in instance_list:
+ self.instances[instance.name] = instance
+
+ def add_tasks_to_queue(self):
+ for instance in self.instances.values():
+ if options.test_only:
+ if instance.run:
+ pipeline.put({"op": "run", "test": instance, "status": "built"})
+ else:
+ if instance.status not in ['passed', 'skipped']:
+ instance.status = None
+ pipeline.put({"op": "cmake", "test": instance})
+
+ return "DONE FEEDING"
+
+ def execute(self):
+ def calc_one_elf_size(instance):
+ if instance.status not in ["failed", "skipped"]:
+ if instance.platform.type != "native":
+ size_calc = instance.calculate_sizes()
+ instance.metrics["ram_size"] = size_calc.get_ram_size()
+ instance.metrics["rom_size"] = size_calc.get_rom_size()
+ instance.metrics["unrecognized"] = size_calc.unrecognized_sections()
+ else:
+ instance.metrics["ram_size"] = 0
+ instance.metrics["rom_size"] = 0
+ instance.metrics["unrecognized"] = []
+
+ instance.metrics["handler_time"] = instance.handler.duration if instance.handler else 0
+
+ info("Adding tasks to the queue...")
+ # We can use a with statement to ensure threads are cleaned up promptly
+ with BoundedExecutor(bound=self.jobs, max_workers=self.jobs) as executor:
+
+ # start a future for a thread which sends work in through the queue
+ future_to_test = {
+ executor.submit(self.add_tasks_to_queue): 'FEEDER DONE'}
+
+ while future_to_test:
+ # check for status of the futures which are currently working
+ done, _ = concurrent.futures.wait(
+ future_to_test, timeout=0.25,
+ return_when=concurrent.futures.FIRST_COMPLETED)
+
+ # if there is incoming work, start a new future
+ while not pipeline.empty():
+ # fetch a url from the queue
+ message = pipeline.get()
+ test = message['test']
+
+ # Start the load operation and mark the future with its URL
+ pb = ProjectBuilder(self, test)
+ future_to_test[executor.submit(pb.process, message)] = test.name
+
+ # process any completed futures
+ for future in done:
+ test = future_to_test[future]
+ try:
+ data = future.result()
+ except Exception as exc:
+ print('%r generated an exception: %s' % (test, exc))
+ else:
+ if data:
+ verbose(data)
+
+ # remove the now completed future
+ del future_to_test[future]
+
+ if options.enable_size_report and not options.cmake_only:
+ # Parallelize size calculation
+ executor = concurrent.futures.ThreadPoolExecutor(self.jobs)
+ futures = [executor.submit(calc_one_elf_size, instance)
+ for instance in self.instances.values()]
+ concurrent.futures.wait(futures)
+ else:
+ for instance in self.instances.values():
+ instance.metrics["ram_size"] = 0
+ instance.metrics["rom_size"] = 0
+ instance.metrics["handler_time"] = instance.handler.duration if instance.handler else 0
+ instance.metrics["unrecognized"] = []
+
+
+ def discard_report(self, filename):
+
+ try:
+ if self.discards is None:
+ raise SanityRuntimeError("apply_filters() hasn't been run!")
+ except Exception as e:
+ error(str(e))
+ sys.exit(2)
+
+ with open(filename, "wt") as csvfile:
+ fieldnames = ["test", "arch", "platform", "reason"]
+ cw = csv.DictWriter(csvfile, fieldnames, lineterminator=os.linesep)
+ cw.writeheader()
+ for instance, reason in sorted(self.discards.items()):
+ rowdict = {"test": instance.testcase.name,
+ "arch": instance.platform.arch,
+ "platform": instance.platform.name,
+ "reason": reason}
+ cw.writerow(rowdict)
+
+
+ def target_report(self, outdir):
+ run = "Sanitycheck"
+ eleTestsuite = None
+
+ platforms = {inst.platform.name for _,inst in self.instances.items()}
+ for platform in platforms:
+ errors = 0
+ passes = 0
+ fails = 0
+ duration = 0
+ skips = 0
+ for _, instance in self.instances.items():
+ if instance.platform.name != platform:
+ continue
+
+ handler_time = instance.metrics.get('handler_time', 0)
+ duration += handler_time
+ for k in instance.results.keys():
+ if instance.results[k] == 'PASS':
+ passes += 1
+ elif instance.results[k] == 'BLOCK':
+ errors += 1
+ elif instance.results[k] == 'SKIP':
+ skips += 1
+ else:
+ fails += 1
+
+ eleTestsuites = ET.Element('testsuites')
+ eleTestsuite = ET.SubElement(eleTestsuites, 'testsuite',
+ name=run, time="%f" % duration,
+ tests="%d" % (errors + passes + fails),
+ failures="%d" % fails,
+ errors="%d" % errors, skipped="%d" %skips)
+
+ handler_time = 0
+
+ # print out test results
+ for _, instance in self.instances.items():
+ if instance.platform.name != platform:
+ continue
+ handler_time = instance.metrics.get('handler_time', 0)
+ for k in instance.results.keys():
+ eleTestcase = ET.SubElement(
+ eleTestsuite, 'testcase', classname="%s:%s" %(instance.platform.name, os.path.basename(instance.testcase.name)),
+ name="%s" % (k), time="%f" %handler_time)
+ if instance.results[k] in ['FAIL', 'BLOCK']:
+ el = None
+
+ if instance.results[k] == 'FAIL':
+ el = ET.SubElement(
+ eleTestcase,
+ 'failure',
+ type="failure",
+ message="failed")
+ elif instance.results[k] == 'BLOCK':
+ el = ET.SubElement(
+ eleTestcase,
+ 'error',
+ type="failure",
+ message="failed")
+ p = os.path.join(options.outdir, instance.platform.name, instance.testcase.name)
+ log_file = os.path.join(p, "handler.log")
+
+ if os.path.exists(log_file):
+ with open(log_file, "rb") as f:
+ log = f.read().decode("utf-8")
+ filtered_string = ''.join(filter(lambda x: x in string.printable, log))
+ el.text = filtered_string
+
+ elif instance.results[k] == 'SKIP':
+ el = ET.SubElement(
+ eleTestcase,
+ 'skipped',
+ type="skipped",
+ message="Skipped")
+
+
+ result = ET.tostring(eleTestsuites)
+ with open(os.path.join(outdir, platform + ".xml"), 'wb') as f:
+ f.write(result)
+
+
+ def xunit_report(self, filename):
fails = 0
passes = 0
errors = 0
+ skips = 0
+ duration = 0
- for name, goal in self.goals.items():
- if goal.failed:
- if goal.reason in ['build_error', 'handler_crash']:
+ for instance in self.instances.values():
+ handler_time = instance.metrics.get('handler_time', 0)
+ duration += handler_time
+ if instance.status == "failed":
+ if instance.reason in ['build_error', 'handler_crash']:
errors += 1
else:
fails += 1
+ elif instance.status == 'skipped':
+ skips += 1
else:
passes += 1
@@ -2669,6 +2910,8 @@
eleTestsuite = None
append = options.only_failed
+ # When we re-run the tests, we re-use the results and update only with
+ # the newly run tests.
if os.path.exists(filename) and append:
tree = ET.parse(filename)
eleTestsuites = tree.getroot()
@@ -2676,39 +2919,40 @@
else:
eleTestsuites = ET.Element('testsuites')
eleTestsuite = ET.SubElement(eleTestsuites, 'testsuite',
- name=run, time="%d" % duration,
- tests="%d" % (errors + passes + fails),
+ name=run, time="%f" % duration,
+ tests="%d" % (errors + passes + fails + skips),
failures="%d" % fails,
- errors="%d" % errors, skip="0")
+ errors="%d" %(errors), skip="%s" %(skips))
- handler_time = "0"
- for name, goal in self.goals.items():
+ for instance in self.instances.values():
- i = self.instances[name]
+ # remove testcases that are a re-run
if append:
for tc in eleTestsuite.findall('testcase'):
if tc.get('classname') == "%s:%s" % (
- i.platform.name, i.test.name):
+ instance.platform.name, instance.testcase.name):
eleTestsuite.remove(tc)
- if not goal.failed and goal.handler:
- handler_time = "%s" %(goal.metrics["handler_time"])
+ handler_time = 0
+ if instance.status != "failed" and instance.handler:
+ handler_time = instance.metrics.get("handler_time", 0)
eleTestcase = ET.SubElement(
eleTestsuite, 'testcase', classname="%s:%s" %
- (i.platform.name, i.test.name), name="%s" %
- (name), time=handler_time)
- if goal.failed:
+ (instance.platform.name, instance.testcase.name), name="%s" %
+ (instance.testcase.name), time="%f" %handler_time)
+
+ if instance.status == "failed":
failure = ET.SubElement(
eleTestcase,
'failure',
type="failure",
- message=goal.reason)
- p = ("%s/%s/%s" % (options.outdir, i.platform.name, i.test.name))
+ message=instance.reason)
+ p = ("%s/%s/%s" % (options.outdir, instance.platform.name, instance.testcase.name))
bl = os.path.join(p, "build.log")
hl = os.path.join(p, "handler.log")
log_file = bl
- if goal.reason != 'build_error':
+ if instance.reason != 'Build error':
if os.path.exists(hl):
log_file = hl
else:
@@ -2720,42 +2964,39 @@
filtered_string = ''.join(filter(lambda x: x in string.printable, log))
failure.text = filtered_string
f.close()
+ elif instance.status == "skipped":
+ ET.SubElement( eleTestcase, 'skipped', type="skipped", message="Skipped")
result = ET.tostring(eleTestsuites)
- f = open(filename, 'wb')
- f.write(result)
- f.close()
+ with open(filename, 'wb') as report:
+ report.write(result)
- def testcase_report(self, filename):
- try:
- if self.goals is None:
- raise SanityRuntimeError("execute() hasn't been run!")
- except Exception as e:
- print(str(e))
- sys.exit(2)
+ def csv_report(self, filename):
with open(filename, "wt") as csvfile:
fieldnames = ["test", "arch", "platform", "passed", "status",
- "extra_args", "qemu", "handler_time", "ram_size",
+ "extra_args", "handler", "handler_time", "ram_size",
"rom_size"]
cw = csv.DictWriter(csvfile, fieldnames, lineterminator=os.linesep)
cw.writeheader()
- for name, goal in sorted(self.goals.items()):
- i = self.instances[name]
- rowdict = {"test": i.test.name,
- "arch": i.platform.arch,
- "platform": i.platform.name,
- "extra_args": " ".join(i.test.extra_args),
- "qemu": i.platform.qemu_support}
- if goal.failed:
+ for instance in sorted(self.instances.values()):
+ rowdict = {"test": instance.testcase.name,
+ "arch": instance.platform.arch,
+ "platform": instance.platform.name,
+ "extra_args": " ".join(instance.testcase.extra_args),
+ "handler": instance.platform.simulation}
+
+ if instance.status in ["failed", "timeout"]:
rowdict["passed"] = False
- rowdict["status"] = goal.reason
+ rowdict["status"] = instance.reason
else:
rowdict["passed"] = True
- if goal.handler:
- rowdict["handler_time"] = goal.metrics["handler_time"]
- rowdict["ram_size"] = goal.metrics["ram_size"]
- rowdict["rom_size"] = goal.metrics["rom_size"]
+ if instance.handler:
+ rowdict["handler_time"] = instance.metrics.get("handler_time", 0)
+ ram_size = instance.metrics.get("ram_size", 0)
+ rom_size = instance.metrics.get("rom_size", 0)
+ rowdict["ram_size"] = ram_size
+ rowdict["rom_size"] = rom_size
cw.writerow(rowdict)
@@ -2808,15 +3049,10 @@
action="store_true",
help="Run only those tests that failed the previous sanity check "
"invocation.")
+
parser.add_argument(
- "-c", "--config", action="append",
- help="Specify platform configuration values filtering. This can be "
- "specified two ways: <config>=<value> or just <config>. The "
- "defconfig for all platforms will be "
- "checked. For the <config>=<value> case, only match defconfig "
- "that have that value defined. For the <config> case, match "
- "defconfig that have that value assigned to any value. "
- "Prepend a '!' to invert the match.")
+ "--retry-failed", type=int, default=0,
+ help="Retry failing tests again, up to the number of times specified.")
test_xor_subtest = case_select.add_mutually_exclusive_group()
@@ -2840,14 +3076,33 @@
"ignored.")
parser.add_argument(
- "-o", "--testcase-report",
- help="""Output a CSV spreadsheet containing results of the test run.
- The handler_time column is left blank for tests that were only
- compiled and not run.""")
+ "-o", "--report-dir",
+ help="""Output reports containing results of the test run into the
+ specified directory.
+ The output will be both in CSV and JUNIT format
+ (sanitycheck.csv and sanitycheck.xml).
+ """)
+
parser.add_argument(
- "-d", "--discard-report",
- help="Output a CSV spreadsheet showing tests that were skipped "
- "and why")
+ "--report-name",
+ help="""Create a report with a custom name.
+ """)
+
+ parser.add_argument("--detailed-report",
+ action="store",
+ metavar="FILENAME",
+ help="""Generate a junit report with detailed testcase results.
+ Unlike the CSV file produced by --testcase-report, this XML
+ report includes only tests which have run and none which were
+ merely built. If an image with multiple tests crashes early then
+ later tests are not accounted for either.""")
+
+ parser.add_argument("--report-excluded",
+ action="store_true",
+ help="""List all tests that are never run based on current scope and
+ coverage. If you are looking for accurate results, run this with
+ --all, but this will take a while...""")
+
parser.add_argument("--compare-report",
help="Use this report file for size comparison")
@@ -2871,12 +3126,6 @@
parser.add_argument("--list-tags", action="store_true",
help="list all tags in selected tests")
- parser.add_argument("--report-excluded",
- action="store_true",
- help="""List all tests that are never run based on current scope and
- coverage. If you are looking for accurate results, run this with
- --all, but this will take a while...""")
-
case_select.add_argument("--list-tests", action="store_true",
help="""List of all sub-test functions recursively found in
all --testcase-root arguments. Note different sub-tests can share
@@ -2890,14 +3139,6 @@
metavar="FILENAME",
help="Export tests case meta-data to a file in CSV format.")
- parser.add_argument("--detailed-report",
- action="store",
- metavar="FILENAME",
- help="""Generate a junit report with detailed testcase results.
- Unlike the CSV file produced by --testcase-report, this XML
- report includes only tests which have run and none which were
- merely built. If an image with multiple tests crashes early then
- later tests are not accounted for either.""")
parser.add_argument("--timestamps",
action="store_true",
@@ -2953,14 +3194,14 @@
test_or_build.add_argument(
"-b", "--build-only", action="store_true",
help="Only build the code, do not execute any of it in QEMU")
+
test_or_build.add_argument(
"--test-only", action="store_true",
help="""Only run device tests with current artifacts, do not build
the code""")
parser.add_argument(
"--cmake-only", action="store_true",
- help="Test on device directly. Specify the serial device to "
- "use with the --device-serial option.")
+ help="Only run cmake, do not build or run.")
parser.add_argument(
"-j", "--jobs", type=int,
@@ -2968,17 +3209,6 @@
"overcommited by factor 2 when --build-only")
parser.add_argument(
- "--device-testing", action="store_true",
- help="Test on device directly. Specify the serial device to "
- "use with the --device-serial option.")
-
- parser.add_argument(
- "-X", "--fixture", action="append", default=[],
- help="Specify a fixture that a board might support")
- parser.add_argument(
- "--device-serial",
- help="Serial device for accessing the board (e.g., /dev/ttyACM0)")
- parser.add_argument(
"--show-footprint", action="store_true",
help="Show footprint statistics and deltas since last release."
)
@@ -2994,7 +3224,7 @@
"--footprint-threshold=0")
parser.add_argument(
"-O", "--outdir",
- default="%s/sanity-out" % os.getcwd(),
+ default=os.path.join(os.getcwd(),"sanity-out"),
help="Output directory for logs and binaries. "
"Default is 'sanity-out' in the current directory. "
"This directory will be deleted unless '--no-clean' is set.")
@@ -3013,9 +3243,11 @@
"%s/scripts/sanity_chk/boards" % ZEPHYR_BASE]
parser.add_argument(
- "-A", "--board-root", action="append", default=board_root_list,
- help="Directory to search for board configuration files. All .yaml "
- "files in the directory will be processed.")
+ "-A", "--board-root", default=board_root_list,
+ help="""Directory to search for board configuration files. All .yaml
+files in the directory will be processed. The directory should have the same
+structure in the main Zephyr tree: boards/<arch>/<board_name>/""")
+
parser.add_argument(
"-z", "--size", action="append",
help="Don't run sanity checks. Instead, produce a report to "
@@ -3037,8 +3269,8 @@
help="deprecated, left for compatibility")
parser.add_argument("-Q", "--error-on-deprecations", action="store_false",
help="Error on deprecation warnings.")
- parser.add_argument("--disable-size-report", action="store_true",
- help="Skip expensive computation of ram/rom segment sizes.")
+ parser.add_argument("--enable-size-report", action="store_true",
+ help="Enable expensive computation of RAM/ROM segment sizes.")
parser.add_argument(
"-x", "--extra-args", action="append", default=[],
@@ -3056,6 +3288,29 @@
)
parser.add_argument(
+ "--device-testing", action="store_true",
+ help="Test on device directly. Specify the serial device to "
+ "use with the --device-serial option.")
+
+ parser.add_argument(
+ "-X", "--fixture", action="append", default=[],
+ help="Specify a fixture that a board might support")
+ parser.add_argument(
+ "--device-serial",
+ help="Serial device for accessing the board (e.g., /dev/ttyACM0)")
+
+ parser.add_argument("--generate-hardware-map",
+ help="""Probe serial devices connected to this platform
+ and create a hardware map file to be used with
+ --device-testing
+ """)
+
+ parser.add_argument("--hardware-map",
+ help="""Load hardware map from a file. This will be used
+ for testing on hardware that is listed in the file.
+ """)
+
+ parser.add_argument(
"--west-flash", nargs='?', const=[],
help="""Uses west instead of ninja or make to flash when running with
--device-testing. Supports comma-separated argument list.
@@ -3080,10 +3335,6 @@
"""
)
- parser.add_argument("--gcov-tool", default=None,
- help="Path to the gcov tool to use for code coverage "
- "reports")
-
parser.add_argument("--enable-coverage", action="store_true",
help="Enable code coverage using gcov.")
@@ -3096,12 +3347,16 @@
"This option may be used multiple times. "
"Default to what was selected with --platform.")
+ parser.add_argument("--gcov-tool", default=None,
+ help="Path to the gcov tool to use for code coverage "
+ "reports")
+
return parser.parse_args()
def log_info(filename):
filename = os.path.relpath(os.path.realpath(filename))
- if INLINE_LOGS:
+ if options.inline_logs:
info("{:-^100}".format(filename))
try:
@@ -3117,75 +3372,6 @@
else:
info("\tsee: " + COLOR_YELLOW + filename + COLOR_NORMAL)
-
-def terse_test_cb(instances, goals, goal):
- total_tests = len(goals)
- total_done = 0
- total_failed = 0
-
- for k, g in goals.items():
- if g.finished:
- total_done += 1
- if g.failed:
- total_failed += 1
-
- if goal.failed:
- i = instances[goal.name]
- info(
- "\n\n{:<25} {:<50} {}FAILED{}: {}".format(
- i.platform.name,
- i.test.name,
- COLOR_RED,
- COLOR_NORMAL,
- goal.reason), False)
- log_info(goal.get_error_log())
- info("", False)
-
- sys.stdout.write(
- "\rtotal complete: %s%4d/%4d%s %2d%% failed: %s%4d%s" %
- (COLOR_GREEN, total_done, total_tests, COLOR_NORMAL,
- int((float(total_done) / total_tests) * 100),
- COLOR_RED if total_failed > 0 else COLOR_NORMAL, total_failed,
- COLOR_NORMAL))
- sys.stdout.flush()
-
-
-def chatty_test_cb(instances, goals, goal):
- i = instances[goal.name]
-
- if VERBOSE < 2 and not goal.finished:
- return
-
- total_tests = len(goals)
- total_tests_width = len(str(total_tests))
- total_done = 0
-
- for k, g in goals.items():
- if g.finished:
- total_done += 1
-
- if goal.failed:
- status = COLOR_RED + "FAILED" + COLOR_NORMAL + ": " + goal.reason
- elif goal.finished:
- status = COLOR_GREEN + "PASSED" + COLOR_NORMAL
- else:
- status = goal.make_state
-
- if goal.handler:
- handler_type = goal.handler.type_str
- htime = goal.metrics.get("handler_time", None)
- if htime:
- handler_type += " {:.3f}s".format(htime)
- else:
- handler_type = "build"
-
- info("{:>{}}/{} {:<25} {:<50} {} ({})".format(
- total_done, total_tests_width, total_tests, i.platform.name,
- i.test.name, status, handler_type))
- if goal.failed:
- log_info(goal.get_error_log())
-
-
def size_report(sc):
info(sc.filename)
info("SECTION NAME VMA LMA SIZE HEX SZ TYPE")
@@ -3243,7 +3429,7 @@
filename = (filename[:-4]) +"gcno"
try:
os.remove(filename)
- except:
+ except Exception:
pass
continue
@@ -3283,9 +3469,9 @@
"--output-file", ztestfile,
"--rc", "lcov_branch_coverage=1"],
stdout=coveragelog)
- files = [coveragefile, ztestfile];
+ files = [coveragefile, ztestfile]
else:
- files = [coveragefile];
+ files = [coveragefile]
for i in ignores:
subprocess.call(
@@ -3303,19 +3489,131 @@
stdout=coveragelog)
if ret==0:
info("HTML report generated: %s"%
- os.path.join(outdir, "coverage","index.html"));
+ os.path.join(outdir, "coverage","index.html"))
+def get_generator():
+ if options.ninja:
+ generator_cmd = "ninja"
+ generator = "Ninja"
+ else:
+ generator_cmd = "make"
+ generator = "Unix Makefiles"
+ return generator_cmd, generator
+
+
+def export_tests(filename, tests):
+ with open(filename, "wt") as csvfile:
+ fieldnames = ['section', 'subsection', 'title', 'reference']
+ cw = csv.DictWriter(csvfile, fieldnames, lineterminator=os.linesep)
+ for test in tests:
+ data = test.split(".")
+ subsec = " ".join(data[1].split("_")).title()
+ rowdict = {
+ "section": data[0].capitalize(),
+ "subsection": subsec,
+ "title": test,
+ "reference": test
+ }
+ cw.writerow(rowdict)
+
+def get_unique_tests(suite):
+ unq = []
+ run_individual_tests = []
+ for _, tc in suite.testcases.items():
+ for c in tc.cases:
+ if options.sub_test and c in options.sub_test:
+ if tc.name not in run_individual_tests:
+ run_individual_tests.append(tc.name)
+ unq.append(c)
+
+ return unq
+
+
+
+def native_and_unit_first(a, b):
+ if a[0].startswith('unit_testing'):
+ return -1
+ if b[0].startswith('unit_testing'):
+ return 1
+ if a[0].startswith('native_posix'):
+ return -1
+ if b[0].startswith('native_posix'):
+ return 1
+ if a[0].split("/",1)[0].endswith("_bsim"):
+ return -1
+ if b[0].split("/",1)[0].endswith("_bsim"):
+ return 1
+
+ return (a > b) - (a < b)
+
+
+run_individual_tests = None
+options = None
def main():
start_time = time.time()
- global VERBOSE, INLINE_LOGS, JOBS, log_file
+ global VERBOSE, log_file
global options
global run_individual_tests
- # XXX: Workaround for #17239
- resource.setrlimit(resource.RLIMIT_NOFILE, (4096, 4096))
options = parse_arguments()
+
+ if options.generate_hardware_map:
+ from serial.tools import list_ports
+ serial_devices = list_ports.comports()
+ filtered = []
+ for d in serial_devices:
+ if d.manufacturer in ['ARM', 'SEGGER', 'MBED', 'STMicroelectronics', 'Atmel Corp.']:
+ s_dev = {}
+ s_dev['platform'] = "unknown"
+ s_dev['id'] = d.serial_number
+ s_dev['serial'] = d.device
+ s_dev['product'] = d.product
+ if s_dev['product'] in ['DAPLink CMSIS-DAP', 'MBED CMSIS-DAP']:
+ s_dev['runner'] = "pyocd"
+ else:
+ s_dev['runner'] = "unknown"
+ s_dev['available'] = True
+ s_dev['connected'] = True
+ filtered.append(s_dev)
+ else:
+ print("Unsupported device (%s): %s" %(d.manufacturer, d))
+
+ if os.path.exists(options.generate_hardware_map):
+ # use existing map
+
+ with open(options.generate_hardware_map, 'r') as yaml_file:
+ hwm = yaml.load(yaml_file, Loader=yaml.FullLoader)
+ # disconnect everything
+ for h in hwm:
+ h['connected'] = False
+
+ for d in filtered:
+ for h in hwm:
+ if d['id'] == h['id'] and d['product'] == h['product']:
+ print("Already in map: %s (%s)" %(d['product'], d['id']))
+ h['connected'] = True
+ h['serial'] = d['serial']
+ d['match'] = True
+
+ new = list(filter(lambda n: not n.get('match', False), filtered))
+ hwm = hwm + new
+
+ #import pprint
+ #pprint.pprint(hwm)
+ with open(options.generate_hardware_map, 'w') as yaml_file:
+ yaml.dump(hwm, yaml_file, default_flow_style=False)
+
+
+ else:
+ # create new file
+ with open(options.generate_hardware_map, 'w') as yaml_file:
+ yaml.dump(filtered, yaml_file, default_flow_style=False)
+
+ return
+
+
if options.west_runner and not options.west_flash:
error("west-runner requires west-flash to be enabled")
sys.exit(1)
@@ -3335,29 +3633,11 @@
size_report(SizeCalculator(fn, []))
sys.exit(0)
-
- if options.device_testing:
- if options.device_serial is None or len(options.platform) != 1:
- sys.exit(1)
-
VERBOSE += options.verbose
- INLINE_LOGS = options.inline_logs
+
if options.log_file:
log_file = open(options.log_file, "w")
- if options.jobs:
- JOBS = options.jobs
- elif options.build_only:
- JOBS = multiprocessing.cpu_count() * 2
- else:
- JOBS = multiprocessing.cpu_count()
-
- # Decrease JOBS for Ninja, if jobs weren't explicitly set
- if options.ninja and not options.jobs:
- JOBS = int(JOBS * 0.75)
-
- info("JOBS: %d" % JOBS);
-
if options.subset:
subset, sets = options.subset.split("/")
if int(subset) > 0 and int(sets) >= int(subset):
@@ -3366,22 +3646,53 @@
error("You have provided a wrong subset value: %s." % options.subset)
return
- if os.path.exists(options.outdir) and not options.no_clean:
- info("Cleaning output directory " + options.outdir)
- shutil.rmtree(options.outdir)
+ # Cleanup
+
+ if options.no_clean or options.only_failed or options.test_only:
+ if os.path.exists(options.outdir):
+ info("Keeping artifacts untouched")
+ elif os.path.exists(options.outdir):
+ for i in range(1,100):
+ new_out = options.outdir + ".{}".format(i)
+ if not os.path.exists(new_out):
+ info("Renaming output directory to {}".format(new_out))
+ shutil.move(options.outdir, new_out)
+ break
+ #shutil.rmtree("%s.old" %options.outdir)
if not options.testcase_root:
options.testcase_root = [os.path.join(ZEPHYR_BASE, "tests"),
os.path.join(ZEPHYR_BASE, "samples")]
- ts = TestSuite(options.board_root, options.testcase_root, options.outdir)
+ suite = TestSuite(options.board_root, options.testcase_root, options.outdir)
+ suite.add_testcases()
+ suite.add_configurations()
- if ts.load_errors:
+ if options.device_testing:
+ if options.hardware_map:
+ suite.load_hardware_map(options.hardware_map)
+ if not options.platform:
+ options.platform = []
+ for platform in suite.connected_hardware:
+ if platform['connected']:
+ options.platform.append(platform['platform'])
+
+ elif options.device_serial: #back-ward compatibility
+ if options.platform and len(options.platform) == 1:
+ suite.load_hardware_map_from_cmdline(options.device_serial,
+ options.platform[0])
+ else:
+ error("""When --device-testing is used with --device-serial, only one
+ platform is allowed""")
+
+
+
+ if suite.load_errors:
sys.exit(1)
if options.list_tags:
tags = set()
- for n,tc in ts.testcases.items():
+ for _, tc in suite.testcases.items():
tags = tags.union(tc.tags)
for t in tags:
@@ -3389,26 +3700,10 @@
return
-
- def export_tests(filename, tests):
- with open(filename, "wt") as csvfile:
- fieldnames = ['section', 'subsection', 'title', 'reference']
- cw = csv.DictWriter(csvfile, fieldnames, lineterminator=os.linesep)
- for test in tests:
- data = test.split(".")
- subsec = " ".join(data[1].split("_")).title()
- rowdict = {
- "section": data[0].capitalize(),
- "subsection": subsec,
- "title": test,
- "reference": test
- }
- cw.writerow(rowdict)
-
if options.export_tests:
cnt = 0
unq = []
- for n,tc in ts.testcases.items():
+ for _, tc in suite.testcases.items():
for c in tc.cases:
unq.append(c)
@@ -3421,22 +3716,9 @@
if options.test:
run_individual_tests = options.test
-
- def get_unique_tests(ts):
- unq = []
- run_individual_tests = []
- for n,tc in ts.testcases.items():
- for c in tc.cases:
- if options.sub_test and c in options.sub_test:
- if tc.name not in run_individual_tests:
- run_individual_tests.append(tc.name)
- unq.append(c)
-
- return unq
-
if options.list_tests or options.sub_test:
cnt = 0
- unq = get_unique_tests(ts)
+ unq = get_unique_tests(suite)
if options.sub_test:
if run_individual_tests:
@@ -3455,16 +3737,16 @@
return
discards = []
- if options.load_tests:
- ts.load_from_file(options.load_tests)
- elif options.only_failed:
- ts.get_last_failed()
+
+ if options.only_failed:
+ suite.get_last_failed()
+ elif options.load_tests:
+ suite.load_from_file(options.load_tests)
+ elif options.test_only:
+ last_run = os.path.join(options.outdir, "sanitycheck.csv")
+ suite.load_from_file(last_run)
else:
- discards = ts.apply_filters()
-
-
- if options.discard_report:
- ts.discard_report(options.discard_report)
+ discards = suite.apply_filters()
if VERBOSE > 1 and discards:
# if we are using command line platform filter, no need to list every
@@ -3478,55 +3760,29 @@
debug(
"{:<25} {:<50} {}SKIPPED{}: {}".format(
i.platform.name,
- i.test.name,
+ i.testcase.name,
COLOR_YELLOW,
COLOR_NORMAL,
reason))
-
- def native_and_unit_first(a, b):
- if a[0].startswith('unit_testing'):
- return -1
- if b[0].startswith('unit_testing'):
- return 1
- if a[0].startswith('native_posix'):
- return -1
- if b[0].startswith('native_posix'):
- return 1
- if a[0].split("/",1)[0].endswith("_bsim"):
- return -1
- if b[0].split("/",1)[0].endswith("_bsim"):
- return 1
-
- return (a > b) - (a < b)
-
- ts.instances = OrderedDict(sorted(ts.instances.items(),
- key=cmp_to_key(native_and_unit_first)))
-
-
-
if options.report_excluded:
- all_tests = set(get_unique_tests(ts))
+ all_tests = set(get_unique_tests(suite))
to_be_run = set()
- for i,p in ts.instances.items():
- to_be_run.update(p.test.cases)
+ for i,p in suite.instances.items():
+ to_be_run.update(p.testcase.cases)
- if (all_tests - to_be_run):
+ if all_tests - to_be_run:
print("Tests that never build or run:")
- for not_run in (all_tests - to_be_run):
+ for not_run in all_tests - to_be_run:
print("- {}".format(not_run))
return
-
- if options.save_tests:
- ts.save_tests(options.save_tests)
- return
-
if options.subset:
-
+ #suite.instances = OrderedDict(sorted(suite.instances.items(),
+ # key=cmp_to_key(native_and_unit_first)))
subset, sets = options.subset.split("/")
- total = len(ts.instances)
+ total = len(suite.instances)
per_set = round(total / int(sets))
start = (int(subset) - 1) * per_set
if subset == sets:
@@ -3534,75 +3790,54 @@
else:
end = start + per_set
- sliced_instances = islice(ts.instances.items(), start, end)
- ts.instances = OrderedDict(sliced_instances)
+ sliced_instances = islice(suite.instances.items(), start, end)
+ suite.instances = OrderedDict(sliced_instances)
- info("%d tests selected, %d tests discarded due to filters" %
- (len(ts.instances), len(discards)))
- if options.dry_run:
+ if options.save_tests:
+ suite.csv_report(options.save_tests)
return
- if VERBOSE or not TERMINAL:
- goals = ts.execute(
- chatty_test_cb,
- ts.instances)
- else:
- goals = ts.execute(
- terse_test_cb,
- ts.instances)
+ info("%d test configurations selected, %d configurations discarded due to filters." %
+ (len(suite.instances), len(discards)))
+
+ if options.dry_run:
+ duration = time.time() - start_time
+ info("Completed in %d seconds" % (duration))
+ return
+
+ retries = options.retry_failed + 1
+ completed = 0
+
+ suite.update()
+ suite.start_time = start_time
+
+ while True:
+ completed += 1
+
+ if completed > 1:
+ info("%d Iteration:" %(completed ))
+ time.sleep(60) # waiting for the system to settle down
+ suite.total_done = suite.total_tests - suite.total_failed
+ suite.total_failed = 0
+
+ suite.execute()
info("", False)
- if options.detailed_report:
- ts.testcase_target_report(options.detailed_report)
+ retries = retries - 1
+ if retries == 0 or suite.total_failed == 0:
+ break
- # figure out which report to use for size comparison
- if options.compare_report:
- report_to_use = options.compare_report
- elif options.last_metrics:
- report_to_use = LAST_SANITY
- else:
- report_to_use = RELEASE_DATA
-
- deltas = ts.compare_metrics(report_to_use)
- warnings = 0
- if deltas and options.show_footprint:
- for i, metric, value, delta, lower_better in deltas:
- if not options.all_deltas and ((delta < 0 and lower_better) or
- (delta > 0 and not lower_better)):
- continue
-
- percentage = (float(delta) / float(value - delta))
- if not options.all_deltas and (percentage <
- (options.footprint_threshold / 100.0)):
- continue
-
- info("{:<25} {:<60} {}{}{}: {} {:<+4}, is now {:6} {:+.2%}".format(
- i.platform.name, i.test.name, COLOR_YELLOW,
- "INFO" if options.all_deltas else "WARNING", COLOR_NORMAL,
- metric, delta, value, percentage))
- warnings += 1
-
- if warnings:
- info("Deltas based on metrics from last %s" %
- ("release" if not options.last_metrics else "run"))
-
- failed = 0
- for name, goal in goals.items():
- if goal.failed:
- failed += 1
- elif goal.metrics.get("unrecognized") and not options.disable_unrecognized_section_test:
- info("%sFAILED%s: %s has unrecognized binary sections: %s" %
- (COLOR_RED, COLOR_NORMAL, goal.name,
- str(goal.metrics["unrecognized"])))
- failed += 1
+ suite.misc_reports(options.compare_report, options.show_footprint,
+ options.all_deltas, options.footprint_threshold, options.last_metrics)
+ suite.save_reports()
if options.coverage:
- if options.gcov_tool == None:
+ if options.gcov_tool is None:
use_system_gcov = False
for plat in options.coverage_platform:
- ts_plat = ts.get_platform(plat)
+ ts_plat = suite.get_platform(plat)
if ts_plat and (ts_plat.type in {"native", "unit"}):
use_system_gcov = True
@@ -3615,24 +3850,17 @@
info("Generating coverage files...")
generate_coverage(options.outdir, ["*generated*", "tests/*", "samples/*"])
- duration = time.time() - start_time
- info("%s%d of %d%s tests passed with %s%d%s warnings in %d seconds" %
- (COLOR_RED if failed else COLOR_GREEN, len(goals) - failed,
- len(goals), COLOR_NORMAL, COLOR_YELLOW if warnings else COLOR_NORMAL,
- warnings, COLOR_NORMAL, duration))
+ suite.duration = time.time() - start_time
+ suite.summary(options.disable_unrecognized_section_test)
- if options.testcase_report:
- ts.testcase_report(options.testcase_report)
- if not options.no_update:
- ts.testcase_xunit_report(LAST_SANITY_XUNIT, duration)
- ts.testcase_report(LAST_SANITY)
- if options.release:
- ts.testcase_report(RELEASE_DATA)
- if log_file:
- log_file.close()
- if failed or (warnings and options.warnings_as_errors):
+ if options.device_testing:
+ print("\nHardware distribution summary:\n")
+ for p in suite.connected_hardware:
+ if p['connected']:
+ print("%s (%s): %d" %(p['platform'], p.get('id', None), p['counter']))
+
+ if suite.total_failed or (suite.warnings and options.warnings_as_errors):
sys.exit(1)
-
if __name__ == "__main__":
main()
