def validate_cache()

in src/buildstream_plugins/sources/git.py [0:0]

• 78 lines of code
• 16 McCabe index (conditional complexity)

    def validate_cache(self):
        discovered_submodules = {}
        unlisted_submodules = []
        invalid_submodules = []

        for submodule in self._recurse_submodules(configure=False):
            discovered_submodules[submodule.path] = submodule.url
            if self._ignoring_submodule(submodule.path):
                continue

            if submodule.path not in self.submodule_overrides:
                unlisted_submodules.append((submodule.path, submodule.url))

        # Warn about submodules which are explicitly configured but do not exist
        for path, url in self.submodule_overrides.items():
            if path not in discovered_submodules:
                invalid_submodules.append((path, url))

        if invalid_submodules:
            detail = []
            for path, url in invalid_submodules:
                detail.append("  Submodule URL '{}' at path '{}'".format(url, path))

            self.warn(
                "{}: Invalid submodules specified".format(self),
                warning_token=WARN_INVALID_SUBMODULE,
                detail="The following submodules are specified in the source "
                "description but do not exist according to the repository\n\n" + "\n".join(detail),
            )

        # Warn about submodules which exist but have not been explicitly configured
        if unlisted_submodules:
            detail = []
            for path, url in unlisted_submodules:
                detail.append("  Submodule URL '{}' at path '{}'".format(url, path))

            self.warn(
                "{}: Unlisted submodules exist".format(self),
                warning_token=WARN_UNLISTED_SUBMODULE,
                detail="The following submodules exist but are not specified "
                + "in the source description\n\n"
                + "\n".join(detail),
            )

        # Assert that the ref exists in the track tag/branch, if track has been specified.
        # Also don't do this check if an exact tag ref is given, as we probably didn't fetch
        # any branch refs
        ref_in_track = False
        if not re.match(EXACT_TAG_PATTERN, self.mirror.ref) and self.tracking:
            _, branch = self.check_output(
                [
                    self.host_git,
                    "branch",
                    "--list",
                    self.tracking,
                    "--contains",
                    self.mirror.ref,
                ],
                cwd=self.mirror.mirror,
            )
            if branch:
                ref_in_track = True
            else:
                _, tag = self.check_output(
                    [
                        self.host_git,
                        "tag",
                        "--list",
                        self.tracking,
                        "--contains",
                        self.mirror.ref,
                    ],
                    cwd=self.mirror.mirror,
                )
                if tag:
                    ref_in_track = True

            if not ref_in_track:
                detail = (
                    "The ref provided for the element does not exist locally "
                    + "in the provided track branch / tag '{}'.\n".format(self.tracking)
                    + "You may wish to track the element to update the ref from '{}' ".format(self.tracking)
                    + "with `bst source track`,\n"
                    + "or examine the upstream at '{}' for the specific ref.".format(self.mirror.url)
                )

                self.warn(
                    "{}: expected ref '{}' was not found in given track '{}' for staged repository: '{}'\n".format(
                        self, self.mirror.ref, self.tracking, self.mirror.url
                    ),
                    detail=detail,
                    warning_token=CoreWarnings.REF_NOT_IN_TRACK,
                )