twister: isolate testplan class

Move testplan class into own file and rename the original twisterlib.

Signed-off-by: Anas Nashif <anas.nashif@intel.com>
This commit is contained in:
Anas Nashif 2022-06-09 14:48:11 -04:00
commit d43e03e19e
5 changed files with 114 additions and 208 deletions

View file

@ -62,7 +62,7 @@ class Handler:
self.binary = None
self.pid_fn = None
self.call_make_run = False
self.call_make_run = True
self.name = instance.name
self.instance = instance

View file

@ -11,10 +11,9 @@ import subprocess
import pickle
import logging
from colorama import Fore
from multiprocessing import Lock, Process, Value
from multiprocessing import Lock, Value
from twister.cmakecache import CMakeCache
from twister.handlers import BinaryHandler, QEMUHandler, DeviceHandler
logger = logging.getLogger('twister')
logger.setLevel(logging.DEBUG)
@ -32,11 +31,8 @@ class ExecutionCounter(object):
self._failed = Value('i', 0)
self._total = Value('i', total)
self._cases = Value('i', 0)
self.lock = Lock()
def summary(self):
logger.debug("--------------------------------")
logger.debug(f"Total Test suites: {self.total}")
@ -220,7 +216,7 @@ class CMake:
if log_msg:
overflow_found = re.findall("region `(FLASH|ROM|RAM|ICCM|DCCM|SRAM)' overflowed by", log_msg)
if overflow_found and not self.overflow_as_errors:
if overflow_found and not self.options.overflow_as_errors:
logger.debug("Test skipped due to {} Overflow".format(overflow_found[0]))
self.instance.status = "skipped"
self.instance.reason = "{} overflow".format(overflow_found[0])
@ -237,7 +233,7 @@ class CMake:
def run_cmake(self, args=[]):
if self.warnings_as_errors:
if not self.options.disable_warnings_as_errors:
ldflags = "-Wl,--fatal-warnings"
cflags = "-Werror"
aflags = "-Werror -Wa,--fatal-warnings"
@ -376,31 +372,25 @@ class FilterBuilder(CMake):
class ProjectBuilder(FilterBuilder):
def __init__(self, tplan, instance, **kwargs):
def __init__(self, tplan, instance, env, **kwargs):
super().__init__(instance.testsuite, instance.platform, instance.testsuite.source_dir, instance.build_dir)
self.log = "build.log"
self.instance = instance
self.testplan = tplan
self.filtered_tests = 0
self.options = env.options
self.lsan = kwargs.get('lsan', False)
self.asan = kwargs.get('asan', False)
self.ubsan = kwargs.get('ubsan', False)
self.valgrind = kwargs.get('valgrind', False)
self.extra_args = kwargs.get('extra_args', [])
self.device_testing = kwargs.get('device_testing', False)
self.cmake_only = kwargs.get('cmake_only', False)
self.cleanup = kwargs.get('cleanup', False)
self.coverage = kwargs.get('coverage', False)
self.inline_logs = kwargs.get('inline_logs', False)
self.generator = kwargs.get('generator', None)
self.generator_cmd = kwargs.get('generator_cmd', None)
self.verbose = kwargs.get('verbose', None)
self.warnings_as_errors = kwargs.get('warnings_as_errors', True)
self.overflow_as_errors = kwargs.get('overflow_as_errors', False)
self.suite_name_check = kwargs.get('suite_name_check', True)
self.seed = kwargs.get('seed', 0)
if self.options.ninja:
self.generator_cmd = "ninja"
self.generator = "Ninja"
else:
self.generator_cmd = "make"
self.generator = "Unix Makefiles"
@staticmethod
def log_info(filename, inline_logs):
@ -436,76 +426,18 @@ class ProjectBuilder(FilterBuilder):
else:
self.log_info("{}".format(b_log), inline_logs)
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.testsuite.type == "unit":
instance.handler = BinaryHandler(instance, "unit")
instance.handler.binary = os.path.join(instance.build_dir, "testbinary")
if self.coverage:
args.append("COVERAGE=1")
elif instance.platform.type == "native":
handler = BinaryHandler(instance, "native")
handler.asan = self.asan
handler.valgrind = self.valgrind
handler.lsan = self.lsan
handler.ubsan = self.ubsan
handler.coverage = self.coverage
handler.binary = os.path.join(instance.build_dir, "zephyr", "zephyr.exe")
instance.handler = handler
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 instance.platform.simulation == "tsim":
instance.handler = BinaryHandler(instance, "tsim")
instance.handler.call_make_run = True
elif self.device_testing:
instance.handler = DeviceHandler(instance, "device")
instance.handler.coverage = self.coverage
elif instance.platform.simulation == "nsim":
if find_executable("nsimdrv"):
instance.handler = BinaryHandler(instance, "nsim")
instance.handler.call_make_run = True
elif instance.platform.simulation == "mdb-nsim":
if find_executable("mdb"):
instance.handler = BinaryHandler(instance, "nsim")
instance.handler.call_make_run = True
elif instance.platform.simulation == "armfvp":
instance.handler = BinaryHandler(instance, "armfvp")
instance.handler.call_make_run = True
elif instance.platform.simulation == "xt-sim":
instance.handler = BinaryHandler(instance, "xt-sim")
instance.handler.call_make_run = True
if instance.handler:
instance.handler.args = args
instance.handler.generator_cmd = self.generator_cmd
instance.handler.generator = self.generator
instance.handler.suite_name_check = self.suite_name_check
def process(self, pipeline, done, message, lock, results):
op = message.get('op')
if not self.instance.handler:
self.setup_handler()
self.instance.setup_handler(self.options)
# The build process, call cmake and build with configured generator
if op == "cmake":
res = self.cmake()
if self.instance.status in ["failed", "error"]:
pipeline.put({"op": "report", "test": self.instance})
elif self.cmake_only:
elif self.options.cmake_only:
if self.instance.status is None:
self.instance.status = "passed"
pipeline.put({"op": "report", "test": self.instance})
@ -571,14 +503,14 @@ class ProjectBuilder(FilterBuilder):
done.put(self.instance)
self.report_out(results)
if self.cleanup and not self.coverage and self.instance.status == "passed":
if self.options.runtime_artifact_cleanup and not self.options.coverage and self.instance.status == "passed":
pipeline.put({
"op": "cleanup",
"test": self.instance
})
elif op == "cleanup":
if self.device_testing:
if self.options.device_testing:
self.cleanup_device_testing_artifacts()
else:
self.cleanup_artifacts()
@ -666,7 +598,7 @@ class ProjectBuilder(FilterBuilder):
Fore.RESET,
instance.reason))
if not self.verbose:
self.log_info_file(self.inline_logs)
self.log_info_file(self.options.inline_logs)
elif instance.status in ["skipped", "filtered"]:
status = Fore.YELLOW + "SKIPPED" + Fore.RESET
results.skipped_configs += 1
@ -705,7 +637,7 @@ class ProjectBuilder(FilterBuilder):
instance.testsuite.name, status, more_info))
if instance.status in ["error", "failed", "timeout"]:
self.log_info_file(self.inline_logs)
self.log_info_file(self.options.inline_logs)
else:
completed_perc = 0
if total_to_do > 0:
@ -787,7 +719,7 @@ class ProjectBuilder(FilterBuilder):
sys.stdout.flush()
def gather_metrics(self, instance):
if self.testplan.enable_size_report and not self.testplan.cmake_only:
if self.options.enable_size_report and not self.options.cmake_only:
self.calc_one_elf_size(instance)
else:
instance.metrics["ram_size"] = 0

