def _set_completed_results()

in lisa/runners/legacy_runner.py [0:0]

• 35 lines of code
• 13 McCabe index (conditional complexity)

    def _set_completed_results(self, completed_cases: List[Dict[str, str]]) -> None:

        new_completed_cases = completed_cases[:]
        not_matched_results = [
            x
            for x in self._results
            if x.status not in [TestStatus.QUEUED, TestStatus.RUNNING]
        ]
        # remove existing completed case
        for completed_case in completed_cases:
            for result in not_matched_results:
                if self._is_matched_infomation(result, completed_case):
                    new_completed_cases.remove(completed_case)
                    not_matched_results.remove(result)
                    break
        if not_matched_results:
            self.log.error(
                f"not matched should be empty, but {not_matched_results}, "
                f"parsed completed cases: {[completed_cases]}"
            )

        # set new completed case information
        running_results = [x for x in self._results if x.status == TestStatus.RUNNING]
        not_matched_cases = new_completed_cases[:]
        for completed_case in new_completed_cases:
            for result in running_results:
                if self._is_matched_infomation(result, completed_case):
                    # complete a case
                    self._set_result(
                        result,
                        information=completed_case,
                    )
                    # every running result just match one
                    running_results.remove(result)
                    not_matched_cases.remove(completed_case)
                    break
        if not_matched_cases:
            self.log.error(
                f"found unmatched completed results: {not_matched_cases}, "
                f"running results: {running_results}"
            )