60     def __init__(self, total=0):  argument
72         self._done = Value('i', 0)
75         self._iteration = Value('i', 0)
79         self._passed = Value('i', 0)
83         self._skipped_configs = Value('i', 0)
87         self._skipped_runtime = Value('i', 0)
91         self._skipped_filter = Value('i', 0)
94         self._skipped_cases = Value('i', 0)
97         self._error = Value('i', 0)
98         self._failed = Value('i', 0)
101         self._total = Value('i', total)
104         self._cases = Value('i', 0)
105         self.lock = Lock()
107     def summary(self):  argument
109         print(f"Total test suites: {self.total}") # actually test instances
110         print(f"Total test cases: {self.cases}")
111         print(f"Executed test cases: {self.cases - self.skipped_cases}")
112         print(f"Skipped test cases: {self.skipped_cases}")
113         print(f"Completed test suites: {self.done}")
114         print(f"Passing test suites: {self.passed}")
115         print(f"Failing test suites: {self.failed}")
116         print(f"Skipped test suites: {self.skipped_configs}")
117         print(f"Skipped test suites (runtime): {self.skipped_runtime}")
118         print(f"Skipped test suites (filter): {self.skipped_filter}")
119         print(f"Errors: {self.error}")
123     def cases(self):  argument
124         with self._cases.get_lock():
125             return self._cases.value
128     def cases(self, value):  argument
129         with self._cases.get_lock():
130             self._cases.value = value
133     def skipped_cases(self):  argument
134         with self._skipped_cases.get_lock():
135             return self._skipped_cases.value
138     def skipped_cases(self, value):  argument
139         with self._skipped_cases.get_lock():
140             self._skipped_cases.value = value
143     def error(self):  argument
144         with self._error.get_lock():
145             return self._error.value
148     def error(self, value):  argument
149         with self._error.get_lock():
150             self._error.value = value
153     def iteration(self):  argument
154         with self._iteration.get_lock():
155             return self._iteration.value
158     def iteration(self, value):  argument
159         with self._iteration.get_lock():
160             self._iteration.value = value
163     def done(self):  argument
164         with self._done.get_lock():
165             return self._done.value
168     def done(self, value):  argument
169         with self._done.get_lock():
170             self._done.value = value
173     def passed(self):  argument
174         with self._passed.get_lock():
175             return self._passed.value
178     def passed(self, value):  argument
179         with self._passed.get_lock():
180             self._passed.value = value
183     def skipped_configs(self):  argument
184         with self._skipped_configs.get_lock():
185             return self._skipped_configs.value
188     def skipped_configs(self, value):  argument
189         with self._skipped_configs.get_lock():
190             self._skipped_configs.value = value
193     def skipped_filter(self):  argument
194         with self._skipped_filter.get_lock():
195             return self._skipped_filter.value
198     def skipped_filter(self, value):  argument
199         with self._skipped_filter.get_lock():
200             self._skipped_filter.value = value
203     def skipped_runtime(self):  argument
204         with self._skipped_runtime.get_lock():
205             return self._skipped_runtime.value
208     def skipped_runtime(self, value):  argument
209         with self._skipped_runtime.get_lock():
210             self._skipped_runtime.value = value
213     def failed(self):  argument
214         with self._failed.get_lock():
215             return self._failed.value
218     def failed(self, value):  argument
219         with self._failed.get_lock():
220             self._failed.value = value
223     def total(self):  argument
224         with self._total.get_lock():
225             return self._total.value
231     def __init__(self, testsuite: TestSuite, platform: Platform, source_dir, build_dir, jobserver):  argument
233         self.cwd = None
234         self.capture_output = True
236         self.defconfig = {}
237         self.cmake_cache = {}
239         self.instance = None
240         self.testsuite = testsuite
241         self.platform = platform
242         self.source_dir = source_dir
243         self.build_dir = build_dir
244         self.log = "build.log"
246         self.default_encoding = sys.getdefaultencoding()
247         self.jobserver = jobserver
249     def parse_generated(self, filter_stages=[]):  argument
250         self.defconfig = {}
253     def run_build(self, args=[]):  argument
255         logger.debug("Building %s for %s" % (self.source_dir, self.platform.name))
263         if self.capture_output:
268         if self.cwd:
269             kwargs['cwd'] = self.cwd
273             p = self.jobserver.popen(cmd, **kwargs)
282         self.instance.build_time += duration
284 …    msg = f"Finished building {self.source_dir} for {self.platform.name} in {duration:.2f} seconds"
287             self.instance.status = "passed"
288             if not self.instance.run:
289                 self.instance.add_missing_case_status("skipped", "Test was built only")
293                 log_msg = out.decode(self.default_encoding)
294 …     with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
302                 log_msg = out.decode(self.default_encoding)
303 …     with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
309                 if overflow_found and not self.options.overflow_as_errors:
311                     self.instance.status = "skipped"
312                     self.instance.reason = "{} overflow".format(overflow_found[0])
313                     change_skip_to_error_if_integration(self.options, self.instance)
314                 elif imgtool_overflow_found and not self.options.overflow_as_errors:
315                     self.instance.status = "skipped"
316                     self.instance.reason = "imgtool overflow"
317                     change_skip_to_error_if_integration(self.options, self.instance)
319                     self.instance.status = "error"
320                     self.instance.reason = "Build failure"
328     def run_cmake(self, args="", filter_stages=[]):  argument
330         if not self.options.disable_warnings_as_errors:
338         if self.instance.sysbuild:
341         logger.debug("Running cmake on %s for %s" % (self.source_dir, self.platform.name))
343             f'-B{self.build_dir}',
344             f'-DTC_RUNID={self.instance.run_id}',
347             f'-G{self.env.generator}'
360         if self.instance.sysbuild and not filter_stages:
361             logger.debug("Building %s using sysbuild" % (self.source_dir))
364                 f'-DAPP_DIR={self.source_dir}'
368                 f'-S{self.source_dir}'
374         cmake_opts = ['-DBOARD={}'.format(self.platform.name)]
377         if self.instance.testsuite.required_snippets:
378 …          cmake_opts = ['-DSNIPPET={}'.format(';'.join(self.instance.testsuite.required_snippets))]
391         if self.capture_output:
396         if self.cwd:
397             kwargs['cwd'] = self.cwd
401             p = self.jobserver.popen(cmd, **kwargs)
407         self.instance.build_time += duration
410             filter_results = self.parse_generated(filter_stages)
411 …msg = f"Finished running cmake {self.source_dir} for {self.platform.name} in {duration:.2f} second…
418             self.instance.status = "error"
419             self.instance.reason = "Cmake build failure"
421             for tc in self.instance.testcases:
422                 tc.status = self.instance.status
424             logger.error("Cmake build failure: %s for %s" % (self.source_dir, self.platform.name))
428             os.makedirs(self.build_dir, exist_ok=True)
429 …     with open(os.path.join(self.build_dir, self.log), "a", encoding=self.default_encoding) as log:
430                 log_msg = out.decode(self.default_encoding)
438     def __init__(self, testsuite: TestSuite, platform: Platform, source_dir, build_dir, jobserver):  argument
441         self.log = "config-twister.log"
443     def parse_generated(self, filter_stages=[]):  argument
445         if self.platform.name == "unit_testing":
448         if self.instance.sysbuild and not filter_stages:
450             domain_path = os.path.join(self.build_dir, "domains.yaml")
453             self.instance.domains = domains
459             cmake_cache_path = os.path.join(self.build_dir, "CMakeCache.txt")
463                 defconfig_path = os.path.join(self.build_dir, "zephyr", ".config")
465             edt_pickle = os.path.join(self.build_dir, "zephyr", "edt.pickle")
472                     m = self.config_re.match(line)
479             self.defconfig = defconfig
490         self.cmake_cache = cmake_conf
493             "ARCH": self.platform.arch,
494             "PLATFORM": self.platform.name
498             filter_data.update(self.defconfig)
499         filter_data.update(self.cmake_cache)
501         if self.instance.sysbuild and self.env.options.device_testing:
505             if self.env.options.west_flash is None:
508                 return {os.path.join(self.platform.name, self.testsuite.name): True}
510         if self.testsuite and self.testsuite.filter:
517                 ret = expr_parser.parse(self.testsuite.filter, filter_data, edt)
521                     "Failed processing %s\n" % self.testsuite.yamlfile)
525                 return {os.path.join(self.platform.name, self.testsuite.name): True}
527                 return {os.path.join(self.platform.name, self.testsuite.name): False}
529             self.platform.filter_data = filter_data
535     def __init__(self, instance: TestInstance, env: TwisterEnv, jobserver, **kwargs):  argument
538         self.log = "build.log"
539         self.instance = instance
540         self.filtered_tests = 0
541         self.options = env.options
542         self.env = env
543         self.duts = None
545     def log_info(self, filename, inline_logs, log_testcases=False):  argument
568                 for tc in self.instance.testcases:
580     def log_info_file(self, inline_logs):  argument
581         build_dir = self.instance.build_dir
589         if os.path.exists(v_log) and "Valgrind" in self.instance.reason:
590             self.log_info("{}".format(v_log), inline_logs)
592             self.log_info("{}".format(pytest_log), inline_logs, log_testcases=True)
594             self.log_info("{}".format(h_log), inline_logs)
596             self.log_info("{}".format(he_log), inline_logs)
598             self.log_info("{}".format(d_log), inline_logs)
600             self.log_info("{}".format(b_log), inline_logs)
603     def process(self, pipeline, done, message, lock, results):  argument
606         self.instance.setup_handler(self.env)
609             ret = self.cmake(filter_stages=self.instance.filter_stages)
610             if self.instance.status in ["failed", "error"]:
611                 pipeline.put({"op": "report", "test": self.instance})
614                 if self.instance.name in ret['filter'] and ret['filter'][self.instance.name]:
615                     logger.debug("filtering %s" % self.instance.name)
616                     self.instance.status = "filtered"
617                     self.instance.reason = "runtime filter"
619                     self.instance.add_missing_case_status("skipped")
620                     pipeline.put({"op": "report", "test": self.instance})
622                     pipeline.put({"op": "cmake", "test": self.instance})
626             ret = self.cmake()
627             if self.instance.status in ["failed", "error"]:
628                 pipeline.put({"op": "report", "test": self.instance})
629             elif self.options.cmake_only:
630                 if self.instance.status is None:
631                     self.instance.status = "passed"
632                 pipeline.put({"op": "report", "test": self.instance})
635                 if self.instance.name in ret['filter'] and ret['filter'][self.instance.name]:
636                     logger.debug("filtering %s" % self.instance.name)
637                     self.instance.status = "filtered"
638                     self.instance.reason = "runtime filter"
640                     self.instance.add_missing_case_status("skipped")
641                     pipeline.put({"op": "report", "test": self.instance})
643                     pipeline.put({"op": "build", "test": self.instance})
646             logger.debug("build test: %s" % self.instance.name)
647             ret = self.build()
649                 self.instance.status = "error"
650                 self.instance.reason = "Build Failure"
651                 pipeline.put({"op": "report", "test": self.instance})
655                 if  self.instance.status == "skipped":
657                     self.instance.add_missing_case_status("skipped", self.instance.reason)
660                     self.instance.add_missing_case_status("blocked", self.instance.reason)
661                     pipeline.put({"op": "report", "test": self.instance})
663                     if self.instance.testsuite.harness in ['ztest', 'test']:
664 …                      logger.debug(f"Determine test cases for test instance: {self.instance.name}")
666                             self.determine_testcases(results)
667                             pipeline.put({"op": "gather_metrics", "test": self.instance})
670                             self.instance.status = "error"
671                             self.instance.reason = str(e)
672                             pipeline.put({"op": "report", "test": self.instance})
674                         pipeline.put({"op": "gather_metrics", "test": self.instance})
677             ret = self.gather_metrics(self.instance)
679                 self.instance.status = "error"
680                 self.instance.reason = "Build Failure at gather_metrics."
681                 pipeline.put({"op": "report", "test": self.instance})
682             elif self.instance.run and self.instance.handler.ready:
683                 pipeline.put({"op": "run", "test": self.instance})
685                 pipeline.put({"op": "report", "test": self.instance})
689             logger.debug("run test: %s" % self.instance.name)
690             self.run()
691             logger.debug(f"run status: {self.instance.name} {self.instance.status}")
694                 self.instance.handler.thread = None
695                 self.instance.handler.duts = None
698                     "test": self.instance,
699                     "status": self.instance.status,
700                     "reason": self.instance.reason
710                 done.put(self.instance)
711                 self.report_out(results)
713             if not self.options.coverage:
714                 if self.options.prep_artifacts_for_testing:
715                     pipeline.put({"op": "cleanup", "mode": "device", "test": self.instance})
716 …         elif self.options.runtime_artifact_cleanup == "pass" and self.instance.status == "passed":
717                     pipeline.put({"op": "cleanup", "mode": "passed", "test": self.instance})
718                 elif self.options.runtime_artifact_cleanup == "all":
719                     pipeline.put({"op": "cleanup", "mode": "all", "test": self.instance})
724                 self.cleanup_device_testing_artifacts()
725 …        elif mode == "passed" or (mode == "all" and self.instance.reason != "Cmake build failure"):
726                 self.cleanup_artifacts()
728     def determine_testcases(self, results):  argument
729         yaml_testsuite_name = self.instance.testsuite.id
732         elf_file = self.instance.get_elf_file()
735 …logger.debug(f"Test instance {self.instance.name} already has {len(self.instance.testcases)} cases…
756             self.instance.testcases.clear()
757             self.instance.testsuite.testcases.clear()
764                 self.instance.add_testcase(name=testcase_id)
765                 self.instance.testsuite.add_testcase(name=testcase_id)
768     def cleanup_artifacts(self, additional_keep: List[str] = []):  argument
769         logger.debug("Cleaning up {}".format(self.instance.build_dir))
788         if self.options.runtime_artifact_cleanup == 'all':
791         allow = [os.path.join(self.instance.build_dir, file) for file in allow]
793         for dirpath, dirnames, filenames in os.walk(self.instance.build_dir, topdown=False):
806     def cleanup_device_testing_artifacts(self):  argument
807         logger.debug("Cleaning up for Device Testing {}".format(self.instance.build_dir))
809         files_to_keep = self._get_binaries()
812         if self.instance.sysbuild:
814             for domain in self.instance.domains.get_domains():
815                 files_to_keep += self._get_artifact_allow_list_for_domain(domain.name)
817         self.cleanup_artifacts(files_to_keep)
819         self._sanitize_files()
821     def _get_artifact_allow_list_for_domain(self, domain: str) -> List[str]:  argument
835     def _get_binaries(self) -> List[str]:  argument
838         self.instance.build_dir), basing on information from platform.binaries
844         platform = self.instance.platform
850         binaries += self._get_binaries_from_runners()
852         if self.instance.sysbuild:
853             for domain in self.instance.domains.get_domains():
854                 binaries += self._get_binaries_from_runners(domain.name)
866     def _get_binaries_from_runners(self, domain='') -> List[str]:  argument
869         self.instance.build_dir) from runners.yaml file. May be used for
873         runners_file_path: str = os.path.join(self.instance.build_dir,
899     def _sanitize_files(self):  argument
904         self._sanitize_runners_file()
905         self._sanitize_zephyr_base_from_files()
907     def _sanitize_runners_file(self):  argument
910         directory for those files is f"{self.instance.build_dir}/zephyr"
912         runners_dir_path: str = os.path.join(self.instance.build_dir, 'zephyr')
938     def _sanitize_zephyr_base_from_files(self):  argument
947             file_path = os.path.join(self.instance.build_dir, file_path)
961     def report_out(self, results):  argument
965         instance = self.instance
976             if self.options.verbose:
987             if not self.options.verbose:
988                 self.log_info_file(self.options.inline_logs)
1005         if self.options.verbose:
1006             if self.options.cmake_only:
1022                      and hasattr(self.instance.handler, 'seed')
1023                      and self.instance.handler.seed is not None ):
1024                     more_info += "/seed: " + str(self.options.seed)
1030                 self.log_info_file(self.options.inline_logs)
1092     def cmake(self, filter_stages=[]):  argument
1093         args = self.cmake_assemble_args(
1094             self.testsuite.extra_args.copy(), # extra_args from YAML
1095             self.instance.handler,
1096             self.testsuite.extra_conf_files,
1097             self.testsuite.extra_overlay_confs,
1098             self.testsuite.extra_dtc_overlay_files,
1099             self.options.extra_args, # CMake extra args
1100             self.instance.build_dir,
1102         return self.run_cmake(args,filter_stages)
1104     def build(self):  argument
1105         harness = HarnessImporter.get_harness(self.instance.testsuite.harness.capitalize())
1106         build_result = self.run_build(['--build', self.build_dir])
1109                 harness.instance = self.instance
1112             self.instance.status = "error"
1113             self.instance.reason = str(error)
1114             logger.error(self.instance.reason)
1118     def run(self):  argument
1120         instance = self.instance
1127                 instance.handler.duts = self.duts
1129             if(self.options.seed is not None and instance.platform.name.startswith("native_")):
1130                 self.parse_generated()
1131                 if('CONFIG_FAKE_ENTROPY_NATIVE_POSIX' in self.defconfig and
1132                     self.defconfig['CONFIG_FAKE_ENTROPY_NATIVE_POSIX'] == 'y'):
1133                     instance.handler.seed = self.options.seed
1135             if self.options.extra_test_args and instance.platform.arch == "posix":
1136                 instance.handler.extra_test_args = self.options.extra_test_args
1154     def gather_metrics(self, instance: TestInstance):  argument
1156         if self.options.create_rom_ram_report:
1157             build_result = self.run_build(['--build', self.build_dir, "--target", "footprint"])
1158         if self.options.enable_size_report and not self.options.cmake_only:
1159             self.calc_size(instance=instance, from_buildlog=self.options.footprint_from_buildlog)
1189     def __init__(self, instances, suites, env=None) -> None:  argument
1190         self.pipeline = None
1191         self.options = env.options
1192         self.env = env
1193         self.instances = instances
1194         self.suites = suites
1195         self.duts = None
1196         self.jobs = 1
1197         self.results = None
1198         self.jobserver = None
1200     def run(self):  argument
1202         retries = self.options.retry_failed + 1
1208         self.results = ExecutionCounter(total=len(self.instances))
1209         self.iteration = 0
1214         if self.options.jobs:
1215             self.jobs = self.options.jobs
1216         elif self.options.build_only:
1217             self.jobs = multiprocessing.cpu_count() * 2
1219             self.jobs = multiprocessing.cpu_count()
1223                 self.jobserver = GNUMakeJobClient.from_environ(jobs=self.options.jobs)
1224                 if not self.jobserver:
1225                     self.jobserver = GNUMakeJobServer(self.jobs)
1226                 elif self.jobserver.jobs:
1227                     self.jobs = self.jobserver.jobs
1230                 self.jobserver = JobClient()
1232             logger.info("JOBS: %d", self.jobs)
1234         self.update_counting_before_pipeline()
1237             self.results.iteration += 1
1239             if self.results.iteration > 1:
1240                 logger.info("%d Iteration:" % (self.results.iteration))
1241                 time.sleep(self.options.retry_interval)  # waiting for the system to settle down
1242                 self.results.done = self.results.total - self.results.failed - self.results.error
1243                 self.results.failed = 0
1244                 if self.options.retry_build_errors:
1245                     self.results.error = 0
1247                 self.results.done = self.results.skipped_filter
1249             self.execute(pipeline, done_queue)
1257                     inst.metrics.update(self.instances[inst.name].metrics)
1260                     self.instances[inst.name] = inst
1265             if self.results.error and self.options.retry_build_errors:
1269             if retries == 0 or ( self.results.failed == 0 and not retry_errors):
1272         self.show_brief()
1274     def update_counting_before_pipeline(self):  argument
1280         for instance in self.instances.values():
1282                 self.results.skipped_filter += 1
1283                 self.results.skipped_configs += 1
1284                 self.results.skipped_cases += len(instance.testsuite.testcases)
1285                 self.results.cases += len(instance.testsuite.testcases)
1287                 self.results.error += 1
1289     def show_brief(self):  argument
1292                     (len(self.suites), len(self.instances),
1293                     self.results.skipped_configs,
1294                     self.results.skipped_filter,
1295                     self.results.skipped_configs - self.results.skipped_filter))
1297 …def add_tasks_to_queue(self, pipeline, build_only=False, test_only=False, retry_build_errors=False…  argument
1298         for instance in self.instances.values():
1315 …instance.filter_stages = self.get_cmake_filter_stages(instance.testsuite.filter, expr_parser.reser…
1323                     if os.path.exists(cache_file) and self.env.options.aggressive_no_clean:
1329     def pipeline_mgr(self, pipeline, done_queue, lock, results):  argument
1332                 with self.jobserver.get_job():
1340                             pb = ProjectBuilder(instance, self.env, self.jobserver)
1341                             pb.duts = self.duts
1353                         pb = ProjectBuilder(instance, self.env, self.jobserver)
1354                         pb.duts = self.duts
1361     def execute(self, pipeline, done):  argument
1364         self.add_tasks_to_queue(pipeline, self.options.build_only, self.options.test_only,
1365                                 retry_build_errors=self.options.retry_build_errors)
1370         for _ in range(self.jobs):
1371             p = Process(target=self.pipeline_mgr, args=(pipeline, done, lock, self.results, ))
1374         logger.debug(f"Launched {self.jobs} jobs")