View file

@ -6,9 +6,15 @@
import os
import hashlib
import random
import logging
from twister.testsuite import TestCase, TestSuite
from twister.handlers import BinaryHandler, QEMUHandler, DeviceHandler
from distutils.spawn import find_executable
logger = logging.getLogger('twister')
logger.setLevel(logging.DEBUG)
class TestInstance:
"""Class representing the execution of a particular TestSuite on a platform
@ -122,6 +128,63 @@ class TestInstance:
return can_run
def setup_handler(self, options):
if self.handler:
return
args = []
handler = None
if self.platform.simulation == "qemu":
handler = QEMUHandler(self, "qemu")
args.append(f"QEMU_PIPE={handler.get_fifo()}")
elif self.testsuite.type == "unit":
handler = BinaryHandler(self, "unit")
handler.binary = os.path.join(self.build_dir, "testbinary")
if options.enable_coverage:
args.append("COVERAGE=1")
handler.call_make_run = False
elif self.platform.type == "native":
handler = BinaryHandler(self, "native")
handler.asan = options.enable_asan
handler.valgrind = options.enable_valgrind
handler.lsan = options.enable_lsan
handler.ubsan = options.enable_ubsan
handler.coverage = options.enable_coverage
handler.call_make_run = False
handler.binary = os.path.join(self.build_dir, "zephyr", "zephyr.exe")
elif self.platform.simulation == "renode":
if find_executable("renode"):
handler = BinaryHandler(self, "renode")
handler.pid_fn = os.path.join(self.build_dir, "renode.pid")
elif self.platform.simulation == "tsim":
handler = BinaryHandler(self, "tsim")
elif options.device_testing:
handler = DeviceHandler(self, "device")
handler.coverage = self.enable_coverage
handler.call_make_run = False
elif self.platform.simulation == "nsim":
if find_executable("nsimdrv"):
handler = BinaryHandler(self, "nsim")
elif self.platform.simulation == "mdb-nsim":
if find_executable("mdb"):
handler = BinaryHandler(self, "nsim")
elif self.platform.simulation == "armfvp":
handler = BinaryHandler(self, "armfvp")
elif self.platform.simulation == "xt-sim":
handler = BinaryHandler(self, "xt-sim")
if handler:
handler.args = args
handler.suite_name_check = not options.disable_suite_name_check
if options.ninja:
handler.generator_cmd = "ninja"
handler.generator = "Ninja"
else:
handler.generator_cmd = "make"
handler.generator = "Unix Makefiles"
self.handler = handler
# Global testsuite parameters
def check_runnable(self, enable_slow=False, filter='buildable', fixtures=[]):

View file

@ -10,29 +10,21 @@ import sys
import re
import subprocess
import shutil
import queue
import glob
import logging
from distutils.spawn import find_executable
import colorama
import json
from multiprocessing import Lock, Process, Value
from typing import List
from twister.cmakecache import CMakeCache
from twister.testsuite import TestCase, TestSuite
from twister.error import TwisterRuntimeError, ConfigurationError, BuildError
from twister.handlers import BinaryHandler, QEMUHandler, DeviceHandler
from twister.testsuite import TestSuite
from twister.error import TwisterRuntimeError
from twister.platform import Platform
from twister.config_parser import TwisterConfigParser
from twister.size_calc import SizeCalculator
from twister.testinstance import TestInstance
from twister.runner import ExecutionCounter, ProjectBuilder
from twister.runner import ProjectBuilder
try:
# Use the C LibYAML parser if available, rather than the Python parser.
@ -168,29 +160,8 @@ class TestPlan:
else:
self.board_roots = board_root_list
self.options = env.options
# Test Plan Options
self.coverage_platform = []
self.build_only = False
self.cmake_only = False
self.cleanup = False
self.enable_slow = False
self.device_testing = False
self.fixtures = []
self.enable_coverage = False
self.enable_ubsan = False
self.enable_lsan = False
self.enable_asan = False
self.enable_valgrind = False
self.extra_args = []
self.inline_logs = False
self.west_flash = None
self.west_runner = None
self.generator = None
self.generator_cmd = None
self.warnings_as_errors = True
self.overflow_as_errors = False
self.quarantine_verify = False
self.retry_build_errors = False
self.suite_name_check = True
self.seed = 0
@ -205,16 +176,10 @@ class TestPlan:
self.outdir = os.path.abspath(outdir)
self.load_errors = 0
self.instances = dict()
self.start_time = 0
self.warnings = 0
# hardcoded for now
self.duts = []
# run integration tests only
self.integration = False
# used during creating shorter build paths
self.link_dir_counter = 0
@ -687,14 +652,14 @@ class TestPlan:
if ts.get("run_id"):
instance.run_id = ts.get("run_id")
if self.device_testing:
if self.options.device_testing:
tfilter = 'runnable'
else:
tfilter = 'buildable'
instance.run = instance.check_runnable(
self.enable_slow,
self.options.enable_slow,
tfilter,
self.fixtures
self.options.fixture
)
instance.metrics['handler_time'] = ts.get('execution_time', 0)
@ -708,7 +673,7 @@ class TestPlan:
instance.reason = None
# test marked as passed (built only) but can run when
# --test-only is used. Reset status to capture new results.
elif status == 'passed' and instance.run and self.test_only:
elif status == 'passed' and instance.run and self.options.test_only:
instance.status = None
instance.reason = None
else:
@ -730,7 +695,7 @@ class TestPlan:
case.output = tc.get('log')
instance.create_overlay(platform, self.enable_asan, self.enable_ubsan, self.enable_coverage, self.coverage_platform)
instance.create_overlay(platform, self.options.enable_asan, self.options.enable_ubsan, self.options.enable_coverage, self.options.coverage_platform)
instance_list.append(instance)
self.add_instances(instance_list)
@ -788,7 +753,7 @@ class TestPlan:
if ts.build_on_all and not platform_filter:
platform_scope = self.platforms
elif ts.integration_platforms and self.integration:
elif ts.integration_platforms and self.options.integration:
self.verify_platforms_existence(
ts.integration_platforms, f"{ts_name} - integration_platforms")
platform_scope = list(filter(lambda item: item.name in ts.integration_platforms, \
@ -796,7 +761,7 @@ class TestPlan:
else:
platform_scope = platforms
integration = self.integration and ts.integration_platforms
integration = self.options.integration and ts.integration_platforms
# If there isn't any overlap between the platform_allow list and the platform_scope
# we set the scope to the platform_allow list
@ -820,9 +785,9 @@ class TestPlan:
tfilter = 'buildable'
instance.run = instance.check_runnable(
self.enable_slow,
self.options.enable_slow,
tfilter,
self.fixtures
self.options.fixture
)
if runnable and self.duts:
for h in self.duts:
@ -844,7 +809,7 @@ class TestPlan:
if runnable and not instance.run:
instance.add_filter("Not runnable on device", Filters.PLATFORM)
if self.integration and ts.integration_platforms and plat.name not in ts.integration_platforms:
if self.options.integration and ts.integration_platforms and plat.name not in ts.integration_platforms:
instance.add_filter("Not part of integration platforms", Filters.TESTSUITE)
if ts.skip:
@ -917,10 +882,10 @@ class TestPlan:
test_configuration = ".".join([instance.platform.name,
instance.testsuite.id])
# skip quarantined tests
if test_configuration in self.quarantine and not self.quarantine_verify:
if test_configuration in self.quarantine and not self.options.quarantine_verify:
instance.add_filter(f"Quarantine: {self.quarantine[test_configuration]}", Filters.QUARENTINE)
# run only quarantined test to verify their statuses (skip everything else)
if self.quarantine_verify and test_configuration not in self.quarantine:
if self.options.quarantine_verify and test_configuration not in self.quarantine:
instance.add_filter("Not under quarantine", Filters.QUARENTINE)
# if nothing stopped us until now, it means this configuration
@ -958,14 +923,14 @@ class TestPlan:
self.add_instances(instance_list)
for _, case in self.instances.items():
case.create_overlay(case.platform, self.enable_asan, self.enable_ubsan, self.enable_coverage, self.coverage_platform)
case.create_overlay(case.platform, self.options.enable_asan, self.options.enable_ubsan, self.options.enable_coverage, self.options.coverage_platform)
self.selected_platforms = set(p.platform.name for p in self.instances.values())
filtered_instances = list(filter(lambda item: item.status == "filtered", self.instances.values()))
for filtered_instance in filtered_instances:
# If integration mode is on all skips on integration_platforms are treated as errors.
if self.integration and filtered_instance.platform.name in filtered_instance.testsuite.integration_platforms \
if self.options.integration and filtered_instance.platform.name in filtered_instance.testsuite.integration_platforms \
and "Quarantine" not in filtered_instance.reason:
# Do not treat this as error if filter type is command line
filters = {t['type'] for t in filtered_instance.filters}
@ -1008,27 +973,8 @@ class TestPlan:
except queue.Empty:
break
else:
test = task['test']
pb = ProjectBuilder(self,
test,
lsan=self.enable_lsan,
asan=self.enable_asan,
ubsan=self.enable_ubsan,
coverage=self.enable_coverage,
extra_args=self.extra_args,
device_testing=self.device_testing,
cmake_only=self.cmake_only,
cleanup=self.cleanup,
valgrind=self.enable_valgrind,
inline_logs=self.inline_logs,
generator=self.generator,
generator_cmd=self.generator_cmd,
verbose=self.verbose,
warnings_as_errors=self.warnings_as_errors,
overflow_as_errors=self.overflow_as_errors,
suite_name_check=self.suite_name_check,
seed=self.seed
)
instance = task['test']
pb = ProjectBuilder(self, instance, self.env)
pb.process(pipeline, done_queue, task, lock, results)
return True
@ -1036,8 +982,8 @@ class TestPlan:
def execute(self, pipeline, done, results):
lock = Lock()
logger.info("Adding tasks to the queue...")
self.add_tasks_to_queue(pipeline, self.build_only, self.test_only,
retry_build_errors=self.retry_build_errors)
self.add_tasks_to_queue(pipeline, self.options.build_only, self.options.test_only,
retry_build_errors=self.options.retry_build_errors)
logger.info("Added initial list of jobs to queue")
processes = []
@ -1119,8 +1065,3 @@ class TestPlan:
instance.build_dir = link_path
self.link_dir_counter += 1
def init(colorama_strip):
colorama.init(strip=colorama_strip)

View file

@ -205,8 +205,7 @@ except ImportError:
sys.path.insert(0, os.path.join(ZEPHYR_BASE, "scripts/pylib/twister"))
import twisterlib
from twisterlib import TestPlan
from twister.testplan import TestPlan
from twister.runner import ExecutionCounter
from twister.enviornment import TwisterEnv, canonical_zephyr_base
from twister.reports import Reporting
@ -844,6 +843,12 @@ def setup_logging(outdir, log_file, verbose, timestamps):
logger.addHandler(fh)
def init_color(colorama_strip):
colorama.init(strip=colorama_strip)
def main():
start_time = time.time()
@ -853,7 +858,7 @@ def main():
color_strip = False if options.force_color else None
colorama.init(strip=color_strip)
twisterlib.init(colorama_strip=color_strip)
init_color(colorama_strip=color_strip)
previous_results = None
# Cleanup
@ -920,34 +925,6 @@ def main():
env.discover()
tplan = TestPlan(options.board_root, options.testsuite_root, env, options.outdir)
# Set testplan options from command line.
tplan.build_only = options.build_only
tplan.cmake_only = options.cmake_only
tplan.cleanup = options.runtime_artifact_cleanup
tplan.test_only = options.test_only
tplan.retry_build_errors = options.retry_build_errors
tplan.enable_slow = options.enable_slow
tplan.device_testing = options.device_testing
tplan.fixtures = options.fixture
tplan.enable_asan = options.enable_asan
tplan.enable_lsan = options.enable_lsan
tplan.enable_ubsan = options.enable_ubsan
tplan.enable_coverage = options.enable_coverage
tplan.enable_valgrind = options.enable_valgrind
tplan.coverage_platform = options.coverage_platform
tplan.inline_logs = options.inline_logs
tplan.enable_size_report = options.enable_size_report
tplan.extra_args = options.extra_args
tplan.west_flash = options.west_flash
tplan.west_runner = options.west_runner
tplan.verbose = VERBOSE
tplan.warnings_as_errors = not options.disable_warnings_as_errors
tplan.integration = options.integration
tplan.overflow_as_errors = options.overflow_as_errors
tplan.suite_name_check = not options.disable_suite_name_check
tplan.seed = options.seed
# get all enabled west projects
west_proj = west_projects()
modules_meta = parse_modules(ZEPHYR_BASE,
@ -956,13 +933,6 @@ def main():
modules = [module.meta.get('name') for module in modules_meta]
tplan.modules = modules
if options.ninja:
tplan.generator_cmd = "ninja"
tplan.generator = "Ninja"
else:
tplan.generator_cmd = "make"
tplan.generator = "Unix Makefiles"
# Set number of jobs
if options.jobs:
tplan.jobs = options.jobs
@ -1015,9 +985,9 @@ def main():
only one platform is allowed""")
# the fixtures given by twister command explicitly should be assigned to each DUTs
if tplan.fixtures:
if options.fixture:
for d in tplan.duts:
d.fixtures.extend(tplan.fixtures)
d.fixtures.extend(options.fixture)
if tplan.load_errors:
sys.exit(1)