def _load_errorsummary()

in mozci/data/sources/artifact/__init__.py [0:0]
66 lines of code
29 McCabe index (conditional complexity)

    def _load_errorsummary(self, task_id) -> None:
        """Load the task's errorsummary.log.

        We gather all data we can and store it in the TASK_* caches so we don't
        have to load it again for a different contract.
        """
        try:
            artifacts = [a["name"] for a in list_artifacts(task_id)]
            paths = [a for a in artifacts if a.endswith("errorsummary.log")]
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 404:
                return
            raise
        except IndexError:
            return

        groups = set()
        group_results = {}
        test_results: Dict[GroupName, List[Tuple[TestName, FailureType]]] = {}

        lines = (
            line
            for path in paths
            for line in get_artifact(task_id, path).iter_lines(decode_unicode=True)
            if line
        )

        for line_text in lines:
            try:
                line = json.loads(line_text)
            except json.decoder.JSONDecodeError:
                logger.log(
                    "WARNING",
                    f"Invalid JSON line in errorsummary.log from task {task_id}",
                )
                continue

            if line["action"] == "test_groups":
                groups |= set(line["groups"]) - {"default"}

            elif line["action"] == "group_result":
                group = line["group"]
                if group not in group_results or line["status"] != "OK":
                    group_results[group] = (line["status"], line["duration"])

            elif line["action"] == "log":
                if task_id not in self.TASK_ERRORS:
                    self.TASK_ERRORS[task_id] = []
                self.TASK_ERRORS[task_id].append(line["message"])

            if line.get("test") and line.get("group"):
                if not test_results.get(line["group"]):
                    test_results[line["group"]] = []

                if (
                    line.get("status") != line.get("expected")
                    and line.get("status") == "TIMEOUT"
                ):
                    failure_type = FailureType.TIMEOUT
                elif (
                    line.get("signature") is not None and line.get("action") == "crash"
                ):
                    failure_type = FailureType.CRASH
                else:
                    failure_type = FailureType.GENERIC

                test_results[line["group"]].append((line["test"], failure_type))

        missing_groups = groups - set(group_results)
        if len(missing_groups) > 0:
            log_level = "DEBUG" if self.is_try else "WARNING"
            logger.log(
                log_level,
                f"Some groups in {task_id} are missing results: {missing_groups}",
            )

        self.TASK_GROUPS[task_id] = {
            group: (result == "OK", duration)
            for group, (result, duration) in group_results.items()
            if result != "SKIP"
        }

        self.TASK_FAILURE_TYPES[task_id] = test_results