51     def __init__(self, total=0):  argument
63         self._done = Value('i', 0)
66         self._iteration = Value('i', 0)
70         self._passed = Value('i', 0)
74         self._skipped_configs = Value('i', 0)
78         self._skipped_runtime = Value('i', 0)
82         self._skipped_filter = Value('i', 0)
85         self._skipped_cases = Value('i', 0)
88         self._error = Value('i', 0)
89         self._failed = Value('i', 0)
92         self._total = Value('i', total)
95         self._cases = Value('i', 0)
96         self.lock = Lock()
98     def summary(self):  argument
100         print(f"Total test suites: {self.total}") # actually test instances
101         print(f"Total test cases: {self.cases}")
102         print(f"Executed test cases: {self.cases - self.skipped_cases}")
103         print(f"Skipped test cases: {self.skipped_cases}")
104         print(f"Completed test suites: {self.done}")
105         print(f"Passing test suites: {self.passed}")
106         print(f"Failing test suites: {self.failed}")
107         print(f"Skipped test suites: {self.skipped_configs}")
108         print(f"Skipped test suites (runtime): {self.skipped_runtime}")
109         print(f"Skipped test suites (filter): {self.skipped_filter}")
110         print(f"Errors: {self.error}")
114     def cases(self):  argument
115         with self._cases.get_lock():
116             return self._cases.value
119     def cases(self, value):  argument
120         with self._cases.get_lock():
121             self._cases.value = value
124     def skipped_cases(self):  argument
125         with self._skipped_cases.get_lock():
126             return self._skipped_cases.value
129     def skipped_cases(self, value):  argument
130         with self._skipped_cases.get_lock():
131             self._skipped_cases.value = value
134     def error(self):  argument
135         with self._error.get_lock():
136             return self._error.value
139     def error(self, value):  argument
140         with self._error.get_lock():
141             self._error.value = value
144     def iteration(self):  argument
145         with self._iteration.get_lock():
146             return self._iteration.value
149     def iteration(self, value):  argument
150         with self._iteration.get_lock():
151             self._iteration.value = value
154     def done(self):  argument
155         with self._done.get_lock():
156             return self._done.value
159     def done(self, value):  argument
160         with self._done.get_lock():
161             self._done.value = value
164     def passed(self):  argument
165         with self._passed.get_lock():
166             return self._passed.value
169     def passed(self, value):  argument
170         with self._passed.get_lock():
171             self._passed.value = value
174     def skipped_configs(self):  argument
175         with self._skipped_configs.get_lock():
176             return self._skipped_configs.value
179     def skipped_configs(self, value):  argument
180         with self._skipped_configs.get_lock():
181             self._skipped_configs.value = value
184     def skipped_filter(self):  argument
185         with self._skipped_filter.get_lock():
186             return self._skipped_filter.value
189     def skipped_filter(self, value):  argument
190         with self._skipped_filter.get_lock():
191             self._skipped_filter.value = value
194     def skipped_runtime(self):  argument
195         with self._skipped_runtime.get_lock():
196             return self._skipped_runtime.value
199     def skipped_runtime(self, value):  argument
200         with self._skipped_runtime.get_lock():
201             self._skipped_runtime.value = value
204     def failed(self):  argument
205         with self._failed.get_lock():
206             return self._failed.value
209     def failed(self, value):  argument
210         with self._failed.get_lock():
211             self._failed.value = value
214     def total(self):  argument
215         with self._total.get_lock():
216             return self._total.value
222     def __init__(self, testsuite, platform, source_dir, build_dir, jobserver):  argument
224         self.cwd = None
225         self.capture_output = True
227         self.defconfig = {}
228         self.cmake_cache = {}
230         self.instance = None
231         self.testsuite = testsuite
232         self.platform = platform
233         self.source_dir = source_dir
234         self.build_dir = build_dir
235         self.log = "build.log"
237         self.default_encoding = sys.getdefaultencoding()
238         self.jobserver = jobserver
240     def parse_generated(self, filter_stages=[]):  argument
241         self.defconfig = {}
244     def run_build(self, args=[]):  argument
246         logger.debug("Building %s for %s" % (self.source_dir, self.platform.name))
254         if self.capture_output:
259         if self.cwd:
260             kwargs['cwd'] = self.cwd
263             p = self.jobserver.popen(cmd, **kwargs)
271             msg = "Finished building %s for %s" % (self.source_dir, self.platform.name)
273             self.instance.status = "passed"
274             if not self.instance.run:
275                 self.instance.add_missing_case_status("skipped", "Test was built only")
276             results = {'msg': msg, "returncode": p.returncode, "instance": self.instance}
279                 log_msg = out.decode(self.default_encoding)
280 …     with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
289                 log_msg = out.decode(self.default_encoding)
290 …     with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
296                 if overflow_found and not self.options.overflow_as_errors:
298                     self.instance.status = "skipped"
299                     self.instance.reason = "{} overflow".format(overflow_found[0])
300                     change_skip_to_error_if_integration(self.options, self.instance)
301                 elif imgtool_overflow_found and not self.options.overflow_as_errors:
302                     self.instance.status = "skipped"
303                     self.instance.reason = "imgtool overflow"
304                     change_skip_to_error_if_integration(self.options, self.instance)
306                     self.instance.status = "error"
307                     self.instance.reason = "Build failure"
311                 "instance": self.instance,
316     def run_cmake(self, args="", filter_stages=[]):  argument
318         if not self.options.disable_warnings_as_errors:
325         logger.debug("Running cmake on %s for %s" % (self.source_dir, self.platform.name))
327             f'-B{self.build_dir}',
328             f'-DTC_RUNID={self.instance.run_id}',
331             f'-G{self.env.generator}'
344         if self.testsuite.sysbuild and not filter_stages:
345             logger.debug("Building %s using sysbuild" % (self.source_dir))
348                 f'-DAPP_DIR={self.source_dir}'
352                 f'-S{self.source_dir}'
358         cmake_opts = ['-DBOARD={}'.format(self.platform.name)]
371         if self.capture_output:
376         if self.cwd:
377             kwargs['cwd'] = self.cwd
380             p = self.jobserver.popen(cmd, **kwargs)
386             filter_results = self.parse_generated(filter_stages)
387             msg = "Finished building %s for %s" % (self.source_dir, self.platform.name)
392             self.instance.status = "error"
393             self.instance.reason = "Cmake build failure"
395             for tc in self.instance.testcases:
396                 tc.status = self.instance.status
398             logger.error("Cmake build failure: %s for %s" % (self.source_dir, self.platform.name))
402             os.makedirs(self.build_dir, exist_ok=True)
403 …     with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
404                 log_msg = out.decode(self.default_encoding)
412     def __init__(self, testsuite, platform, source_dir, build_dir, jobserver):  argument
415         self.log = "config-twister.log"
417     def parse_generated(self, filter_stages=[]):  argument
419         if self.platform.name == "unit_testing":
422         if self.testsuite.sysbuild and not filter_stages:
424             domain_path = os.path.join(self.build_dir, "domains.yaml")
427             self.instance.domains = domains
433             cmake_cache_path = os.path.join(self.build_dir, "CMakeCache.txt")
437                 defconfig_path = os.path.join(self.build_dir, "zephyr", ".config")
439             edt_pickle = os.path.join(self.build_dir, "zephyr", "edt.pickle")
446                     m = self.config_re.match(line)
453             self.defconfig = defconfig
464         self.cmake_cache = cmake_conf
467             "ARCH": self.platform.arch,
468             "PLATFORM": self.platform.name
472             filter_data.update(self.defconfig)
473         filter_data.update(self.cmake_cache)
475         if self.testsuite.sysbuild and self.env.options.device_testing:
481             if self.env.options.west_flash is None:
484                 return {os.path.join(self.platform.name, self.testsuite.name): True}
485             elif "--erase" in self.env.options.west_flash:
488                 return {os.path.join(self.platform.name, self.testsuite.name): True}
490         if self.testsuite and self.testsuite.filter:
497                 res = expr_parser.parse(self.testsuite.filter, filter_data, edt)
501                     "Failed processing %s\n" % self.testsuite.yamlfile)
505                 return {os.path.join(self.platform.name, self.testsuite.name): True}
507                 return {os.path.join(self.platform.name, self.testsuite.name): False}
509             self.platform.filter_data = filter_data
515     def __init__(self, instance, env, jobserver, **kwargs):  argument
518         self.log = "build.log"
519         self.instance = instance
520         self.filtered_tests = 0
521         self.options = env.options
522         self.env = env
523         self.duts = None
543     def log_info_file(self, inline_logs):  argument
544         build_dir = self.instance.build_dir
550         if os.path.exists(v_log) and "Valgrind" in self.instance.reason:
551             self.log_info("{}".format(v_log), inline_logs)
553             self.log_info("{}".format(h_log), inline_logs)
555             self.log_info("{}".format(d_log), inline_logs)
557             self.log_info("{}".format(b_log), inline_logs)
560     def process(self, pipeline, done, message, lock, results):  argument
563         self.instance.setup_handler(self.env)
566             res = self.cmake(filter_stages=self.instance.filter_stages)
567             if self.instance.status in ["failed", "error"]:
568                 pipeline.put({"op": "report", "test": self.instance})
571                 if self.instance.name in res['filter'] and res['filter'][self.instance.name]:
572                     logger.debug("filtering %s" % self.instance.name)
573                     self.instance.status = "filtered"
574                     self.instance.reason = "runtime filter"
576                     self.instance.add_missing_case_status("skipped")
577                     pipeline.put({"op": "report", "test": self.instance})
579                     pipeline.put({"op": "cmake", "test": self.instance})
583             res = self.cmake()
584             if self.instance.status in ["failed", "error"]:
585                 pipeline.put({"op": "report", "test": self.instance})
586             elif self.options.cmake_only:
587                 if self.instance.status is None:
588                     self.instance.status = "passed"
589                 pipeline.put({"op": "report", "test": self.instance})
592                 if self.instance.name in res['filter'] and res['filter'][self.instance.name]:
593                     logger.debug("filtering %s" % self.instance.name)
594                     self.instance.status = "filtered"
595                     self.instance.reason = "runtime filter"
597                     self.instance.add_missing_case_status("skipped")
598                     pipeline.put({"op": "report", "test": self.instance})
600                     pipeline.put({"op": "build", "test": self.instance})
603             logger.debug("build test: %s" % self.instance.name)
604             res = self.build()
606                 self.instance.status = "error"
607                 self.instance.reason = "Build Failure"
608                 pipeline.put({"op": "report", "test": self.instance})
612                 if  self.instance.status == "skipped":
614                     self.instance.add_missing_case_status("skipped", self.instance.reason)
617                     self.instance.add_missing_case_status("blocked", self.instance.reason)
618                     pipeline.put({"op": "report", "test": self.instance})
620                     logger.debug(f"Determine test cases for test instance: {self.instance.name}")
621                     self.determine_testcases(results)
622                     pipeline.put({"op": "gather_metrics", "test": self.instance})
625             self.gather_metrics(self.instance)
626             if self.instance.run and self.instance.handler.ready:
627                 pipeline.put({"op": "run", "test": self.instance})
629                 pipeline.put({"op": "report", "test": self.instance})
633             logger.debug("run test: %s" % self.instance.name)
634             self.run()
635             logger.debug(f"run status: {self.instance.name} {self.instance.status}")
638                 self.instance.handler.thread = None
639                 self.instance.handler.duts = None
642                     "test": self.instance,
643                     "status": self.instance.status,
644                     "reason": self.instance.reason
654                 done.put(self.instance)
655                 self.report_out(results)
657             if not self.options.coverage:
658                 if self.options.prep_artifacts_for_testing:
659                     pipeline.put({"op": "cleanup", "mode": "device", "test": self.instance})
660 …         elif self.options.runtime_artifact_cleanup == "pass" and self.instance.status == "passed":
661                     pipeline.put({"op": "cleanup", "mode": "passed", "test": self.instance})
662                 elif self.options.runtime_artifact_cleanup == "all":
663                     pipeline.put({"op": "cleanup", "mode": "all", "test": self.instance})
668                 self.cleanup_device_testing_artifacts()
669 …        elif mode == "passed" or (mode == "all" and self.instance.reason != "Cmake build failure"):
670                 self.cleanup_artifacts()
672     def determine_testcases(self, results):  argument
673         yaml_testsuite_name = self.instance.testsuite.id
676         elf = ELFFile(open(self.instance.get_elf_file(), "rb"))
678 …logger.debug(f"Test instance {self.instance.name} already has {len(self.instance.testcases)} cases…
698             self.instance.testcases.clear()
699             self.instance.testsuite.testcases.clear()
706                 self.instance.add_testcase(name=testcase_id)
707                 self.instance.testsuite.add_testcase(name=testcase_id)
710     def cleanup_artifacts(self, additional_keep=[]):  argument
711         logger.debug("Cleaning up {}".format(self.instance.build_dir))
727         if self.options.runtime_artifact_cleanup == 'all':
730         allow = [os.path.join(self.instance.build_dir, file) for file in allow]
732         for dirpath, dirnames, filenames in os.walk(self.instance.build_dir, topdown=False):
745     def cleanup_device_testing_artifacts(self):  argument
746         logger.debug("Cleaning up for Device Testing {}".format(self.instance.build_dir))
748         files_to_keep = self._get_binaries()
751         self.cleanup_artifacts(files_to_keep)
753         self._sanitize_files()
755     def _get_binaries(self) -> List[str]:  argument
758         self.instance.build_dir), basing on information from platform.binaries
764         platform = self.instance.platform
769         binaries += self._get_binaries_from_runners()
781     def _get_binaries_from_runners(self) -> List[str]:  argument
784         self.instance.build_dir) from runners.yaml file.
786         runners_file_path: str = os.path.join(self.instance.build_dir, 'zephyr', 'runners.yaml')
811     def _sanitize_files(self):  argument
816         self._sanitize_runners_file()
817         self._sanitize_zephyr_base_from_files()
819     def _sanitize_runners_file(self):  argument
822         directory for those files is f"{self.instance.build_dir}/zephyr"
824         runners_dir_path: str = os.path.join(self.instance.build_dir, 'zephyr')
850     def _sanitize_zephyr_base_from_files(self):  argument
859             file_path = os.path.join(self.instance.build_dir, file_path)
873     def report_out(self, results):  argument
877         instance = self.instance
888             if self.options.verbose:
899             if not self.options.verbose:
900                 self.log_info_file(self.options.inline_logs)
917         if self.options.verbose:
918             if self.options.cmake_only:
932                      and hasattr(self.instance.handler, 'seed')
933                      and self.instance.handler.seed is not None ):
934                     more_info += "/seed: " + str(self.options.seed)
940                 self.log_info_file(self.options.inline_logs)
996     def cmake(self, filter_stages=[]):  argument
997         args = self.cmake_assemble_args(
998             self.testsuite.extra_args.copy(), # extra_args from YAML
999             self.instance.handler,
1000             self.testsuite.extra_conf_files,
1001             self.testsuite.extra_overlay_confs,
1002             self.testsuite.extra_dtc_overlay_files,
1003             self.options.extra_args, # CMake extra args
1004             self.instance.build_dir,
1007         res = self.run_cmake(args,filter_stages)
1010     def build(self):  argument
1011         res = self.run_build(['--build', self.build_dir])
1014     def run(self):  argument
1016         instance = self.instance
1020                 instance.handler.duts = self.duts
1022             if(self.options.seed is not None and instance.platform.name.startswith("native_posix")):
1023                 self.parse_generated()
1024                 if('CONFIG_FAKE_ENTROPY_NATIVE_POSIX' in self.defconfig and
1025                     self.defconfig['CONFIG_FAKE_ENTROPY_NATIVE_POSIX'] == 'y'):
1026                     instance.handler.seed = self.options.seed
1028             if self.options.extra_test_args and instance.platform.arch == "posix":
1029                 instance.handler.extra_test_args = self.options.extra_test_args
1040     def gather_metrics(self, instance: TestInstance):  argument
1041         if self.options.enable_size_report and not self.options.cmake_only:
1042             self.calc_size(instance=instance, from_buildlog=self.options.footprint_from_buildlog)
1071     def __init__(self, instances, suites, env=None) -> None:  argument
1072         self.pipeline = None
1073         self.options = env.options
1074         self.env = env
1075         self.instances = instances
1076         self.suites = suites
1077         self.duts = None
1078         self.jobs = 1
1079         self.results = None
1080         self.jobserver = None
1082     def run(self):  argument
1084         retries = self.options.retry_failed + 1
1090         self.results = ExecutionCounter(total=len(self.instances))
1091         self.iteration = 0
1096         if self.options.jobs:
1097             self.jobs = self.options.jobs
1098         elif self.options.build_only:
1099             self.jobs = multiprocessing.cpu_count() * 2
1101             self.jobs = multiprocessing.cpu_count()
1105                 self.jobserver = GNUMakeJobClient.from_environ(jobs=self.options.jobs)
1106                 if not self.jobserver:
1107                     self.jobserver = GNUMakeJobServer(self.jobs)
1108                 elif self.jobserver.jobs:
1109                     self.jobs = self.jobserver.jobs
1112                 self.jobserver = JobClient()
1114             logger.info("JOBS: %d", self.jobs)
1116         self.update_counting_before_pipeline()
1119             self.results.iteration += 1
1121             if self.results.iteration > 1:
1122                 logger.info("%d Iteration:" % (self.results.iteration))
1123                 time.sleep(self.options.retry_interval)  # waiting for the system to settle down
1124                 self.results.done = self.results.total - self.results.failed - self.results.error
1125                 self.results.failed = 0
1126                 if self.options.retry_build_errors:
1127                     self.results.error = 0
1129                 self.results.done = self.results.skipped_filter
1131             self.execute(pipeline, done_queue)
1139                     inst.metrics.update(self.instances[inst.name].metrics)
1142                     self.instances[inst.name] = inst
1147             if self.results.error and self.options.retry_build_errors:
1151             if retries == 0 or ( self.results.failed == 0 and not retry_errors):
1154         self.show_brief()
1156     def update_counting_before_pipeline(self):  argument
1162         for instance in self.instances.values():
1164                 self.results.skipped_filter += 1
1165                 self.results.skipped_configs += 1
1166                 self.results.skipped_cases += len(instance.testsuite.testcases)
1167                 self.results.cases += len(instance.testsuite.testcases)
1169                 self.results.error += 1
1171     def show_brief(self):  argument
1174                     (len(self.suites), len(self.instances),
1175                     self.results.skipped_configs,
1176                     self.results.skipped_filter,
1177                     self.results.skipped_configs - self.results.skipped_filter))
1179 …def add_tasks_to_queue(self, pipeline, build_only=False, test_only=False, retry_build_errors=False…  argument
1180         for instance in self.instances.values():
1197 …instance.filter_stages = self.get_cmake_filter_stages(instance.testsuite.filter, expr_parser.reser…
1206     def pipeline_mgr(self, pipeline, done_queue, lock, results):  argument
1208             with self.jobserver.get_job():
1216                         pb = ProjectBuilder(instance, self.env, self.jobserver)
1217                         pb.duts = self.duts
1229                     pb = ProjectBuilder(instance, self.env, self.jobserver)
1230                     pb.duts = self.duts
1234     def execute(self, pipeline, done):  argument
1237         self.add_tasks_to_queue(pipeline, self.options.build_only, self.options.test_only,
1238                                 retry_build_errors=self.options.retry_build_errors)
1243         for job in range(self.jobs):
1245             p = Process(target=self.pipeline_mgr, args=(pipeline, done, lock, self.results, ))